copilotRequestHandler.js

"use strict";
var __defProp = Object.defineProperty;
var __getOwnPropDesc = Object.getOwnPropertyDescriptor;
var __getOwnPropNames = Object.getOwnPropertyNames;
var __hasOwnProp = Object.prototype.hasOwnProperty;
var __export = (target, all) => {
  for (var name in all)
    __defProp(target, name, { get: all[name], enumerable: true });
};
var __copyProps = (to, from, except, desc) => {
  if (from && typeof from === "object" || typeof from === "function") {
    for (let key of __getOwnPropNames(from))
      if (!__hasOwnProp.call(to, key) && key !== except)
        __defProp(to, key, { get: () => from[key], enumerable: !(desc = __getOwnPropDesc(from, key)) || desc.enumerable });
  }
  return to;
};
var __toCommonJS = (mod) => __copyProps(__defProp({}, "__esModule", { value: true }), mod);
var copilotRequestHandler_exports = {};
__export(copilotRequestHandler_exports, {
  CopilotRequestHandler: () => CopilotRequestHandler,
  CopilotWebSocketCloseStatus: () => CopilotWebSocketCloseStatus,
  CopilotWebSocketForwarder: () => CopilotWebSocketForwarder,
  CopilotWebSocketHandler: () => CopilotWebSocketHandler,
  createCopilotRequestAdapter: () => createCopilotRequestAdapter
});
module.exports = __toCommonJS(copilotRequestHandler_exports);
const sharedTextDecoder = new TextDecoder("utf-8", { fatal: false });
const sharedTextEncoder = new TextEncoder();
const kBridge = /* @__PURE__ */ Symbol("copilotWebSocketResponseBridge");
const kCompletion = /* @__PURE__ */ Symbol("copilotWebSocketCompletion");
const kOpen = /* @__PURE__ */ Symbol("copilotWebSocketOpen");
const kSuppressCloseOnDispose = /* @__PURE__ */ Symbol("copilotWebSocketSuppressCloseOnDispose");
const kHandle = /* @__PURE__ */ Symbol("copilotRequestHandle");
class CopilotWebSocketCloseStatus {
  constructor(description, errorCode, error) {
    this.description = description;
    this.errorCode = errorCode;
    this.error = error;
  }
  description;
  errorCode;
  error;
  static normalClosure = new CopilotWebSocketCloseStatus();
}
class CopilotWebSocketHandler {
  #response;
  #completion;
  #resolveCompletion;
  #closed = false;
  [kSuppressCloseOnDispose] = false;
  context;
  constructor(context) {
    this.context = context;
    const bridge = context[kBridge];
    if (!bridge) {
      throw new Error("WebSocket response bridge is not attached");
    }
    this.#response = bridge;
    this.#completion = new Promise((resolve) => {
      this.#resolveCompletion = resolve;
    });
  }
  async sendResponseMessage(data) {
    await this.#response.write(data);
  }
  async close(status = CopilotWebSocketCloseStatus.normalClosure) {
    if (this.#closed) {
      return;
    }
    this.#closed = true;
    if (status.error) {
      await this.#response.error({
        message: status.description ?? status.error.message,
        code: status.errorCode
      });
    } else {
      await this.#response.end();
    }
    this.#resolveCompletion(status);
  }
  async [Symbol.asyncDispose]() {
    if (!this[kSuppressCloseOnDispose] && !this.#closed) {
      await this.close(CopilotWebSocketCloseStatus.normalClosure);
    }
  }
  /** @internal */
  get [kCompletion]() {
    return this.#completion;
  }
  /** @internal */
  async [kOpen]() {
  }
}
class CopilotWebSocketForwarder extends CopilotWebSocketHandler {
  #upstream = null;
  constructor(context) {
    super(context);
  }
  sendRequestMessage(data) {
    if (this.#upstream?.readyState !== WebSocket.OPEN) {
      return;
    }
    this.#upstream.send(data);
  }
  /** @internal */
  async [kOpen]() {
    if (this.#upstream) {
      return;
    }
    const upstream = new WebSocket(this.context.url);
    upstream.binaryType = "arraybuffer";
    this.#upstream = upstream;
    upstream.addEventListener("message", (event) => {
      void this.sendResponseMessage(normalizeWsData(event.data)).catch(
        async (err) => {
          await this.close(
            new CopilotWebSocketCloseStatus(
              err instanceof Error ? err.message : String(err),
              void 0,
              err instanceof Error ? err : new Error(String(err))
            )
          );
        }
      );
    });
    upstream.addEventListener("close", () => {
      void this.close(CopilotWebSocketCloseStatus.normalClosure);
    });
    upstream.addEventListener("error", () => {
      void this.close(
        new CopilotWebSocketCloseStatus(
          "WebSocket error",
          void 0,
          new Error("WebSocket error")
        )
      );
    });
    await new Promise((resolve, reject) => {
      if (upstream.readyState === WebSocket.OPEN) {
        resolve();
        return;
      }
      upstream.addEventListener("open", () => resolve(), { once: true });
      upstream.addEventListener("error", () => reject(new Error("WebSocket error")), {
        once: true
      });
    });
  }
  async close(status = CopilotWebSocketCloseStatus.normalClosure) {
    try {
      if (this.#upstream?.readyState === WebSocket.OPEN || this.#upstream?.readyState === WebSocket.CONNECTING) {
        this.#upstream?.close();
      }
    } catch {
    }
    await super.close(status);
  }
  async [Symbol.asyncDispose]() {
    try {
      await super[Symbol.asyncDispose]();
    } finally {
      try {
        this.#upstream?.close();
      } catch {
      }
    }
  }
}
class CopilotRequestHandler {
  sendRequest(request, ctx) {
    return fetch(request, { signal: ctx.signal });
  }
  openWebSocket(ctx) {
    return Promise.resolve(new CopilotWebSocketForwarder(ctx));
  }
  /** @internal */
  async [kHandle](exchange) {
    const bridge = new CopilotWebSocketResponseBridge(exchange);
    const ctx = {
      requestId: exchange.requestId,
      sessionId: exchange.sessionId,
      transport: exchange.transport,
      url: exchange.url,
      headers: exchange.headers,
      signal: exchange.signal,
      [kBridge]: bridge
    };
    if (exchange.transport === "websocket") {
      await this.#handleWebSocket(exchange, ctx);
    } else {
      await this.#handleHttp(exchange, ctx);
    }
  }
  async #handleHttp(exchange, ctx) {
    const request = await buildFetchRequest(exchange);
    const response = await this.sendRequest(request, ctx);
    await streamResponse(response, exchange);
  }
  async #handleWebSocket(exchange, ctx) {
    const handler = await this.openWebSocket(ctx);
    try {
      await handler[kOpen]();
      await ctx[kBridge].start();
      let cancelled;
      const clientSettled = (async () => {
        for await (const chunk of exchange.requestBody) {
          await handler.sendRequestMessage(decodeFrame(chunk));
        }
        return "client-complete";
      })().catch((err) => {
        cancelled = err;
        return "client-error";
      });
      const first = await Promise.race([
        clientSettled,
        handler[kCompletion].then(() => "server-done")
      ]);
      if (first === "client-error") {
        handler[kSuppressCloseOnDispose] = true;
        throw cancelled instanceof Error ? cancelled : new Error(String(cancelled));
      }
      if (first === "client-complete") {
        await handler.close(CopilotWebSocketCloseStatus.normalClosure);
        await handler[kCompletion];
        return;
      }
      const status = await handler[kCompletion];
      if (status.error) {
        throw status.error;
      }
    } finally {
      await handler[Symbol.asyncDispose]();
    }
  }
}
function createCopilotRequestAdapter(handler, getServerRpc) {
  const pending = /* @__PURE__ */ new Map();
  function getOrCreate(requestId) {
    let exchange = pending.get(requestId);
    if (!exchange) {
      exchange = new CopilotRequestExchange(requestId, getServerRpc);
      pending.set(requestId, exchange);
    }
    return exchange;
  }
  async function run(exchange) {
    try {
      await handler[kHandle](exchange);
      if (!exchange.finished) {
        await finalize(
          exchange,
          502,
          "Copilot request handler returned without finalising the response (call responseBody.end() or .error())."
        );
      }
    } catch (err) {
      if (exchange.cancelled || exchange.signal.aborted) {
        await finalize(exchange, 499, "Request cancelled by runtime", "cancelled");
        return;
      }
      const message = err instanceof Error ? err.message : String(err);
      await finalize(exchange, 502, message);
    } finally {
      pending.delete(exchange.requestId);
    }
  }
  return {
    async httpRequestStart(params) {
      const exchange = getOrCreate(params.requestId);
      exchange.setContext(params);
      void run(exchange);
      return {};
    },
    async httpRequestChunk(params) {
      routeChunk(getOrCreate(params.requestId), params);
      return {};
    }
  };
}
async function finalize(exchange, status, message, code) {
  if (exchange.finished) {
    return;
  }
  try {
    if (!exchange.started) {
      await exchange.startResponse({ status, headers: {} });
    }
    await exchange.errorResponse({ message, code });
  } catch {
  }
}
function routeChunk(exchange, params) {
  if (params.cancel) {
    exchange.pushCancel(params.cancelReason);
    return;
  }
  if (params.data && params.data.length > 0) {
    exchange.pushChunk(decodeChunkData(params.data, !!params.binary));
  }
  if (params.end) {
    exchange.pushEnd();
  }
}
class CopilotRequestExchange {
  requestId;
  sessionId;
  method = "GET";
  url = "";
  headers = {};
  transport = "http";
  #getServerRpc;
  #abort = new AbortController();
  #buffer = [];
  #waker = null;
  #drained = false;
  #started = false;
  #finished = false;
  #cancelled = false;
  constructor(requestId, getServerRpc) {
    this.requestId = requestId;
    this.#getServerRpc = getServerRpc;
  }
  /** Fill in the request context once the matching start frame arrives. */
  setContext(params) {
    this.sessionId = params.sessionId;
    this.method = params.method;
    this.url = params.url;
    this.headers = params.headers;
    this.transport = params.transport ?? "http";
  }
  get signal() {
    return this.#abort.signal;
  }
  get started() {
    return this.#started;
  }
  get finished() {
    return this.#finished;
  }
  get cancelled() {
    return this.#cancelled;
  }
  // --- Request body feed (driven by the adapter as chunk frames arrive) ---
  pushChunk(chunk) {
    this.#push({ chunk });
  }
  pushEnd() {
    this.#push({ end: true });
  }
  pushCancel(reason) {
    this.#cancelled = true;
    this.#abort.abort();
    this.#push({ cancel: { reason } });
  }
  #push(item) {
    this.#buffer.push(item);
    const w = this.#waker;
    this.#waker = null;
    w?.();
  }
  /**
   * Request body bytes, yielded as they arrive. A cancel frame surfaces as a
   * thrown error so the handler's upstream call is torn down.
   */
  get requestBody() {
    return {
      [Symbol.asyncIterator]: () => ({
        next: async () => {
          if (this.#drained) {
            return { value: void 0, done: true };
          }
          while (this.#buffer.length === 0) {
            await new Promise((resolve) => {
              this.#waker = resolve;
            });
          }
          const item = this.#buffer.shift();
          if (item.cancel) {
            this.#drained = true;
            throw new Error(
              item.cancel.reason ? `Request cancelled by runtime: ${item.cancel.reason}` : "Request cancelled by runtime"
            );
          }
          if (item.end) {
            this.#drained = true;
            return { value: void 0, done: true };
          }
          return { value: item.chunk ?? new Uint8Array(), done: false };
        }
      })
    };
  }
  // --- Response emit (driven by the handler). Strict state machine: ---
  // startResponse once -> 0..N writeResponse -> exactly one of
  // endResponse / errorResponse.
  async startResponse(init) {
    if (this.#started) {
      throw new Error("Copilot request response start() called twice.");
    }
    if (this.#finished) {
      throw new Error("Copilot request response already finished.");
    }
    this.#started = true;
    await this.#rpc().llmInference.httpResponseStart({
      requestId: this.requestId,
      status: init.status,
      statusText: init.statusText,
      headers: init.headers ?? {}
    });
  }
  async writeResponse(data) {
    if (this.#cancelled) {
      throw new Error("Copilot request was cancelled by the runtime.");
    }
    if (!this.#started) {
      throw new Error("Copilot request response write() called before start().");
    }
    if (this.#finished) {
      throw new Error("Copilot request response write() called after end()/error().");
    }
    const isString = typeof data === "string";
    await this.#rpc().llmInference.httpResponseChunk({
      requestId: this.requestId,
      data: isString ? data : Buffer.from(data).toString("base64"),
      binary: !isString,
      end: false
    });
  }
  async endResponse() {
    if (this.#finished) {
      return;
    }
    this.#finished = true;
    await this.#rpc().llmInference.httpResponseChunk({
      requestId: this.requestId,
      data: "",
      end: true
    });
  }
  async errorResponse(error) {
    if (this.#finished) {
      return;
    }
    this.#finished = true;
    await this.#rpc().llmInference.httpResponseChunk({
      requestId: this.requestId,
      data: "",
      end: true,
      error: { message: error.message, code: error.code }
    });
  }
  #rpc() {
    const r = this.#getServerRpc();
    if (!r) {
      throw new Error("Copilot request response used after RPC connection closed.");
    }
    return r;
  }
}
const FORBIDDEN_REQUEST_HEADERS = /* @__PURE__ */ new Set([
  "host",
  "connection",
  "content-length",
  "transfer-encoding",
  "keep-alive",
  "upgrade",
  "proxy-connection",
  "te",
  "trailer"
]);
async function buildFetchRequest(exchange) {
  const headers = new Headers();
  for (const [name, values] of Object.entries(exchange.headers)) {
    if (!values) {
      continue;
    }
    if (FORBIDDEN_REQUEST_HEADERS.has(name.toLowerCase())) {
      continue;
    }
    for (const value of values) {
      headers.append(name, value);
    }
  }
  const method = exchange.method.toUpperCase();
  const hasBody = method !== "GET" && method !== "HEAD";
  let body;
  if (hasBody) {
    const buffered = await drainAsync(exchange.requestBody);
    if (buffered.length > 0) {
      body = buffered;
    }
  } else {
    await drainAsync(exchange.requestBody);
  }
  return new Request(exchange.url, { method, headers, body });
}
async function drainAsync(stream) {
  const parts = [];
  let total = 0;
  for await (const chunk of stream) {
    parts.push(chunk);
    total += chunk.byteLength;
  }
  if (parts.length === 0) {
    return new Uint8Array(0);
  }
  if (parts.length === 1) {
    return parts[0];
  }
  const out = new Uint8Array(total);
  let off = 0;
  for (const part of parts) {
    out.set(part, off);
    off += part.byteLength;
  }
  return out;
}
async function streamResponse(response, exchange) {
  await exchange.startResponse({
    status: response.status,
    statusText: response.statusText || void 0,
    headers: headersToMultiMap(response.headers)
  });
  const body = response.body;
  if (!body) {
    await exchange.endResponse();
    return;
  }
  const reader = body.getReader();
  try {
    for (; ; ) {
      const { value, done } = await reader.read();
      if (done) {
        break;
      }
      if (value && value.byteLength > 0) {
        await exchange.writeResponse(value);
      }
    }
    await exchange.endResponse();
  } finally {
    reader.releaseLock();
  }
}
function headersToMultiMap(headers) {
  const out = {};
  headers.forEach((value, name) => {
    if (name.toLowerCase() === "set-cookie") {
      return;
    }
    const list = out[name] ?? (out[name] = []);
    list.push(value);
  });
  const setCookies = headers.getSetCookie();
  if (setCookies.length > 0) {
    out["set-cookie"] = setCookies;
  }
  return out;
}
function decodeChunkData(data, binary) {
  if (binary) {
    return new Uint8Array(Buffer.from(data, "base64"));
  }
  return sharedTextEncoder.encode(data);
}
function decodeFrame(chunk) {
  return sharedTextDecoder.decode(chunk);
}
function normalizeWsData(data) {
  if (typeof data === "string") {
    return data;
  }
  if (data instanceof Uint8Array) {
    return data;
  }
  if (data instanceof ArrayBuffer) {
    return new Uint8Array(data);
  }
  return new Uint8Array();
}
class CopilotWebSocketResponseBridge {
  #exchange;
  #started = false;
  #completed = false;
  #serial = Promise.resolve();
  constructor(exchange) {
    this.#exchange = exchange;
  }
  /** Emit the 101 upgrade head now, acknowledging the WebSocket connect. */
  start() {
    return this.#run(false, () => Promise.resolve());
  }
  write(data) {
    return this.#run(false, () => this.#exchange.writeResponse(data));
  }
  end() {
    return this.#run(true, () => this.#exchange.endResponse());
  }
  error(error) {
    return this.#run(true, () => this.#exchange.errorResponse(error));
  }
  #run(terminal, action) {
    const task = this.#serial.then(async () => {
      if (this.#completed) {
        return;
      }
      if (!this.#started) {
        this.#started = true;
        await this.#exchange.startResponse({ status: 101, headers: {} });
      }
      if (terminal) {
        this.#completed = true;
      }
      await action();
    });
    this.#serial = task.catch(() => {
    });
    return task;
  }
}
// Annotate the CommonJS export names for ESM import in node:
0 && (module.exports = {
  CopilotRequestHandler,
  CopilotWebSocketCloseStatus,
  CopilotWebSocketForwarder,
  CopilotWebSocketHandler,
  createCopilotRequestAdapter
});