cursor.ts

declare const EdgeRuntime: string | undefined;
/**
 * CursorExecutor — talks to Cursor's agent.v1.AgentService/Run endpoint.
 *
 * cursor-agent (CLI) and the cursor IDE both use this RPC for every model id
 * (auto, composer-*, claude-*, gpt-*, gemini-*). The legacy
 * aiserver.v1.ChatService/StreamUnifiedChatWithTools rejects "auto" and
 * "composer-*" with errors, so we migrated this executor over.
 *
 * Wire format & schema details live in ../utils/cursorAgentProtobuf.ts.
 */

import { BaseExecutor, mergeUpstreamExtraHeaders } from "./base.ts";
import { PROVIDERS, HTTP_STATUS } from "../config/constants.ts";
import { getAccessToken } from "../services/tokenRefresh.ts";
import {
  buildAgentRequestBody,
  decodeAgentServerMessage,
  decodeExecServerEvent,
  decodeKvServerEvent,
  encodeRequestContextResponse,
  encodeKvGetBlobResult,
  encodeKvSetBlobResult,
  encodeExecReadRejected,
  encodeExecWriteRejected,
  encodeExecDeleteRejected,
  encodeExecLsRejected,
  encodeExecShellRejected,
  encodeExecBackgroundShellSpawnRejected,
  encodeExecGrepError,
  encodeExecFetchError,
  encodeExecWriteShellStdinError,
  encodeExecDiagnosticsResult,
  flattenMessages,
  openAIToolsToMcpDefs,
  type ChatMessage,
  type EncodedImage,
  type ExecServerEvent,
  type McpToolDefinition,
  type OpenAITool,
} from "../utils/cursorAgentProtobuf.ts";
import { resolveCursorImages, extractImageUrls, CursorImageError } from "../utils/cursorImages.ts";
import {
  estimateInputTokens,
  estimateOutputTokens,
  addBufferToUsage,
} from "../utils/usageTracking.ts";
import {
  formatCursorAgentClientVersion,
  getCursorAgentCliVersion,
} from "../utils/cursorAgentCliVersion.ts";
import { sanitizeErrorMessage } from "../utils/error.ts";
import { generateToolCallId } from "../translator/helpers/toolCallHelper.ts";
import {
  parseComposerToolCalls,
  createStreamingState,
  feedStreamingChunk,
  type StreamingState as ComposerStreamingState,
} from "../utils/composerToolCalls.ts";
import { cursorSessionManager, type CursorSession } from "../services/cursorSessionManager.ts";
import {
  CursorApiKeyExchangeError,
  invalidateCursorSessionToken,
  isCursorApiKey,
  resolveCursorBearerToken,
  stripCursorOAuthTokenPrefix,
} from "../services/cursorApiKeyAuth.ts";
import crypto from "crypto";
import * as fs from "node:fs";
import * as zlib from "node:zlib";
import { promisify } from "node:util";
import { toolChoiceDirectiveLine, buildCursorOutputConstraints } from "./cursor/prompt.ts";
import {
  bridgeCursorBuiltinTool,
  bridgeCursorNativeTodoWrite,
  extractLatestTodoHistory,
  selectCursorBridgeTools,
  type CursorClientPlatform,
  type CursorTodoHistoryItem,
} from "./cursor/builtinToolBridge.ts";
import {
  isComposerModel,
  visibleComposerContentFromThinking,
  composerReasoningRemainder,
} from "./cursor/composer.ts";
import { CursorServerConfigError, resolveCursorAgentUrl } from "./cursor/agentEndpoint.ts";
import {
  classifyCursorError,
  isCursorBenignCancelError,
  resolveCursorEmptyTurnError,
  type ClassifiedCursorError,
} from "./cursor/cursorErrors.ts";
import { getActiveSyncedCatalog } from "../../src/lib/db/models/activeSyncedCatalog.ts";
// Composer helpers re-exported for external importers (tests).
export {
  isComposerModel,
  visibleComposerContentFromThinking,
  composerReasoningRemainder,
} from "./cursor/composer.ts";

// Reject reason text aligned with kaitranntt/CLIProxyAPIPlus — proven to
// keep cursor's model from retrying the same built-in tool indefinitely.
// The model adapts and either answers from context or uses declared MCP tools.
const BUILTIN_TOOL_REJECT_REASON =
  "Tool not available in this environment. Use the MCP tools provided instead.";
const gunzipAsync = promisify(zlib.gunzip);

// Tool-commit directive — adapted from composer-api's TOOL_SYSTEM_DIRECTIVE.
// composer-2.5 otherwise narrates intent ("Checking the weather...") and ends
// the turn ~20% of the time instead of actually invoking a declared tool. This
// directive, prepended to the user text only when the request declares tools,
// tells the model to commit to the tool call rather than describe it as prose.
const TOOL_COMMIT_DIRECTIVE = [
  "You are serving an OpenAI-compatible API request and the client has provided executable tools.",
  "When a tool is needed to answer (real-time data, web/search lookups, file or project operations), you MUST issue the actual tool call. Do NOT describe what you are about to do as prose and then stop — call the tool.",
  "Answer directly only when no tool is needed.",
  "Do not emit duplicate tool calls: call each operation once, then continue after the tool result is returned.",
  "Never claim that tools are unavailable.",
].join("\n");

// NOTE: composer-api primes the model into "agent mode" with a fabricated
// prior switch_mode exchange (AGENT_MODE_PRIMER). On OmniRoute's native-tool
// agent endpoint that primer is counterproductive — it references a
// non-existent switch_mode tool and measurably LOWERED the tool-call rate in
// live A/B (56% vs 69%), so it is intentionally not ported.

/**
 * Build the ExecClientMessage frame that responds to a built-in tool request.
 * Returns null for the request_context handshake (caller handles separately
 * to inject MCP tools in Phase 3) and for exec_mcp (model is invoking a
 * declared MCP tool — Phase 5 surfaces this as an OpenAI tool_calls delta).
 */
function buildExecRejection(event: ExecServerEvent): Buffer | null {
  switch (event.kind) {
    case "exec_request_context":
    case "exec_mcp":
      return null;
    case "exec_read":
      return encodeExecReadRejected(
        event.execMsgId,
        event.execId,
        event.path,
        BUILTIN_TOOL_REJECT_REASON
      );
    case "exec_write":
      return encodeExecWriteRejected(
        event.execMsgId,
        event.execId,
        event.path,
        BUILTIN_TOOL_REJECT_REASON
      );
    case "exec_delete":
      return encodeExecDeleteRejected(
        event.execMsgId,
        event.execId,
        event.path,
        BUILTIN_TOOL_REJECT_REASON
      );
    case "exec_ls":
      return encodeExecLsRejected(
        event.execMsgId,
        event.execId,
        event.path,
        BUILTIN_TOOL_REJECT_REASON
      );
    case "exec_grep":
      return encodeExecGrepError(event.execMsgId, event.execId, BUILTIN_TOOL_REJECT_REASON);
    case "exec_diagnostics":
      // Diagnostics has no rejection variant — return an empty success.
      return encodeExecDiagnosticsResult(event.execMsgId, event.execId);
    case "exec_shell":
    case "exec_shell_stream":
      return encodeExecShellRejected(
        event.execMsgId,
        event.execId,
        event.command,
        event.workingDir,
        BUILTIN_TOOL_REJECT_REASON
      );
    case "exec_bg_shell":
      return encodeExecBackgroundShellSpawnRejected(
        event.execMsgId,
        event.execId,
        event.command,
        event.workingDir,
        BUILTIN_TOOL_REJECT_REASON
      );
    case "exec_fetch":
      return encodeExecFetchError(
        event.execMsgId,
        event.execId,
        event.url,
        BUILTIN_TOOL_REJECT_REASON
      );
    case "exec_write_shell_stdin":
      return encodeExecWriteShellStdinError(
        event.execMsgId,
        event.execId,
        BUILTIN_TOOL_REJECT_REASON
      );
  }
}

// Detect cloud environment (Edge runtime, Cloudflare Workers, etc.)
const isCloudEnv = () => {
  if (typeof caches !== "undefined" && typeof caches === "object") return true;
  if (typeof EdgeRuntime !== "undefined") return true;
  return false;
};

// Lazy import http2 (only in Node.js environment)
let http2: typeof import("http2") | null = null;
if (!isCloudEnv()) {
  try {
    http2 = await import("http2");
  } catch {
    http2 = null;
  }
}

// Phase 10: CURSOR_DEBUG=1 enables verbose streaming debug logs (decoded
// frame summaries, exec router dispatches, session lifecycle events).
// CURSOR_STREAM_DEBUG is kept as a backward-compatible alias.
const CURSOR_DEBUG = process.env.CURSOR_DEBUG === "1" || process.env.CURSOR_STREAM_DEBUG === "1";
const debugLog = (...args: unknown[]) => {
  if (CURSOR_DEBUG) console.log(...args);
};

// Phase 8: max wall-clock time before we give up on the upstream and abort
// the stream. Cursor's longest-observed plain chat takes ~90s; tool-using
// turns can be longer. Five minutes is generous but bounded. A malformed env
// value (NaN / non-positive) falls back to the default rather than breaking
// setTimeout.
const CURSOR_STREAM_TIMEOUT_MS = (() => {
  const parsed = parseInt(process.env.CURSOR_STREAM_TIMEOUT_MS || "300000", 10);
  return Number.isInteger(parsed) && parsed > 0 ? parsed : 300000;
})();

// Upper bound on a single Connect-RPC frame. The 4-byte length prefix can
// declare up to 4 GiB; a corrupt or hostile upstream could send a huge length
// that forces driveH2's rolling buffer to grow unbounded (OOM) while it waits
// for bytes that never arrive. Real cursor frames are well under 1 MiB
// (largest observed: a ~13 KB KV blob), so 16 MiB is a generous ceiling that
// turns the failure into a clean stream error instead of memory exhaustion.
const CURSOR_MAX_FRAME_BYTES = 16 * 1024 * 1024;

type CursorHttpResponse = {
  status: number;
  headers: Record<string, unknown>;
  body: Buffer;
};

function tryParseJsonError(payload: Buffer): { message: string; status: number } | null {
  if (payload.length < 2 || payload[0] !== 0x7b) return null;
  try {
    const text = payload.toString("utf8");
    if (!text.includes('"error"')) return null;
    const parsed = JSON.parse(text);
    const err = parsed?.error || {};
    const rawMessage =
      err?.details?.[0]?.debug?.details?.title ||
      err?.details?.[0]?.debug?.details?.detail ||
      err?.message ||
      (typeof err?.code === "string" ? `${err.code}: ${text}` : text);
    const codeHint =
      typeof err?.code === "string" &&
      !String(rawMessage).toLowerCase().includes(err.code.toLowerCase())
        ? `${err.code}: ${rawMessage}`
        : String(rawMessage);
    const classified = classifyCursorError(codeHint);
    return { message: classified.message, status: classified.status };
  } catch {
    return null;
  }
}

/** True when the turn produced no client-visible assistant payload. */
function isCursorEmptyTurn(ctx: StreamCtx): boolean {
  return (
    ctx.totalText.length === 0 &&
    ctx.thinkingText.length === 0 &&
    ctx.toolCalls.length === 0 &&
    !ctx.composerInlineToolCallsEmitted
  );
}

// ─── Phase 4: streaming dispatch context ───────────────────────────────────
//
// One StreamCtx flows through a single execute() call. It owns the live
// SSE emission state (responseId, created timestamp, model id, role-chunk
// flag) plus aggregate state (totalText, tokenDelta) needed for the final
// usage chunk and JSON-mode aggregation. Phases 5 (tool calls) and 8
// (end-signal hardening) extend it.

export type StreamCtx = {
  responseId: string;
  created: number;
  model: string;
  emit: (chunk: string) => void;
  emittedRoleChunk: boolean;
  totalText: string;
  thinkingText: string;
  tokenDelta: number;
  // End-signal tracking (Phase 8 hardens this further).
  receivedText: boolean;
  kvAfterTextSeen: boolean;
  endReason: "turn_ended" | "kv_after_text" | "tool_calls" | "server_end" | null;
  // Mid-stream JSON error (rare; emitted once with the error code).
  midStreamError: { message: string; status: number } | null;
  // Phase 5: tool-call indexing for parallel calls. Each McpArgs gets a
  // monotonically-increasing index in the OpenAI delta. emittedToolCalls
  // tracks how many were emitted so finalizeSseStream picks the right
  // finish_reason ("tool_calls" vs "stop").
  emittedToolCallIndex: number;
  // Captured tool calls (for JSON-mode aggregation). Each entry maps to
  // one OpenAI tool_calls[] item.
  toolCalls: Array<{
    id: string;
    name: string;
    argumentsJson: string;
  }>;
  // Phase 6: maps OpenAI tool_call_id → cursor exec info, so a follow-up
  // role:"tool" message can be answered on the open h2 stream via
  // encodeExecMcpResult.
  pendingToolCalls: Map<string, { execMsgId: number; execId: string; toolName: string }>;
  // Built-in Cursor tools are bridged to external OpenAI tool calls by first
  // rejecting the native request. Their result therefore cannot resume on the
  // same h2 stream and must use the existing full-history cold-resume path.
  requiresColdResume: boolean;
  // Composer thinking-as-content (decolua/9router#1310): tracks how much of
  // the visible suffix (after the last `</think>`) has already been streamed
  // out as `content` deltas, so we only emit the incremental tail per frame.
  composerVisibleEmittedLength: number;
  // Composer DeepSeek-format inline tool-call parser state (decolua/9router#1335).
  // Null for non-Composer models (no overhead). When set, the streaming parser
  // holds back text inside `<｜tool▁calls▁begin｜>...<｜tool▁calls▁end｜>` markers
  // and emits structured tool_calls SSE chunks once the block closes.
  composerToolParserState: ComposerStreamingState | null;
  // True once we've emitted structured tool_calls from the inline Composer parser
  // (to avoid double-emitting if the block appears in multiple accumulated frames).
  composerInlineToolCallsEmitted: boolean;
};

export function newStreamCtx(model: string, emit: (chunk: string) => void): StreamCtx {
  return {
    responseId: `chatcmpl-cursor-${Date.now()}`,
    created: Math.floor(Date.now() / 1000),
    model,
    emit,
    emittedRoleChunk: false,
    totalText: "",
    thinkingText: "",
    tokenDelta: 0,
    receivedText: false,
    kvAfterTextSeen: false,
    endReason: null,
    midStreamError: null,
    emittedToolCallIndex: 0,
    toolCalls: [],
    pendingToolCalls: new Map(),
    requiresColdResume: false,
    composerVisibleEmittedLength: 0,
    composerToolParserState: isComposerModel(model) ? createStreamingState() : null,
    composerInlineToolCallsEmitted: false,
  };
}

function emitChunk(ctx: StreamCtx, delta: object, finishReason: string | null = null) {
  const payload = {
    id: ctx.responseId,
    object: "chat.completion.chunk",
    created: ctx.created,
    model: ctx.model,
    choices: [{ index: 0, delta, finish_reason: finishReason }],
  };
  ctx.emit(`data: ${JSON.stringify(payload)}\n\n`);
}

/**
 * Emit a terminal OpenAI SSE error matching `buildStreamErrorChunks` shape
 * (`finish_reason: "error"` + `error.message`) so #8649 sawError stands down
 * and Model Test All keeps the classified Cursor message.
 */
export function emitCursorSseError(ctx: StreamCtx, classified: ClassifiedCursorError): void {
  const payload = {
    id: ctx.responseId,
    object: "chat.completion.chunk",
    created: ctx.created,
    model: ctx.model,
    choices: [{ index: 0, delta: {}, finish_reason: "error" }],
    error: {
      message: classified.message,
      type: classified.type,
    },
  };
  ctx.emit(`data: ${JSON.stringify(payload)}\n\n`);
  ctx.emit("data: [DONE]\n\n");
}

export function buildCursorUsage(ctx: StreamCtx, body: { messages?: ChatMessage[] }) {
  const promptTokens = estimateInputTokens(body);
  const completionTokens =
    ctx.tokenDelta > 0
      ? ctx.tokenDelta
      : estimateOutputTokens(ctx.totalText.length + ctx.thinkingText.length);
  const usage: Record<string, unknown> = {
    prompt_tokens: promptTokens,
    completion_tokens: completionTokens,
    total_tokens: promptTokens + completionTokens,
    estimated: true,
  };
  if (ctx.thinkingText.length > 0) {
    usage.completion_tokens_details = {
      reasoning_tokens: estimateOutputTokens(ctx.thinkingText.length),
    };
  }
  return addBufferToUsage(usage);
}

function emitUsage(ctx: StreamCtx, body: { messages?: ChatMessage[] }) {
  // Always emit a usage chunk on the success path — the OpenAI streaming
  // contract is that every completed response carries usage. buildCursorUsage
  // already degrades cleanly to prompt-only counts when the model produced no
  // text/thinking (e.g. an empty turn), so there's no need to skip it. The
  // mid-stream-error path in finalizeSseStream returns before calling this, so
  // errored responses still don't get a spurious usage chunk.
  const usage = buildCursorUsage(ctx, body);
  const payload = {
    id: ctx.responseId,
    object: "chat.completion.chunk",
    created: ctx.created,
    model: ctx.model,
    choices: [],
    usage,
  };
  ctx.emit(`data: ${JSON.stringify(payload)}\n\n`);
}

function emitDone(ctx: StreamCtx) {
  ctx.emit("data: [DONE]\n\n");
}

export function inferCursorClientPlatform(
  messages: ChatMessage[]
): CursorClientPlatform | undefined {
  const systemMessages = messages.filter((message) => message.role === "system");
  if (systemMessages.length === 0) return undefined;
  const text = flattenMessages(systemMessages);
  const platforms = new Set<CursorClientPlatform>();
  const metadataPattern =
    /\b(?:client\s+)?(?:platform|os|operating\s+system)\s*[:=]\s*["']?(win32|windows|linux|darwin|macos|posix)\b/gi;
  for (const match of text.matchAll(metadataPattern)) {
    platforms.add(/^(?:win32|windows)$/i.test(match[1]) ? "windows" : "posix");
  }
  return platforms.size === 1 ? [...platforms][0] : undefined;
}

/** Emit one complete OpenAI-compatible structured tool call. */
function emitStructuredToolCall(
  ctx: StreamCtx,
  toolName: string,
  args: Record<string, unknown>
): string {
  if (!ctx.emittedRoleChunk) {
    emitChunk(ctx, { role: "assistant", content: "" });
    ctx.emittedRoleChunk = true;
  }
  const idx = ctx.emittedToolCallIndex++;
  const openAIToolCallId = generateToolCallId();
  const argumentsJson = JSON.stringify(args);
  emitChunk(ctx, {
    tool_calls: [
      {
        index: idx,
        id: openAIToolCallId,
        type: "function",
        function: { name: toolName, arguments: "" },
      },
    ],
  });
  emitChunk(ctx, {
    tool_calls: [
      {
        index: idx,
        function: { arguments: argumentsJson },
      },
    ],
  });
  ctx.toolCalls.push({ id: openAIToolCallId, name: toolName, argumentsJson });
  return openAIToolCallId;
}

/**
 * Process one decoded Connect-RPC frame payload: dispatch ExecServerMessage
 * events (rejection / context ack / mcp_args), decode AgentServerMessage
 * interaction updates, and emit OpenAI SSE deltas for any text content.
 *
 * Returns true if an end-of-response signal was observed.
 *
 * The h2 `req` (used to write rejection acks back on the same stream) is
 * passed via opts so this function works for both the streaming h2 path
 * and the buffered fetch fallback (where opts.req is undefined).
 *
 * Mutates `ackedExecIds` so each exec_id is dispatched exactly once even
 * when the same payload is seen multiple times during incremental decoding.
 */
export function processFrame(
  payload: Buffer,
  ctx: StreamCtx,
  ackedExecIds: Set<string>,
  opts: {
    h2Req?: import("http2").ClientHttp2Stream;
    mcpTools?: McpToolDefinition[];
    blobStore?: Map<string, Buffer>;
    clientPlatform?: CursorClientPlatform;
    todoHistory?: CursorTodoHistoryItem[];
  } = {}
): void {
  // 1. JSON error envelope (Connect-RPC style — usually status > 200).
  const jsonError = tryParseJsonError(payload);
  if (jsonError) {
    if (ctx.totalText.length === 0) {
      ctx.midStreamError = jsonError;
      ctx.endReason = "server_end";
    } else {
      // Already streamed content — terminate cleanly.
      ctx.endReason = "server_end";
    }
    return;
  }

  // 2a. KV server message: cursor requesting a blob (system prompt) or
  // saving an assistant turn. We reply on the same stream so the model
  // proceeds. The opaque request_metadata is echoed so cursor can match
  // request to response.
  const kvEvent = decodeKvServerEvent(payload);
  if (kvEvent && opts.h2Req) {
    if (kvEvent.kind === "kv_get_blob") {
      const hex = kvEvent.blobId.toString("hex");
      const blob = opts.blobStore?.get(hex) ?? Buffer.alloc(0);
      try {
        opts.h2Req.write(encodeKvGetBlobResult(kvEvent.kvId, blob, kvEvent.requestMetadata));
      } catch (e) {
        console.debug(`[CURSOR] KV get_blob write failed:`, e);
      }
    } else if (kvEvent.kind === "kv_set_blob") {
      if (opts.blobStore) {
        opts.blobStore.set(kvEvent.blobId.toString("hex"), kvEvent.blobData);
      }
      try {
        opts.h2Req.write(encodeKvSetBlobResult(kvEvent.kvId, kvEvent.requestMetadata));
      } catch (e) {
        console.debug(`[CURSOR] KV set_blob write failed:`, e);
      }
    }
  }

  // 2b. ExecServerMessage dispatch (request_context, built-in rejection, mcp).
  // Dedup by kind+execId+execMsgId — request_context and mcp_args both
  // arrive with empty execId in the current cursor schema, so a single
  // execId-only set would collapse them.
  const event = decodeExecServerEvent(payload);
  const dedupKey = event ? `${event.kind}:${event.execId}:${event.execMsgId}` : "";
  if (event && !ackedExecIds.has(dedupKey)) {
    ackedExecIds.add(dedupKey);
    if (event.kind === "exec_request_context") {
      if (opts.h2Req) {
        try {
          // Cursor receives tools via AgentRunRequest.mcp_tools (request body)
          // — sending them again in the request_context ack causes the
          // server to stall silently. Empty ack only.
          opts.h2Req.write(encodeRequestContextResponse(event.execMsgId, event.execId));
        } catch (e) {
          console.debug(`[CURSOR] request_context ack write failed:`, e);
        }
      }
    } else if (event.kind === "exec_mcp") {
      // Phase 5: surface the model-invoked MCP tool as an OpenAI tool_calls
      // SSE delta. Two chunks are emitted per call: an init chunk with the
      // tool's id+name+empty args, then a chunk with the JSON-stringified
      // args. Parallel tool calls share one finish chunk (Phase 8 closes).
      const openAIToolCallId = emitStructuredToolCall(ctx, event.toolName, event.args ?? {});
      // Phase 6: remember the cursor exec ids so a follow-up role:"tool"
      // message can be replied with encodeExecMcpResult on the open h2 stream.
      ctx.pendingToolCalls.set(openAIToolCallId, {
        execMsgId: event.execMsgId,
        execId: event.execId,
        toolName: event.toolName,
      });
      // Cursor pauses after mcp_args waiting for the client to either send
      // a tool result via ExecMcpResult or close the stream. We mark
      // endReason now so driveH2 returns; the session manager keeps the h2
      // alive for the next OpenAI call (which arrives with role:"tool").
      ctx.endReason = "tool_calls";
    } else {
      // Cursor/Fable frequently chooses its native Shell tool even when the
      // OpenAI client declared external tools. If a schema-compatible shell
      // tool exists, surface the native request as a structured OpenAI call.
      // We still send the typed rejection upstream, then close this h2 stream;
      // the role:"tool" follow-up is resumed cold from the full history.
      const bridge = bridgeCursorBuiltinTool(event, opts.mcpTools ?? [], opts.clientPlatform);
      const rejection = buildExecRejection(event);
      if (rejection && opts.h2Req) {
        try {
          opts.h2Req.write(rejection);
        } catch (e) {
          console.debug(`[CURSOR] exec rejection write failed:`, e);
        }
      }
      if (bridge) {
        emitStructuredToolCall(ctx, bridge.toolName, bridge.arguments);
        ctx.requiresColdResume = true;
        ctx.endReason = "tool_calls";
      }
    }
  }

  // 3. Interaction update deltas → OpenAI SSE chunks.
  let deltas;
  try {
    deltas = decodeAgentServerMessage(payload);
  } catch (err) {
    debugLog("[cursor-agent] decode failed:", (err as Error).message);
    return;
  }
  for (const d of deltas) {
    if (d.kind === "native_todo_write") {
      const dedupKey = `native_todo_write:${d.toolCallId}`;
      if (!ackedExecIds.has(dedupKey)) {
        ackedExecIds.add(dedupKey);
        const bridge = bridgeCursorNativeTodoWrite(d, opts.mcpTools ?? [], opts.todoHistory);
        if (bridge) {
          emitStructuredToolCall(ctx, bridge.toolName, bridge.arguments);
          ctx.requiresColdResume = true;
          ctx.endReason = "tool_calls";
        }
      }
    } else if (d.kind === "text" && d.text) {
      if (!ctx.emittedRoleChunk) {
        emitChunk(ctx, { role: "assistant", content: "" });
        ctx.emittedRoleChunk = true;
      }
      ctx.totalText += d.text;
      ctx.receivedText = true;
      emitChunk(ctx, { content: d.text });
    } else if (d.kind === "thinking" && d.text) {
      if (!ctx.emittedRoleChunk) {
        emitChunk(ctx, { role: "assistant", content: "" });
        ctx.emittedRoleChunk = true;
      }
      ctx.thinkingText += d.text;
      ctx.receivedText = true;
      // Composer (decolua/9router#1310) encodes the visible reply inside the
      // thinking field, after a final `</think>` marker. Emit the post-marker
      // suffix as plain `content` (so OpenAI-compatible clients see the reply)
      // and keep the pre-marker chain-of-thought out of `reasoning_content` —
      // it was never intended for the user.
      if (isComposerModel(ctx.model)) {
        const visible = visibleComposerContentFromThinking(ctx.thinkingText);
        if (visible.length > ctx.composerVisibleEmittedLength) {
          // Feed the full accumulated visible text into the DeepSeek inline
          // tool-call streaming parser (decolua/9router#1335). It tracks how
          // much has already been safely emitted and returns only the new
          // safe delta — i.e. text that precedes any `<｜tool▁calls▁begin｜>`
          // marker (or a partial prefix of one). When the closing marker
          // arrives, it sets ready=true and provides the parsed tool_calls.
          if (ctx.composerToolParserState) {
            const parseOut = feedStreamingChunk(ctx.composerToolParserState, visible);
            // composerVisibleEmittedLength tracks what the parser has "emitted"
            // — stays in sync via state.emitted.
            ctx.composerVisibleEmittedLength = ctx.composerToolParserState.emitted;
            if (parseOut.safeDelta) {
              ctx.totalText += parseOut.safeDelta;
              emitChunk(ctx, { content: parseOut.safeDelta });
            }
            if (
              parseOut.ready &&
              parseOut.toolCalls.length > 0 &&
              !ctx.composerInlineToolCallsEmitted
            ) {
              ctx.composerInlineToolCallsEmitted = true;
              for (const tc of parseOut.toolCalls) {
                const toolCallIndex = ctx.emittedToolCallIndex++;
                ctx.toolCalls.push({
                  id: tc.id,
                  name: tc.function.name,
                  argumentsJson: tc.function.arguments,
                });
                emitChunk(ctx, {
                  tool_calls: [
                    {
                      index: toolCallIndex,
                      id: tc.id,
                      type: "function",
                      function: { name: tc.function.name, arguments: tc.function.arguments },
                    },
                  ],
                });
              }
            }
          } else {
            // Non-composer or state not initialised — fall back to direct emit.
            const deltaContent = visible.slice(ctx.composerVisibleEmittedLength);
            ctx.composerVisibleEmittedLength = visible.length;
            ctx.totalText += deltaContent;
            emitChunk(ctx, { content: deltaContent });
          }
        }
      } else {
        emitChunk(ctx, { reasoning_content: d.text });
      }
    } else if (d.kind === "token_delta") {
      ctx.tokenDelta += d.tokens;
    } else if (d.kind === "turn_ended") {
      if (ctx.endReason !== "tool_calls") ctx.endReason = "turn_ended";
    } else if (d.kind === "tool_call_completed" && ctx.toolCalls.length > 0) {
      // Phase 6: model paused awaiting tool result. driveH2 returns but the
      // h2 stream stays open — the session manager keeps it alive for the
      // next OpenAI call (which will arrive with role:"tool" results).
      ctx.endReason = "tool_calls";
    } else if (
      d.kind === "kv_server_message" &&
      ctx.receivedText &&
      ctx.endReason !== "tool_calls"
    ) {
      // Cursor short-circuits turn_ended for plain chats — kv_server_message
      // after text means the model finished and the server is saving the
      // turn. Phase 8 keeps both signals as defense-in-depth.
      //
      // Safe vs tool calls (composer family only): when the model invokes a
      // tool, the exec_mcp event always arrives at or before this kv
      // checkpoint (verified across many live composer-2.5 trials — a tool call
      // never follows kv_after_text), so endReason is already "tool_calls" by
      // the time we get here. Ending on kv_after_text therefore never truncates
      // a pending tool call on composer.
      //
      // Non-composer models (cursor/grok-4.5-high, auto, ...) emit the KV
      // checkpoint as a blob-store side-channel frame (envelope field 4,
      // kv_get_blob/kv_set_blob) with NO turn-completion semantics, and it can
      // arrive while the model is still streaming a long preamble BEFORE a
      // pending exec_mcp. Ending the turn there drops that exec_mcp, leaving a
      // narration-only finish_reason "stop" with zero tool_calls (#10215). On
      // this family only the real terminal signals (turn_ended,
      // tool_call_completed, server_end) decide — kvAfterTextSeen is kept purely
      // as an observational flag, never as the turn terminator.
      ctx.kvAfterTextSeen = true;
      if (isComposerModel(ctx.model)) {
        ctx.endReason = "kv_after_text";
      }
    }
  }
}

export class CursorExecutor extends BaseExecutor {
  constructor(provider: "cursor" | "cursor-api" = "cursor") {
    super(provider, PROVIDERS[provider]);
  }

  buildUrl() {
    return PROVIDERS.cursor.baseUrl;
  }

  /**
   * API-key connections carry a `crsr_…` key that api2.cursor.sh does not
   * accept as a Bearer; swap it for the exchanged session token before the
   * h2 stream is opened. OAuth/IDE-session connections pass through untouched.
   */
  async resolveExecutionCredentials(credentials) {
    if (!isCursorApiKey(credentials?.apiKey)) return credentials;
    try {
      const accessToken = await resolveCursorBearerToken(credentials);
      return { ...credentials, accessToken };
    } catch (err) {
      const status =
        err instanceof CursorApiKeyExchangeError ? err.status : HTTP_STATUS.SERVER_ERROR;
      const message = err instanceof Error ? err.message : String(err);
      return new Response(
        JSON.stringify({
          error: {
            message: sanitizeErrorMessage(message),
            type: status === HTTP_STATUS.UNAUTHORIZED ? "authentication_error" : "connection_error",
            code: "",
          },
        }),
        { status, headers: { "Content-Type": "application/json" } }
      );
    }
  }

  buildHeaders(credentials) {
    const ghostMode = credentials.providerSpecificData?.ghostMode !== false;
    const cleanToken = stripCursorOAuthTokenPrefix(credentials.accessToken ?? "");
    const requestId = crypto.randomUUID();
    const traceParent = `00-${crypto.randomBytes(16).toString("hex")}-${crypto.randomBytes(8).toString("hex")}-01`;

    // Mirrors cursor-agent's actual headers for agent.v1.AgentService/Run.
    // Notably: no x-cursor-checksum, no machineId, no x-amzn-trace-id.
    // Only advertise gzip (not brotli) — our Connect-RPC frame decoder
    // only handles gzip-compressed message bodies.
    return {
      authorization: `Bearer ${cleanToken}`,
      "backend-traceparent": traceParent,
      "connect-accept-encoding": "gzip",
      "connect-protocol-version": "1",
      "content-type": "application/connect+proto",
      traceparent: traceParent,
      "user-agent": "connect-es/1.6.1",
      "x-cursor-client-type": "cli",
      "x-cursor-client-version": formatCursorAgentClientVersion(getCursorAgentCliVersion()),
      "x-ghost-mode": ghostMode ? "true" : "false",
      "x-original-request-id": requestId,
      "x-request-id": requestId,
    };
  }

  /**
   * Build the request body and return it alongside the request-scoped
   * blobStore. cursor's models (auto, claude-*, gpt-*) don't reliably
   * follow system-role content delivered via the KV blob channel — even
   * though the blob is requested and our reply is accepted, the model
   * proceeds without applying the prompt.
   *
   * As a pragmatic workaround we prepend the system content into the
   * UserMessage text (the pre-Phase-7 behavior). The KV-blob handshake
   * machinery is still in place for any future schema where cursor honors
   * root_prompt_messages_json semantically — verified end-to-end with
   * wire-tap captures.
   */
  /**
   * Assemble the user text + resolved tools shared by the sync (transformRequest)
   * and async (buildRequest) request builders. Image resolution is intentionally
   * NOT done here — it's async and only the cold-path buildRequest needs it.
   */
  private assembleTextAndTools(body: {
    messages?: ChatMessage[];
    tools?: unknown;
    tool_choice?: unknown;
    max_tokens?: unknown;
    max_completion_tokens?: unknown;
    stop?: unknown;
    response_format?: unknown;
  }): { userText: string; tools: OpenAITool[] | undefined } {
    const messages: ChatMessage[] = body.messages || [];
    const declaredTools: OpenAITool[] | undefined = Array.isArray(body.tools)
      ? (body.tools as OpenAITool[])
      : undefined;
    // tool_choice:"none" means "do not call any tool" — honor it by advertising
    // no tools at all (matches OpenAI semantics; composer-api does the same).
    const tools = body.tool_choice === "none" ? undefined : declaredTools;

    // flattenMessages prepends any role:"system" messages into the user
    // text (proven path that cursor's models honor). Image parts in the content
    // are ignored here (they carry no text) and resolved separately.
    let userText = flattenMessages(messages);

    // When the request declares tools, prepend the tool-commit directive so
    // composer-2.5 reliably invokes them instead of narrating intent and
    // stopping. Measured live: tool-call rate ~53% → ~88% with the directive.
    // tool_choice "required"/specific-function add a forcing line on top.
    // Default-on; set CURSOR_TOOL_DIRECTIVE=0 to opt out. See TOOL_COMMIT_DIRECTIVE.
    if (tools && tools.length > 0 && process.env.CURSOR_TOOL_DIRECTIVE !== "0") {
      userText = `${TOOL_COMMIT_DIRECTIVE}${toolChoiceDirectiveLine(body.tool_choice)}\n\n${userText}`;
    }

    // Surface OpenAI output params cursor ignores natively (response_format /
    // max_tokens / stop) as trailing prompt constraints.
    userText += buildCursorOutputConstraints(body);

    return { userText, tools };
  }

  /**
   * Resolve any OpenAI image_url parts in the request's user messages into
   * inlined cursor images. Returns undefined when the request carries no
   * images (keeps the request byte-identical to the text-only path). Throws
   * CursorImageError on invalid / oversized / SSRF-blocked input.
   */
  private async resolveRequestImages(body: {
    messages?: ChatMessage[];
  }): Promise<EncodedImage[] | undefined> {
    const messages: ChatMessage[] = body.messages || [];
    const imageUrls: string[] = [];
    for (const m of messages) {
      // Images only ride on user turns (the openai-to-cursor translator keeps
      // them only there). System/assistant/tool turns carry no vision input.
      if (m.role === "user") {
        for (const u of extractImageUrls(m.content)) imageUrls.push(u);
      }
    }
    if (imageUrls.length === 0) return undefined;
    return resolveCursorImages(imageUrls);
  }

  /**
   * Exact ids from the active Cursor synced catalog. Empty/unavailable →
   * undefined so resolveRequestedModel keeps #7289 offline splitting.
   */
  private async loadLiveCatalogIds(): Promise<ReadonlySet<string> | undefined> {
    try {
      const catalog = await getActiveSyncedCatalog(this.provider);
      if (!catalog.models.length) return undefined;
      return new Set(catalog.models.map((model) => model.id));
    } catch {
      return undefined;
    }
  }

  private async buildRequest(
    model: string,
    body: {
      messages?: ChatMessage[];
      tools?: unknown;
      tool_choice?: unknown;
      conversation_id?: string;
      max_tokens?: unknown;
      max_completion_tokens?: unknown;
      stop?: unknown;
      response_format?: unknown;
    }
  ): Promise<{ body: Uint8Array; blobStore: Map<string, Buffer> }> {
    const { userText, tools } = this.assembleTextAndTools(body);
    const [images, liveCatalogIds] = await Promise.all([
      this.resolveRequestImages(body),
      this.loadLiveCatalogIds(),
    ]);

    const blobStore = new Map<string, Buffer>();
    const requestBody = buildAgentRequestBody({
      modelId: model,
      userText,
      conversationId: body.conversation_id,
      tools,
      blobStore,
      images,
      liveCatalogIds,
    });
    return { body: requestBody, blobStore };
  }

  transformRequest(model, body, _stream, _credentials) {
    // Sync interface method (not used by cursor's own execute() path, which
    // uses the async buildRequest). Text-only — image resolution is async.
    const { userText, tools } = this.assembleTextAndTools(body);
    const blobStore = new Map<string, Buffer>();
    return buildAgentRequestBody({
      modelId: model,
      userText,
      conversationId: body.conversation_id,
      tools,
      blobStore,
    });
  }

  // ─── h2 lifecycle: open + drive (Phase 4 streaming refactor) ─────────────
  //
  // openH2 establishes the bidirectional stream and waits for the response
  // headers (so we can decide whether to commit to a streaming SSE Response
  // or return an error). driveH2 then consumes data events incrementally,
  // dispatching frames through processFrame so SSE chunks land on the
  // ReadableStream controller as the upstream produces them.
  //
  // The fetch fallback (cloud envs without http2) preserves the legacy
  // buffer-then-decode behavior — Connect-RPC bidirectional ack-on-same-stream
  // can't run over a one-shot fetch anyway.

  private async openH2(
    url: string,
    headers: Record<string, string>,
    body: Uint8Array,
    signal?: AbortSignal
  ): Promise<{
    status: number;
    headers: Record<string, string | number>;
    client: import("http2").ClientHttp2Session;
    req: import("http2").ClientHttp2Stream;
    initialBytes: Buffer;
    consumeError: () => Promise<Buffer>;
  }> {
    if (!http2) throw new Error("http2 module not available");

    return new Promise((resolve, reject) => {
      const urlObj = new URL(url);
      const client = http2!.connect(`https://${urlObj.host}`);
      const earlyChunks: Buffer[] = [];
      let resolved = false;

      client.on("error", (err) => {
        if (!resolved) reject(err);
      });

      const req = client.request({
        ":method": "POST",
        ":path": urlObj.pathname,
        ":authority": urlObj.host,
        ":scheme": "https",
        ...headers,
      });

      const onAbort = () => {
        try {
          req.close();
          client.close();
        } catch {
          // Expected: connection may already be closed
        }
        if (!resolved) {
          resolved = true;
          reject(new Error("aborted"));
        }
      };
      if (signal) signal.addEventListener("abort", onAbort);

      req.on("response", (h) => {
        if (resolved) return;
        resolved = true;
        const status = Number(h[":status"] ?? HTTP_STATUS.SERVER_ERROR);
        // For non-200 statuses, drain the remaining body for an error message.
        // The caller calls consumeError() to await the full body.
        const consumeError = () =>
          new Promise<Buffer>((res) => {
            const out = [...earlyChunks];
            req.on("data", (c) => out.push(Buffer.from(c)));
            req.on("end", () => {
              try {
                req.close();
                client.close();
              } catch {
                // Expected: connection may already be closed
              }
              if (signal) signal.removeEventListener("abort", onAbort);
              res(Buffer.concat(out));
            });
            req.on("error", () => {
              try {
                req.close();
                client.close();
              } catch {
                // Expected: connection may already be closed
              }
              if (signal) signal.removeEventListener("abort", onAbort);
              res(Buffer.concat(out));
            });
          });
        resolve({
          status,
          headers: h as Record<string, string | number>,
          client,
          req,
          initialBytes: Buffer.concat(earlyChunks),
          consumeError,
        });
      });

      // Buffer any data that arrives before the response event resolves.
      // (In practice the response event fires first, but this guards against
      // implementation differences in node:http2.)
      req.on("data", (chunk) => {
        if (!resolved) earlyChunks.push(Buffer.from(chunk));
      });

      req.on("error", (err) => {
        if (!resolved) {
          resolved = true;
          if (signal) signal.removeEventListener("abort", onAbort);
          reject(err);
        }
      });

      // Bidirectional streaming: write the init message but DO NOT send
      // END_STREAM — cursor's server stops responding once we close our side.
      // Guard the write like every h2Req.write in processFrame: a synchronous
      // failure here (e.g. stream already torn down) would otherwise leave the
      // request hung until the safety timeout instead of failing fast.
      try {
        req.write(body);
      } catch (err) {
        if (!resolved) {
          resolved = true;
          if (signal) signal.removeEventListener("abort", onAbort);
          try {
            req.close();
            client.close();
          } catch {
            // Expected: connection may already be closed
          }
          reject(err instanceof Error ? err : new Error(String(err)));
        }
      }
    });
  }

  /**
   * Drive an open h2 stream to completion. processFrame populates ctx as
   * each Connect-RPC frame is decoded; the loop closes when ctx.endReason
   * is set (turn_ended, kv_after_text, server_end) or the stream errors.
   *
   * Phase 8 will add a max-stream safety timeout here.
   */
  private driveH2(
    h2: {
      req: import("http2").ClientHttp2Stream;
      client: import("http2").ClientHttp2Session;
      initialBytes: Buffer;
    },
    ctx: StreamCtx,
    mcpTools: McpToolDefinition[] | undefined,
    blobStore: Map<string, Buffer> | undefined,
    clientPlatform: CursorClientPlatform | undefined,
    todoHistory: CursorTodoHistoryItem[] | undefined,
    signal?: AbortSignal
  ): Promise<void> {
    const ackedExecIds = new Set<string>();
    // Rolling buffer: chunks arrive on `data`, get appended, and consumed
    // frames are sliced off so we don't re-scan + re-concat on every event
    // (avoids O(N²) for long-running streams).
    let buf: Buffer = h2.initialBytes.length > 0 ? h2.initialBytes : Buffer.alloc(0);

    return new Promise((resolve, reject) => {
      let scanning = false;
      let settled = false;
      // Phase 8: safety timeout. If neither turn_ended, kv_after_text, nor
      // server-end fires within CURSOR_STREAM_TIMEOUT_MS, abort the stream
      // so a stuck upstream doesn't keep the response open indefinitely.
      const safetyTimer = setTimeout(() => {
        if (ctx.endReason) return;
        debugLog("[cursor-agent] stream safety timeout fired");
        teardown();
        reject(new Error("cursor-agent stream timed out"));
      }, CURSOR_STREAM_TIMEOUT_MS);

      const onData = (chunk: Buffer) => {
        if (CURSOR_DEBUG && process.env.CURSOR_DUMP_FILE) {
          fs.appendFileSync(process.env.CURSOR_DUMP_FILE, chunk);
        }
        buf = buf.length === 0 ? Buffer.from(chunk) : Buffer.concat([buf, chunk]);
        void tryScan();
      };
      const onEnd = () => {
        if (settled) return;
        settled = true;
        if (!ctx.endReason) ctx.endReason = "server_end";
        detachListeners();
        resolve();
      };
      const onErr = (err: Error) => {
        if (settled) return;
        settled = true;
        teardown();
        reject(err);
      };
      const onAbort = () => {
        if (settled) return;
        settled = true;
        teardown();
        reject(new Error("aborted"));
      };

      // detachListeners removes data/end/error/abort handlers and clears the
      // safety timer. Called on successful resolve when the caller keeps the
      // h2 alive (Phase 6 session reuse).
      const detachListeners = () => {
        clearTimeout(safetyTimer);
        h2.req.off("data", onData);
        h2.req.off("end", onEnd);
        h2.req.off("error", onErr);
        if (signal) signal.removeEventListener("abort", onAbort);
      };
      // teardown additionally closes the h2 stream. Used on error / abort /
      // safety-timeout — the connection isn't worth keeping at that point.
      const teardown = () => {
        detachListeners();
        try {
          h2.req.close();
          h2.client.close();
        } catch {
          // Expected: connection may already be closed during teardown
        }
      };

      if (signal) signal.addEventListener("abort", onAbort);

      const hasCompleteFrame = () => buf.length >= 5 && buf.length >= 5 + buf.readUInt32BE(1);

      const tryScan = async () => {
        if (scanning || settled) return;
        scanning = true;
        try {
          let pos = 0;
          while (!settled && pos + 5 <= buf.length) {
            const length = buf.readUInt32BE(pos + 1);
            if (length > CURSOR_MAX_FRAME_BYTES) {
              // Refuse to buffer an implausibly large frame — fail fast instead
              // of letting the rolling buffer grow toward OOM.
              settled = true;
              teardown();
              reject(new Error(`cursor-agent frame too large (${length} bytes)`));
              return;
            }
            if (pos + 5 + length > buf.length) break; // partial frame; wait
            const flag = buf[pos];
            const raw = buf.subarray(pos + 5, pos + 5 + length);
            // Per-frame error isolation: if gunzip or processFrame throws on
            // one frame, log and skip past it instead of getting stuck on
            // the same offset and hanging until the safety timer fires.
            try {
              const payload = flag & 0x1 ? await gunzipAsync(raw) : raw;
              if (settled) return;
              processFrame(payload, ctx, ackedExecIds, {
                h2Req: h2.req,
                mcpTools,
                blobStore,
                clientPlatform,
                todoHistory,
              });
            } catch (err) {
              debugLog(
                "[cursor-agent] frame decode failed at pos",
                pos,
                ":",
                (err as Error).message
              );
            }
            pos += 5 + length;
            if (ctx.endReason) {
              buf = buf.subarray(pos);
              settled = true;
              detachListeners();
              resolve();
              return;
            }
          }
          // Splice off processed bytes so the buffer stays bounded.
          if (pos > 0) buf = buf.subarray(pos);
        } finally {
          scanning = false;
        }

        if (!settled && hasCompleteFrame()) {
          void tryScan();
        }
      };

      h2.req.on("data", onData);
      h2.req.on("end", onEnd);
      h2.req.on("error", onErr);

      // Process any bytes already buffered from openH2.
      void tryScan();
    });
  }

  async execute({ model, body, stream, credentials, signal, log, upstreamExtraHeaders }) {
    const fallbackUrl = this.buildUrl();
    const executionCredentials = await this.resolveExecutionCredentials(credentials);
    if (executionCredentials instanceof Response) {
      return {
        response: executionCredentials,
        url: fallbackUrl,
        headers: {},
        transformedBody: body,
      };
    }
    let url: string;
    try {
      url = await resolveCursorAgentUrl(executionCredentials, signal);
    } catch (err) {
      const message = err instanceof Error ? err.message : String(err);
      const headers = this.buildHeaders(executionCredentials);
      return {
        response: new Response(
          JSON.stringify({
            error: {
              message: sanitizeErrorMessage(message),
              type: "connection_error",
              code: "",
            },
          }),
          {
            status: err instanceof CursorServerConfigError ? err.status : HTTP_STATUS.SERVER_ERROR,
            headers: { "Content-Type": "application/json" },
          }
        ),
        url: fallbackUrl,
        headers,
        transformedBody: body,
      };
    }
    const headers = this.buildHeaders(executionCredentials);
    mergeUpstreamExtraHeaders(headers, upstreamExtraHeaders);

    const messages: ChatMessage[] = body.messages || [];
    const conversationId: string =
      typeof body.conversation_id === "string" && body.conversation_id
        ? body.conversation_id
        : crypto.randomUUID();
    const lastMessage = messages[messages.length - 1];
    const isToolFollowUp = lastMessage?.role === "tool";

    // Tools embedded in the RequestContext ack throughout the turn —
    // synced with mcp_tools in the encoded request body.
    const declaredMcpTools: McpToolDefinition[] | undefined = Array.isArray(body.tools)
      ? openAIToolsToMcpDefs(body.tools as OpenAITool[])
      : undefined;
    const mcpTools = selectCursorBridgeTools(declaredMcpTools, body.tool_choice);
    const clientPlatform = inferCursorClientPlatform(messages);
    const todoHistory = extractLatestTodoHistory(messages);

    // Sanitize error messages: strip stack traces and absolute paths to
    // prevent information exposure. Shared helper in utils/error.ts.
    const buildErrorResponse = (status: number, message: string, type = "invalid_request_error") =>
      new Response(
        JSON.stringify({ error: { message: sanitizeErrorMessage(message), type, code: "" } }),
        { status, headers: { "Content-Type": "application/json" } }
      );

    // Cursor's agent.v1.AgentService/Run is a bidirectional Connect-RPC:
    // request_context, KV blob lookups, and exec rejections must be
    // written back on the same h2 stream while the response is still
    // being read. One-shot fetch can't do that, so cloud/edge runtimes
    // without node:http2 cannot drive cursor at all — fail fast with a
    // clear error rather than silently producing incomplete output.
    if (!http2) {
      return {
        response: buildErrorResponse(
          501,
          "Cursor provider requires Node.js http2, which is unavailable in this runtime (Edge / Cloudflare Workers / similar). Run OmniRoute on a Node.js runtime to use cursor.",
          "unsupported_runtime"
        ),
        url,
        headers,
        transformedBody: body,
      };
    }

    // ── h2 path with inline session manager (Phase 6) ──
    //
    // 1. If this is a tool-result follow-up (last message role:"tool") AND
    //    we have an alive session for the conversation, send the tool
    //    result on the existing h2 stream (inline resume).
    // 2. Otherwise, open a fresh h2 stream, send a new RunRequest, and
    //    register it as a session.
    //
    // Cold-resume fallback (acquire returns undefined, or sendToolResult
    // doesn't match): always lands on path #2, which now flattens the full
    // history (including role:"tool" messages) into UserText via
    // flattenMessages.

    type H2Like = {
      req: import("http2").ClientHttp2Stream;
      client: import("http2").ClientHttp2Session;
      initialBytes: Buffer;
    };

    let session: CursorSession | undefined;
    let h2: H2Like;
    let blobStore: Map<string, Buffer>;

    if (isToolFollowUp) {
      session = cursorSessionManager.acquire(conversationId);
      // #9029: content-based session match when client lacks conversation_id.
      if (!session && !body.conversation_id)
        session = cursorSessionManager.findByToolCallIds(
          messages.filter((m) => m.role === "tool" && m.tool_call_id).map((m) => m.tool_call_id!)
        );
    }

    if (session) {
      // Inline resume: send ExecMcpResult only for tool messages whose
      // tool_call_id is currently pending in this session. Older tool
      // messages from prior turns are already consumed by cursor and
      // sit in the request history harmlessly — sending them again
      // would either be a no-op or wedge the session, so we skip.
      // We require at least one match so we don't reuse the session
      // for a request that has no relevant tool results.
      blobStore = session.blobStore;
      let matched = 0;
      let hadFailure = false;
      for (const msg of messages) {
        if (msg.role !== "tool") continue;
        const id = msg.tool_call_id ?? "";
        if (!session.pendingToolCalls.has(id)) continue;
        const content = typeof msg.content === "string" ? msg.content : "";
        if (cursorSessionManager.sendToolResult(session, id, content, false)) {
          matched++;
        } else {
          hadFailure = true;
          break;
        }
      }
      if (matched === 0 || hadFailure) {
        cursorSessionManager.close(session);
        session = undefined;
      } else {
        h2 = {
          client: session.h2Client,
          req: session.h2Req,
          initialBytes: Buffer.alloc(0),
        };
      }
    }

    if (!session) {
      // Cold path: open fresh h2 stream with the full message history
      // flattened into UserText (Phase 6 flattenMessages handles role:"tool"
      // and assistant.tool_calls). buildRequest also resolves any image_url
      // parts (base64 / remote) into inlined cursor images.
      let built;
      try {
        built = await this.buildRequest(model, body);
      } catch (err) {
        // Image resolution failures (invalid / oversized / SSRF-blocked) are
        // client errors — return a sanitized 400 rather than a 500.
        if (err instanceof CursorImageError) {
          return {
            response: buildErrorResponse(err.status, err.message, "invalid_request_error"),
            url,
            headers,
            transformedBody: body,
          };
        }
        const message = err instanceof Error ? err.message : String(err);
        return {
          response: buildErrorResponse(HTTP_STATUS.SERVER_ERROR, message, "connection_error"),
          url,
          headers,
          transformedBody: body,
        };
      }
      blobStore = built.blobStore;
      let opened;
      try {
        opened = await this.openH2(url, headers, built.body, signal);
      } catch (err) {
        const message = err instanceof Error ? err.message : String(err);
        return {
          response: buildErrorResponse(HTTP_STATUS.SERVER_ERROR, message, "connection_error"),
          url,
          headers,
          transformedBody: body,
        };
      }
      if (opened.status !== 200) {
        const errBuf = await opened.consumeError();
        const errText = errBuf.toString("utf8") || "Unknown error";
        if (opened.status === HTTP_STATUS.UNAUTHORIZED && isCursorApiKey(credentials.apiKey)) {
          invalidateCursorSessionToken(credentials.apiKey);
        }
        return {
          response: buildErrorResponse(opened.status, `[${opened.status}]: ${errText}`),
          url,
          headers,
          transformedBody: body,
        };
      }
      h2 = opened;
      session = cursorSessionManager.open(conversationId, opened.client, opened.req, blobStore);
    }

    // Closure to share the post-drive lifecycle between stream/non-stream paths.
    const sessionToUse = session;
    const finishLifecycle = (ctx: StreamCtx, errored: boolean) => {
      // Persist any new pendingToolCalls from this turn into the session.
      for (const [id, info] of ctx.pendingToolCalls) {
        sessionToUse.pendingToolCalls.set(id, info);
      }
      if (errored || ctx.endReason !== "tool_calls" || ctx.requiresColdResume) {
        cursorSessionManager.close(sessionToUse);
      } else {
        cursorSessionManager.release(sessionToUse, "awaiting_tool_result");
      }
    };

    // Stream mode: ReadableStream that emits SSE chunks as they're decoded.
    if (stream !== false) {
      const enc = new TextEncoder();
      const sseStream = new ReadableStream(
        {
          start: async (controller) => {
            const ctx = newStreamCtx(model, (s) => controller.enqueue(enc.encode(s)));
            try {
              await this.driveH2(h2, ctx, mcpTools, blobStore, clientPlatform, todoHistory, signal);
              this.finalizeSseStream(ctx, body);
              finishLifecycle(ctx, false);
              controller.close();
            } catch (err) {
              // OpenCodex: NGHTTP2_CANCEL after client-tool suspend is expected — finish
              // the SSE turn instead of surfacing a transport failure.
              if (
                isCursorBenignCancelError(err) &&
                (ctx.totalText.length > 0 || ctx.pendingToolCalls.size > 0)
              ) {
                this.finalizeSseStream(ctx, body);
                finishLifecycle(ctx, false);
                controller.close();
                return;
              }
              finishLifecycle(ctx, true);
              controller.error(err);
            }
          },
        },
        { highWaterMark: 16384 }
      );
      return {
        response: new Response(sseStream, {
          status: 200,
          headers: {
            "Content-Type": "text/event-stream",
            "Cache-Control": "no-cache",
            Connection: "keep-alive",
          },
        }),
        url,
        headers,
        transformedBody: body,
      };
    }

    // Non-streaming: drive to completion, return chat.completion JSON.
    const ctx = newStreamCtx(model, () => {});
    try {
      await this.driveH2(h2, ctx, mcpTools, blobStore, clientPlatform, todoHistory, signal);
    } catch (err) {
      if (
        isCursorBenignCancelError(err) &&
        (ctx.totalText.length > 0 || ctx.pendingToolCalls.size > 0)
      ) {
        finishLifecycle(ctx, false);
        return {
          response: this.buildResponseFromCtx(ctx, body),
          url,
          headers,
          transformedBody: body,
        };
      }
      finishLifecycle(ctx, true);
      const message = err instanceof Error ? err.message : String(err);
      const classified = classifyCursorError(message);
      return {
        response: buildErrorResponse(classified.status, classified.message, classified.type),
        url,
        headers,
        transformedBody: body,
      };
    }
    finishLifecycle(ctx, false);
    return {
      response: this.buildResponseFromCtx(ctx, body),
      url,
      headers,
      transformedBody: body,
    };
  }

  /**
   * Emit the trailing SSE chunks (finish + usage + DONE) onto an already-open
   * stream. Called once driveH2 returns and ctx.endReason is set. The
   * mid-stream-error path emits an error chunk instead.
   */
  private finalizeSseStream(ctx: StreamCtx, body: { messages?: ChatMessage[] }) {
    if (ctx.midStreamError && ctx.totalText.length === 0) {
      emitCursorSseError(ctx, classifyCursorError(ctx.midStreamError.message));
      return;
    }

    // Silent empty turn (auth accepted, no text) — surface actionable error instead of
    // an empty assistant completion that chatCore maps to opaque "empty content" 502.
    if (isCursorEmptyTurn(ctx) && ctx.endReason && ctx.endReason !== "tool_calls") {
      emitCursorSseError(
        ctx,
        resolveCursorEmptyTurnError({
          upstreamMessage: ctx.midStreamError?.message,
        })
      );
      return;
    }

    if (!ctx.emittedRoleChunk) {
      // Edge case: empty response. Emit a role chunk so clients see at least
      // one delta before finish.
      emitChunk(ctx, { role: "assistant", content: "" });
    }

    // End-of-stream Composer inline tool-call fallback (decolua/9router#1335):
    // if the entire response arrived as a single big chunk (or the streaming
    // parser state never reached "ready"), try a full non-streaming parse on
    // the accumulated visible content so we still emit structured tool_calls
    // and don't leak the markers as plain text.
    if (isComposerModel(ctx.model) && !ctx.composerInlineToolCallsEmitted && ctx.totalText) {
      const parsed = parseComposerToolCalls(ctx.totalText);
      if (parsed.toolCalls.length > 0) {
        ctx.composerInlineToolCallsEmitted = true;
        // Replace totalText with the residual (markers stripped).
        ctx.totalText = parsed.content;
        for (const tc of parsed.toolCalls) {
          const toolCallIndex = ctx.emittedToolCallIndex++;
          ctx.toolCalls.push({
            id: tc.id,
            name: tc.function.name,
            argumentsJson: tc.function.arguments,
          });
          emitChunk(ctx, {
            tool_calls: [
              {
                index: toolCallIndex,
                id: tc.id,
                type: "function",
                function: { name: tc.function.name, arguments: tc.function.arguments },
              },
            ],
          });
        }
      }
    }

    // OpenAI finish_reason: "tool_calls" if the model invoked any declared
    // tool, else "stop". A turn with mixed text + tool_calls finishes with
    // "tool_calls" (the tool calls are the actionable signal for the client).
    const finishReason = ctx.toolCalls.length > 0 ? "tool_calls" : "stop";
    emitChunk(ctx, {}, finishReason);
    emitUsage(ctx, body);
    emitDone(ctx);
  }

  /**
   * Build a non-streaming chat.completion JSON Response from a fully-driven
   * StreamCtx. The streaming path emits chunks live via finalizeSseStream
   * and never calls this method.
   */
  private buildResponseFromCtx(ctx: StreamCtx, body: { messages?: ChatMessage[] }): Response {
    if (ctx.midStreamError && ctx.totalText.length === 0) {
      const classified = classifyCursorError(ctx.midStreamError.message);
      return new Response(
        JSON.stringify({
          error: {
            message: classified.message,
            type: classified.type,
          },
        }),
        {
          status: classified.status,
          headers: { "Content-Type": "application/json" },
        }
      );
    }

    if (isCursorEmptyTurn(ctx) && ctx.endReason && ctx.endReason !== "tool_calls") {
      const empty = resolveCursorEmptyTurnError({
        upstreamMessage: ctx.midStreamError?.message,
      });
      return new Response(
        JSON.stringify({
          error: {
            message: empty.message,
            type: empty.type,
          },
        }),
        {
          status: empty.status,
          headers: { "Content-Type": "application/json" },
        }
      );
    }

    // Non-streaming: chat.completion shape. Include tool_calls in the
    // assistant message when the model invoked any (Phase 5).

    // Composer DeepSeek inline tool-call fallback (decolua/9router#1335): for
    // non-streaming requests, the streaming parser never runs — parse the
    // accumulated visible content once here instead.
    if (isComposerModel(ctx.model) && !ctx.composerInlineToolCallsEmitted && ctx.totalText) {
      const parsed = parseComposerToolCalls(ctx.totalText);
      if (parsed.toolCalls.length > 0) {
        ctx.composerInlineToolCallsEmitted = true;
        ctx.totalText = parsed.content;
        for (const tc of parsed.toolCalls) {
          ctx.toolCalls.push({
            id: tc.id,
            name: tc.function.name,
            argumentsJson: tc.function.arguments,
          });
        }
      }
    }

    const usage = buildCursorUsage(ctx, body);
    const finishReason = ctx.toolCalls.length > 0 ? "tool_calls" : "stop";
    const message: {
      role: "assistant";
      content: string | null;
      reasoning_content?: string;
      tool_calls?: Array<{
        id: string;
        type: "function";
        function: { name: string; arguments: string };
      }>;
    } = {
      role: "assistant",
      content: ctx.totalText.length > 0 ? ctx.totalText : null,
    };
    if (ctx.thinkingText.length > 0) {
      // Composer: strip the visible reply (after `</think>`) from the reasoning
      // payload so it is not duplicated — it already lives in message.content
      // via the processFrame thinking handler.
      const reasoningPayload = isComposerModel(ctx.model)
        ? composerReasoningRemainder(ctx.thinkingText)
        : ctx.thinkingText;
      if (reasoningPayload.length > 0) {
        message.reasoning_content = reasoningPayload;
      }
    }
    if (ctx.toolCalls.length > 0) {
      message.tool_calls = ctx.toolCalls.map((tc) => ({
        id: tc.id,
        type: "function",
        function: { name: tc.name, arguments: tc.argumentsJson },
      }));
    }
    return new Response(
      JSON.stringify({
        id: ctx.responseId,
        object: "chat.completion",
        created: ctx.created,
        model: ctx.model,
        choices: [
          {
            index: 0,
            message,
            finish_reason: finishReason,
          },
        ],
        usage,
      }),
      { status: 200, headers: { "Content-Type": "application/json" } }
    );
  }

  async refreshCredentials(credentials, log) {
    if (!credentials?.refreshToken) {
      log?.warn?.(
        "TOKEN_REFRESH",
        "Cursor: no refresh token available, re-authentication required"
      );
      return null;
    }
    const result = await getAccessToken("cursor", credentials, log);
    if (!result || result.error) {
      log?.warn?.(
        "TOKEN_REFRESH",
        `Cursor: token refresh failed${result?.error ? ` (${result.error})` : ""} — re-authentication required`
      );
      return null;
    }
    return result;
  }
}

export default CursorExecutor;