/**
 * This file is part of the NocoBase (R) project.
 * Copyright (c) 2020-2024 NocoBase Co., Ltd.
 * Authors: NocoBase Team.
 *
 * This project is dual-licensed under AGPL-3.0 and NocoBase Commercial License.
 * For more information, please refer to: https://www.nocobase.com/agreement.
 */

import { LLMProvider, LLMProviderMeta } from '@nocobase/plugin-ai';
import { EmbeddingProvider, SupportedModel } from '../utils/ai-types';
import { EmbeddingsInterface } from '@langchain/core/embeddings';
import { Model } from '@nocobase/database';
import path from 'node:path';
import fs from 'node:fs/promises';
import axios from 'axios';
import { Context } from '@nocobase/actions';

// Keepalive marker — zero-width space prefix to distinguish from real content
const KEEPALIVE_PREFIX = '\u200B\u200B\u200B';
const DEFAULT_KEEPALIVE_INTERVAL_MS = 5000;
const DEFAULT_STREAM_CHUNK_SIZE = 512;
const DEFAULT_REQUEST_TIMEOUT_MS = 30 * 60 * 1000;

function cloneDeep<T>(value: T): T {
  if (value == null || typeof value !== 'object') {
    return value;
  }

  const structuredCloneFn = (globalThis as any).structuredClone;
  if (typeof structuredCloneFn === 'function') {
    try {
      return structuredCloneFn(value);
    } catch {
      // Fall through to JSON cloning for plain API payloads.
    }
  }

  return JSON.parse(JSON.stringify(value));
}

/**
 * Resolve a module from the main NocoBase app's node_modules.
 */
function requireFromApp(moduleName: string) {
  const appNodeModules = process.env.NODE_MODULES_PATH || path.join(process.cwd(), 'node_modules');
  const resolved = require.resolve(moduleName, { paths: [appNodeModules] });
  return require(resolved);
}

let _ChatOpenAI: any = null;
function getChatOpenAI() {
  if (!_ChatOpenAI) {
    const mod = requireFromApp('@langchain/openai');
    _ChatOpenAI = mod.ChatOpenAI;
  }
  return _ChatOpenAI;
}

/**
 * Lazy-load ChatOpenAICompletions — the lower-level class used as the base
 * for ReasoningChatOpenAI so we can support reasoning_content round-trips
 * required by models like DeepSeek-R1.
 */
let _ChatOpenAICompletions: any = null;
function getChatOpenAICompletions() {
  if (!_ChatOpenAICompletions) {
    const mod = requireFromApp('@langchain/openai');
    _ChatOpenAICompletions = mod.ChatOpenAICompletions;
  }
  return _ChatOpenAICompletions;
}

/**
 * Strip the `__thought__<base64>` suffix that Gemini models append to
 * tool call IDs during streaming. The suffix is excessively long and
 * causes errors when langgraph reads messages back from history.
 */
export function stripGeminiThoughtSuffix(id: string | undefined): string | undefined {
  if (!id || typeof id !== 'string') return id;
  const idx = id.indexOf('__thought__');
  return idx !== -1 ? id.substring(0, idx) : id;
}

/**
 * Build tool_calls key for reasoning content map lookup.
 */
function getToolCallsKey(toolCalls: Array<{ id?: string; name?: string; function?: { name?: string } }> = []): string {
  return toolCalls
    .map((tc) => {
      const id = tc?.id ?? '';
      const name = tc?.name ?? tc?.function?.name ?? '';
      return `${id}:${name}`;
    })
    .join('|');
}

/**
 * Collect reasoning_content from history messages keyed by their tool_calls.
 * This is needed because some APIs (DeepSeek) require reasoning_content to
 * be present in assistant messages that precede tool results.
 */
function collectReasoningMap(messages: any[]): Map<string, string> {
  const reasoningMap = new Map<string, string>();
  for (const message of messages ?? []) {
    if (message?.getType?.() !== 'ai' && message?._getType?.() !== 'ai') continue;
    if (!message?.tool_calls?.length) continue;
    const reasoningContent = message?.additional_kwargs?.reasoning_content;
    if (typeof reasoningContent !== 'string' || !reasoningContent) continue;
    const key = getToolCallsKey(message.tool_calls);
    if (key) reasoningMap.set(key, reasoningContent);
  }
  return reasoningMap;
}

/**
 * Patch request messages to restore reasoning_content on assistant messages
 * that have tool_calls — needed for APIs that require it on round-trip.
 */
function patchRequestMessagesReasoning(request: any, reasoningMap?: Map<string, string>): void {
  if (!reasoningMap?.size || !Array.isArray(request?.messages)) return;
  const lastMsg = request.messages.at(-1);
  if (lastMsg?.role !== 'tool') return;
  for (const msg of request.messages) {
    if (msg?.role !== 'assistant') continue;
    if (!Array.isArray(msg.tool_calls) || msg.tool_calls.length === 0) continue;
    if (msg.reasoning_content) continue;
    const key = getToolCallsKey(msg.tool_calls);
    const rc = key ? reasoningMap.get(key) : undefined;
    if (rc) msg.reasoning_content = rc;
  }
}

const REASONING_MAP_KEY = '__nb_reasoning_map';

/**
 * Create a ReasoningChatOpenAI class that extends ChatOpenAICompletions.
 * This patches reasoning_content into the request messages before sending
 * to the API, which is required for models like DeepSeek-R1 that need
 * reasoning_content present in assistant messages during tool call cycles.
 */
function createReasoningChatClass() {
  const ChatOpenAICompletions = getChatOpenAICompletions();
  if (!ChatOpenAICompletions) {
    // Fallback — completions class not available, use plain ChatOpenAI
    return getChatOpenAI();
  }

  return class ReasoningChatOpenAI extends ChatOpenAICompletions {
    async _generate(messages: any[], options: any, runManager?: any) {
      const reasoningMap = collectReasoningMap(messages);
      return super._generate(messages, { ...(options || {}), [REASONING_MAP_KEY]: reasoningMap }, runManager);
    }

    async *_streamResponseChunks(messages: any[], options: any, runManager?: any) {
      const reasoningMap =
        options?.[REASONING_MAP_KEY] instanceof Map
          ? (options[REASONING_MAP_KEY] as Map<string, string>)
          : collectReasoningMap(messages);

      const stream = super._streamResponseChunks(
        messages,
        { ...(options || {}), [REASONING_MAP_KEY]: reasoningMap },
        runManager,
      );

      for await (const chunk of stream) {
        yield chunk;
      }
    }

    _convertCompletionsDeltaToBaseMessageChunk(delta: any, rawResponse: any, defaultRole?: any) {
      const messageChunk = super._convertCompletionsDeltaToBaseMessageChunk(delta, rawResponse, defaultRole);
      if (delta?.reasoning_content) {
        messageChunk.additional_kwargs = {
          ...(messageChunk.additional_kwargs || {}),
          reasoning_content: delta.reasoning_content,
        };
      }
      return messageChunk;
    }

    _convertCompletionsMessageToBaseMessage(message: any, rawResponse: any) {
      const langChainMessage = super._convertCompletionsMessageToBaseMessage(message, rawResponse);
      if (message?.reasoning_content) {
        langChainMessage.additional_kwargs = {
          ...(langChainMessage.additional_kwargs || {}),
          reasoning_content: message.reasoning_content,
        };
      }
      return langChainMessage;
    }

    async completionWithRetry(request: any, requestOptions?: any): Promise<any> {
      const reasoningMap = requestOptions?.[REASONING_MAP_KEY] as Map<string, string> | undefined;
      patchRequestMessagesReasoning(request, reasoningMap);
      return super.completionWithRetry(request, requestOptions) as any;
    }
  };
}

let _AIMessageChunk: any = null;
function getAIMessageChunk() {
  if (!_AIMessageChunk) {
    const mod = requireFromApp('@langchain/core/messages');
    _AIMessageChunk = mod.AIMessageChunk;
  }
  return _AIMessageChunk;
}

function createAIMessageChunk(fields: Record<string, any>) {
  const AIMessageChunk = getAIMessageChunk();
  return new AIMessageChunk(fields);
}

function createKeepAliveChunk() {
  return createAIMessageChunk({
    content: KEEPALIVE_PREFIX,
    additional_kwargs: { __keepalive: true },
  });
}

function normalizePositiveNumber(value: any, fallback: number): number {
  const n = Number(value);
  return Number.isFinite(n) && n > 0 ? n : fallback;
}

function hasToolCallDelta(chunk: any): boolean {
  return Boolean(chunk?.tool_call_chunks?.length || chunk?.tool_calls?.length);
}

function createSplitMessageChunk(source: any, content: string, index: number, total: number) {
  const isFirst = index === 0;
  const isLast = index === total - 1;
  const fields: Record<string, any> = {
    content,
    additional_kwargs: isFirst ? cloneDeep(source?.additional_kwargs || {}) : {},
    response_metadata: isLast ? cloneDeep(source?.response_metadata || {}) : {},
  };

  if (isFirst && source?.id) {
    fields.id = source.id;
  }
  if (isFirst && source?.name) {
    fields.name = source.name;
  }
  if (isLast && source?.usage_metadata) {
    fields.usage_metadata = cloneDeep(source.usage_metadata);
  }
  if (isLast && source?.tool_calls?.length) {
    fields.tool_calls = cloneDeep(source.tool_calls);
  }
  if (isLast && source?.invalid_tool_calls?.length) {
    fields.invalid_tool_calls = cloneDeep(source.invalid_tool_calls);
  }

  return createAIMessageChunk(fields);
}

function splitTextChunk(chunk: any, chunkSize: number): any[] {
  const content = chunk?.content;
  if (
    typeof content !== 'string' ||
    content.length <= chunkSize ||
    content === KEEPALIVE_PREFIX ||
    hasToolCallDelta(chunk)
  ) {
    return [chunk];
  }

  const parts: string[] = [];
  for (let i = 0; i < content.length; i += chunkSize) {
    parts.push(content.slice(i, i + chunkSize));
  }
  return parts.map((part, index) => createSplitMessageChunk(chunk, part, index, parts.length));
}

function splitStreamChunk(chunk: any, chunkSize: number): any[] {
  return splitTextChunk(chunk, chunkSize);
}

async function* streamWithKeepAliveAndChunking(
  iterable: AsyncIterable<any>,
  options: { keepAlive: boolean; intervalMs: number; chunkSize: number },
) {
  const iterator = iterable?.[Symbol.asyncIterator]?.();
  if (!iterator) {
    return;
  }

  if (!options.keepAlive) {
    while (true) {
      const { value: chunk, done: chunkDone } = await iterator.next();
      if (chunkDone) break;
      for (const splitChunk of splitStreamChunk(chunk, options.chunkSize)) {
        yield splitChunk;
      }
    }
    return;
  }

  let done = false;
  let toolCallActive = false;
  let next = iterator.next();

  try {
    while (true) {
      let timer: ReturnType<typeof setTimeout> | null = null;
      const result = await Promise.race([
        next.then(
          (value) => ({ type: 'value' as const, value }),
          (error) => ({ type: 'error' as const, error }),
        ),
        new Promise<{ type: 'timeout' }>((resolve) => {
          timer = setTimeout(() => resolve({ type: 'timeout' }), options.intervalMs);
        }),
      ]);

      if (timer) {
        clearTimeout(timer);
      }

      if (result.type === 'timeout') {
        if (!toolCallActive) {
          yield createKeepAliveChunk();
        }
        continue;
      }

      if (result.type === 'error') {
        throw result.error;
      }

      if (result.value.done) {
        done = true;
        break;
      }

      const chunks = splitStreamChunk(result.value.value, options.chunkSize);
      for (const chunk of chunks) {
        if (hasToolCallDelta(chunk)) {
          toolCallActive = true;
        } else if (chunk?.content && chunk.content !== KEEPALIVE_PREFIX) {
          toolCallActive = false;
        }
        yield chunk;
      }

      next = iterator.next();
    }
  } finally {
    if (!done && typeof iterator.return === 'function') {
      try {
        await iterator.return();
      } catch {
        // Ignore cleanup errors from aborted upstream iterators.
      }
    }
  }
}

function stripToolCallTags(content: string): string | null {
  if (typeof content !== 'string') {
    return content;
  }
  return content.replace(/<[|｜]tool▁(?:calls▁begin|calls▁end|call▁begin|call▁end|sep)[|｜]>/g, '');
}

function extractTextContent(content: any, contentPath?: string): string {
  if (contentPath && contentPath !== 'auto') {
    try {
      const keys = contentPath.split('.');
      let result = content;
      for (const key of keys) {
        if (result == null) break;
        result = result[key];
      }
      if (typeof result === 'string') return result;
    } catch {
      // Fall through to auto
    }
  }
  if (typeof content === 'string') return content;
  if (Array.isArray(content)) {
    return content
      .filter((block: any) => block && block.type === 'text')
      .map((block: any) => block.text ?? '')
      .join('');
  }
  if (content && typeof content === 'object' && content.text) {
    return String(content.text);
  }
  return '';
}

/**
 * Detect whether a MIME type is text-decodable (UTF-8 safe).
 */
function isTextMimetype(mimetype?: string): boolean {
  if (!mimetype) return false;
  // All text/* subtypes are UTF-8 decodable
  if (mimetype.startsWith('text/')) return true;
  // Common text-based application types
  const TEXT_APPLICATION_TYPES = new Set([
    'application/json',
    'application/xml',
    'application/xhtml+xml',
    'application/atom+xml',
    'application/rss+xml',
    'application/csv',
    'application/javascript',
    'application/typescript',
    'application/x-javascript',
    'application/x-typescript',
    'application/x-yaml',
    'application/yaml',
    'application/x-json',
    'application/geo+json',
    'application/ld+json',
    'application/manifest+json',
    'application/graphql',
    'application/x-www-form-urlencoded',
    'application/toml',
    'application/x-sh',
    'application/x-shellscript',
    'application/sql',
  ]);
  return TEXT_APPLICATION_TYPES.has(mimetype);
}

function safeParseJSON(str: any, fieldName?: string): any {
  if (!str || typeof str !== 'string') return {};
  try {
    return JSON.parse(str);
  } catch (e) {
    // Warn so misconfigured JSON doesn't silently fall through to defaults
    console.warn(`[CustomLLM] Failed to parse ${fieldName || 'JSON config'}: ${(e as Error).message}`);
    return {};
  }
}

/**
 * Get a nested value from an object using a dot-path string.
 * e.g. getByPath({a:{b:"hello"}}, "a.b") => "hello"
 */
function getByPath(obj: any, dotPath: string): any {
  if (!obj || !dotPath) return undefined;
  const keys = dotPath.split('.');
  let current = obj;
  for (const key of keys) {
    if (current == null) return undefined;
    current = current[key];
  }
  return current;
}

/**
 * Create a custom fetch that intercepts LLM responses and maps them
 * from a non-standard format to OpenAI-compatible format.
 *
 * responseMapping config example:
 * {
 *   "content": "message.response"       // dot-path to the content field
 *   "role": "message.role"              // optional, dot-path to role (default: "assistant")
 *   "id": "id"                          // optional, dot-path to response id
 *   "tool_calls": "message.tool_calls"  // optional, dot-path to tool_calls array
 *   "finish_reason": "finish_reason"    // optional, dot-path to finish_reason
 * }
 */
function getHeaderValue(headers: HeadersInit | undefined, name: string): string | undefined {
  if (!headers) return undefined;
  if (headers instanceof Headers) return headers.get(name) || undefined;

  const lowerName = name.toLowerCase();
  if (Array.isArray(headers)) {
    const item = headers.find(([key]) => key.toLowerCase() === lowerName);
    return item?.[1];
  }

  for (const [key, value] of Object.entries(headers)) {
    if (key.toLowerCase() === lowerName) {
      return String(value);
    }
  }
  return undefined;
}

function parseRetryAfterSeconds(value: string | null): number | null {
  if (!value) return null;
  const seconds = Number(value);
  if (Number.isFinite(seconds) && seconds >= 0) return seconds * 1000;
  const date = Date.parse(value);
  if (Number.isFinite(date)) return Math.max(0, date - Date.now());
  return null;
}

export function createMappingFetch(
  responseMapping: Record<string, string>,
  retry?: { maxRetries?: number; delayMs?: number },
) {
  const contentPath = responseMapping.content;
  if (!contentPath) return undefined; // No mapping needed

  // Resolve path for tool_calls — if not set, try the standard OpenAI paths as fallback
  const toolCallsPath = responseMapping.tool_calls;
  const finishReasonPath = responseMapping.finish_reason;

  // Bounded non-streaming retry on 429/5xx. Streaming (SSE) responses are
  // never retried — a live stream cannot be replayed safely.
  const maxRetries = retry ? normalizePositiveNumber(retry.maxRetries, 2) : 0;
  const retryDelayMs = normalizePositiveNumber(retry?.delayMs, 1000);

  return async (url: RequestInfo | URL, init?: RequestInit): Promise<Response> => {
    const acceptHeader = getHeaderValue(init?.headers, 'accept') || '';
    const isStreamingRequest = acceptHeader.includes('text/event-stream');

    // Preserve the OpenAI SDK's AbortSignal. The SDK already applies its
    // request timeout before headers; replacing the signal here causes
    // user aborts/timeouts to be ignored and previously imposed a hard 120s cap.
    let response = await fetch(url, init);

    if (!isStreamingRequest && maxRetries > 0 && (response.status === 429 || response.status >= 500)) {
      for (let attempt = 0; attempt < maxRetries && (response.status === 429 || response.status >= 500); attempt++) {
        const retryAfter = parseRetryAfterSeconds(response.headers.get('retry-after'));
        const waitMs = retryAfter ?? retryDelayMs * (attempt + 1);
        await new Promise((resolve) => setTimeout(resolve, waitMs));
        response = await fetch(url, init);
      }
    }

    // Only intercept successful JSON responses
    if (!response.ok) return response;

    const contentType = response.headers.get('content-type') || '';

    // Handle streaming responses (SSE) — transform each chunk
    if (contentType.includes('text/event-stream') || acceptHeader.includes('text/event-stream')) {
      const reader = response.body?.getReader();
      if (!reader) return response;

      const stream = new ReadableStream({
        async start(controller) {
          const decoder = new TextDecoder();
          const encoder = new TextEncoder();
          let buffer = '';

          const enqueueChunked = (baseMapped: any) => {
            // SSE proxy layer chunk size — intentionally smaller than
            // DEFAULT_STREAM_CHUNK_SIZE (512) because we're splitting raw
            // HTTP response body bytes before LangChain parses them.
            const SSE_MAPPING_CHUNK_SIZE = 128;
            const delta = baseMapped.choices[0].delta;
            const content = delta.content;
            const toolCalls = delta.tool_calls;

            let hasEmitted = false;

            // 1. Process and stream content if it exists
            if (content !== undefined && content !== null) {
              const contentStr = String(content);
              if (contentStr.length > SSE_MAPPING_CHUNK_SIZE) {
                for (let i = 0; i < contentStr.length; i += SSE_MAPPING_CHUNK_SIZE) {
                  const chunkMapped = cloneDeep(baseMapped);
                  const newDelta = { ...delta };
                  if (i > 0) delete newDelta.role; // Only send role on first chunk
                  delete newDelta.tool_calls; // Handled separately

                  newDelta.content = contentStr.slice(i, i + SSE_MAPPING_CHUNK_SIZE);
                  chunkMapped.choices[0].delta = newDelta;

                  // Clear finish_reason for intermediate chunks or if toolCalls will follow
                  if (i + SSE_MAPPING_CHUNK_SIZE < contentStr.length || toolCalls) {
                    chunkMapped.choices[0].finish_reason = null;
                  }
                  controller.enqueue(encoder.encode(`data: ${JSON.stringify(chunkMapped)}\n\n`));
                  hasEmitted = true;
                }
              } else {
                const chunkMapped = cloneDeep(baseMapped);
                const newDelta = { ...delta };
                delete newDelta.tool_calls;
                newDelta.content = contentStr;
                chunkMapped.choices[0].delta = newDelta;

                if (toolCalls) chunkMapped.choices[0].finish_reason = null;
                controller.enqueue(encoder.encode(`data: ${JSON.stringify(chunkMapped)}\n\n`));
                hasEmitted = true;
              }
            }

            // 2. Process and stream tool_calls if they exist
            if (toolCalls && toolCalls.length > 0) {
              let needsChunking = false;
              for (const tc of toolCalls) {
                if (tc.function?.arguments && tc.function.arguments.length > SSE_MAPPING_CHUNK_SIZE) {
                  needsChunking = true;
                  break;
                }
              }

              if (needsChunking) {
                const toolCallsCopy = cloneDeep(toolCalls);
                let maxLen = 0;
                for (const tc of toolCallsCopy) {
                  if (tc.function?.arguments) {
                    maxLen = Math.max(maxLen, tc.function.arguments.length);
                  }
                }

                for (let i = 0; i < maxLen; i += SSE_MAPPING_CHUNK_SIZE) {
                  const chunkMapped = cloneDeep(baseMapped);
                  const newDelta = !hasEmitted && i === 0 ? { ...delta } : {};
                  if (delta.role && i === 0) newDelta.role = delta.role;
                  if ('content' in newDelta) delete newDelta.content;
                  newDelta.tool_calls = [];
                  chunkMapped.choices[0].delta = newDelta;

                  // Only keep finish_reason on the very last chunk
                  if (i + SSE_MAPPING_CHUNK_SIZE < maxLen) {
                    chunkMapped.choices[0].finish_reason = null;
                  }

                  for (const tc of toolCallsCopy) {
                    const args = tc.function?.arguments || '';
                    if (i < args.length) {
                      const chunkTc = cloneDeep(tc);
                      if (chunkTc.function) {
                        chunkTc.function.arguments = args.slice(i, i + SSE_MAPPING_CHUNK_SIZE);
                      }
                      // Strip metadata on subsequent chunks to conform to OpenAI stream protocol
                      if (i > 0) {
                        delete chunkTc.id;
                        delete chunkTc.type;
                        if (chunkTc.function) delete chunkTc.function.name;
                      }
                      chunkMapped.choices[0].delta.tool_calls.push(chunkTc);
                    }
                  }
                  controller.enqueue(encoder.encode(`data: ${JSON.stringify(chunkMapped)}\n\n`));
                  hasEmitted = true;
                }
              } else {
                const chunkMapped = cloneDeep(baseMapped);
                const newDelta = !hasEmitted ? { ...delta } : {};
                if (delta.role) newDelta.role = delta.role;
                if ('content' in newDelta) delete newDelta.content;
                newDelta.tool_calls = toolCalls;
                chunkMapped.choices[0].delta = newDelta;
                controller.enqueue(encoder.encode(`data: ${JSON.stringify(chunkMapped)}\n\n`));
                hasEmitted = true;
              }
            }

            // 3. Fallback if chunk had no content and no tool_calls (e.g., finish_reason only)
            if (!hasEmitted) {
              controller.enqueue(encoder.encode(`data: ${JSON.stringify(baseMapped)}\n\n`));
            }
          };

          try {
            // Accumulate `data:` lines for one SSE event (SSE allows multiple data
            // lines per event). Flush on a blank line.
            let dataLines: string[] = [];

            const flushEvent = () => {
              if (dataLines.length === 0) return;
              const rawData = dataLines.join('\n');
              dataLines = [];

              const trimmedData = rawData.trim();
              if (trimmedData === '[DONE]') {
                controller.enqueue(encoder.encode('data: [DONE]\n\n'));
                return;
              }

              let parsed: any;
              try {
                parsed = JSON.parse(rawData);
              } catch {
                // Non-JSON / foreign-format event — drop rather than re-emitting raw.
                return;
              }

              const mappedContent = getByPath(parsed, contentPath);

              // Extract tool_calls from the response (Issue #1)
              // Try custom path first, then fall back to standard OpenAI chunk paths
              const mappedToolCalls = toolCallsPath
                ? getByPath(parsed, toolCallsPath)
                : getByPath(parsed, 'choices.0.delta.tool_calls') ?? getByPath(parsed, 'delta.tool_calls');

              const mappedFinishReason = finishReasonPath
                ? getByPath(parsed, finishReasonPath)
                : getByPath(parsed, 'choices.0.finish_reason') ?? getByPath(parsed, 'finish_reason');

              if (mappedContent === undefined && !mappedToolCalls && mappedFinishReason === undefined) {
                // No content/tool_calls/finish_reason — drop (e.g. usage-only chunks).
                return;
              }

              // Build the delta — include content, tool_calls, and finish_reason
              const delta: Record<string, any> = { role: 'assistant' };
              if (mappedContent !== undefined) {
                delta.content = String(mappedContent);
              }
              if (mappedToolCalls) {
                delta.tool_calls = mappedToolCalls;
              }

              const mapped = {
                id: getByPath(parsed, responseMapping.id || 'id') || 'chatcmpl-custom',
                object: 'chat.completion.chunk',
                created: Math.floor(Date.now() / 1000),
                model: 'custom',
                choices: [
                  {
                    index: 0,
                    delta,
                    finish_reason: mappedFinishReason ?? null,
                  },
                ],
              };

              enqueueChunked(mapped);
            };

            // eslint-disable-next-line no-constant-condition
            while (true) {
              const { done, value } = await reader.read();
              if (done) {
                buffer += decoder.decode();
              } else {
                buffer += decoder.decode(value, { stream: true });
              }

              const lines = buffer.split('\n');
              buffer = lines.pop() || '';

              for (const rawLine of lines) {
                const line = rawLine.replace(/\r$/, '');
                const trimmed = line.trim();

                if (trimmed === '') {
                  // Blank line terminates the current event.
                  flushEvent();
                  continue;
                }
                if (trimmed.startsWith(':')) {
                  // SSE comment — drop.
                  continue;
                }

                const dataMatch = trimmed.match(/^data\s*:\s?(.*)$/);
                if (dataMatch) {
                  dataLines.push(dataMatch[1]);
                }
                // Other fields (event:, id:, retry:) carry no payload we need.
              }

              if (done) {
                // Flush a trailing event that lacked a terminating blank line.
                if (buffer) {
                  const line = buffer.replace(/\r$/, '').trim();
                  const dataMatch = line.match(/^data\s*:\s?(.*)$/);
                  if (dataMatch) dataLines.push(dataMatch[1]);
                }
                flushEvent();
                controller.close();
                break;
              }
            }
          } catch (err) {
            controller.error(err);
          }
        },
      });

      return new Response(stream, {
        status: response.status,
        statusText: response.statusText,
        headers: new Headers({
          'content-type': 'text/event-stream',
        }),
      });
    }

    // Handle non-streaming JSON responses
    if (contentType.includes('application/json')) {
      const rawBody = await response.text();
      let body: any;
      try {
        body = JSON.parse(rawBody);
      } catch {
        // Not valid JSON despite the content-type — pass the body through
        return new Response(rawBody, {
          status: response.status,
          statusText: response.statusText,
          headers: new Headers({ 'content-type': contentType }),
        });
      }
      const mappedContent = getByPath(body, contentPath);

      // Extract tool_calls for non-streaming (Issue #1)
      const mappedToolCalls = toolCallsPath
        ? getByPath(body, toolCallsPath)
        : getByPath(body, 'choices.0.message.tool_calls') ?? getByPath(body, 'message.tool_calls');

      const mappedFinishReason = finishReasonPath
        ? getByPath(body, finishReasonPath)
        : getByPath(body, 'choices.0.finish_reason') ?? getByPath(body, 'finish_reason');

      if (mappedContent !== undefined || mappedToolCalls) {
        const message: Record<string, any> = {
          role: getByPath(body, responseMapping.role || '') || 'assistant',
        };
        if (mappedContent !== undefined) {
          message.content = String(mappedContent);
        } else {
          // When only tool_calls, content should be null (OpenAI convention)
          message.content = null;
        }
        if (mappedToolCalls) {
          message.tool_calls = mappedToolCalls;
        }

        const mapped = {
          id: getByPath(body, responseMapping.id || 'id') || 'chatcmpl-custom',
          object: 'chat.completion',
          created: Math.floor(Date.now() / 1000),
          model: 'custom',
          choices: [
            {
              index: 0,
              message,
              finish_reason: mappedFinishReason ?? (mappedToolCalls ? 'tool_calls' : 'stop'),
            },
          ],
          usage: body.usage ?? { prompt_tokens: 0, completion_tokens: 0, total_tokens: 0 },
        };

        return new Response(JSON.stringify(mapped), {
          status: response.status,
          statusText: response.statusText,
          headers: new Headers({
            'content-type': 'application/json',
          }),
        });
      }

      // Mapping found nothing — reconstruct the response because the body
      // was already consumed above
      return new Response(JSON.stringify(body), {
        status: response.status,
        statusText: response.statusText,
        headers: new Headers({
          'content-type': 'application/json',
        }),
      });
    }

    return response;
  };
}

/**
 * Check if a text string is a keepalive marker.
 */
function isKeepAlive(text: string): boolean {
  return typeof text === 'string' && text.startsWith(KEEPALIVE_PREFIX);
}

/* bindTools empty-tool-properties fix is now integrated into patchRunnableForSanitization below */

/**
 * Strip Gemini __thought__ suffixes from tool_calls on an AIMessage (mutates in place).
 */
function sanitizeAIMessageToolCalls(msg: any): void {
  if (!msg) return;
  if (msg.tool_calls) {
    for (const tc of msg.tool_calls) {
      tc.id = stripGeminiThoughtSuffix(tc.id);
    }
  }
  if (msg.tool_call_chunks) {
    for (const tc of msg.tool_call_chunks) {
      tc.id = stripGeminiThoughtSuffix(tc.id);
    }
  }
}

/**
 * Estimate token count for a message.
 * ⚠️ ROUGH HEURISTIC ONLY — not suitable for hard budget enforcement.
 * Uses CJK chars ~1 token each, Latin ~4 chars/token. Does NOT account
 * for tool_calls payload, role/structure overhead, or emoji (which cost
 * 2-3 tokens each but are counted as 1 here). Consider js-tiktoken if
 * accuracy is critical.
 */
function estimateTokens(msg: any): number {
  let text = '';
  if (typeof msg.content === 'string') text = msg.content;
  else if (Array.isArray(msg.content)) {
    text = msg.content.map((c: any) => c.text || '').join('');
  }
  if (!text) return 0;
  let tokens = 0;
  for (const char of text) {
    // CJK Unified Ideographs and related blocks are ~1 token each
    tokens += char.charCodeAt(0) > 0x2e80 ? 1 : 0.25;
  }
  return Math.ceil(tokens);
}

/**
 * Truncate messages to fit within a token budget.
 *
 * IMPORTANT: Messages are removed in complete "conversation turns" to
 * preserve the strict AI↔Tool message pairing required by LangChain and
 * the OpenAI API. A turn is:
 *   - A single user/system message, OR
 *   - An AI message with tool_calls + all its corresponding ToolMessages
 *
 * The system message (if first) and the most recent messages are preserved.
 */
function truncateMessages(messages: any[], maxTokens: number): any[] {
  if (!Array.isArray(messages) || messages.length === 0) return messages;

  let totalTokens = messages.reduce((sum, m) => sum + estimateTokens(m), 0);
  if (totalTokens <= maxTokens) return messages;

  const result = [...messages];
  // Preserve system message separately
  const systemMsg =
    result.length > 0 && (result[0].getType?.() === 'system' || result[0]._getType?.() === 'system')
      ? result.shift()
      : null;
  const systemTokenCost = systemMsg ? estimateTokens(systemMsg) : 0;
  totalTokens -= systemTokenCost;

  // Group remaining messages into conversation turns.
  // A turn is: [user/human] or [ai + tool*] (ai message with its tool responses)
  const turns: any[][] = [];
  let i = 0;
  while (i < result.length) {
    const msg = result[i];
    const msgType = msg.getType?.() ?? msg._getType?.() ?? msg.role ?? '';

    if (msgType === 'ai' || msgType === 'assistant') {
      // Group this AI message with all following tool messages
      const turn = [msg];
      i++;
      while (i < result.length) {
        const next = result[i];
        const nextType = next.getType?.() ?? next._getType?.() ?? next.role ?? '';
        if (nextType === 'tool') {
          turn.push(next);
          i++;
        } else {
          break;
        }
      }
      turns.push(turn);
    } else {
      turns.push([msg]);
      i++;
    }
  }

  // Remove oldest turns until under budget (including system message cost),
  // keeping at least the last turn
  const effectiveBudget = maxTokens - systemTokenCost;
  while (turns.length > 1 && totalTokens > effectiveBudget) {
    const removed = turns.shift() ?? [];
    for (const msg of removed) {
      totalTokens -= estimateTokens(msg);
    }
  }

  // Flatten turns back into a flat message array
  const truncated = turns.flat();
  if (systemMsg) {
    truncated.unshift(systemMsg);
  }
  return truncated;
}

/**
 * Patch a runnable (model or bound model) so that `invoke` and `stream`
 * sanitize tool call IDs on every AIMessage output.
 * Also patches `bindTools` and `bind` so that derived runnables inherit
 * the sanitization — this is critical because langgraph calls
 * `model.bindTools(tools)` and then uses the BOUND model.
 */
function patchRunnableForSanitization(
  runnable: any,
  options?: {
    enableTokenTruncation?: boolean;
    maxContextTokens?: number;
    enableToolRetry?: boolean;
    maxToolRetries?: number;
    streamKeepAlive?: boolean;
    keepAliveIntervalMs?: number;
    streamChunkSize?: number;
  },
): any {
  if (!runnable || runnable.__toolCallSanitized) return runnable;
  runnable.__toolCallSanitized = true;

  // Patch invoke — covers non-streaming and internal streaming-via-invoke
  const origInvoke = runnable.invoke?.bind(runnable);
  if (origInvoke) {
    runnable.invoke = async function (...args: any[]) {
      let messages = args[0];
      if (options?.enableTokenTruncation && options.maxContextTokens) {
        messages = truncateMessages(messages, options.maxContextTokens);
        args[0] = messages;
      }

      const retries = options?.enableToolRetry ? options.maxToolRetries || 1 : 0;
      for (let attempt = 0; attempt <= retries; attempt++) {
        try {
          const result = await origInvoke(...args);
          sanitizeAIMessageToolCalls(result);
          return result;
        } catch (e: any) {
          // Only retry model-output failures (e.g. invalid tool-call JSON).
          // HTTP/transport errors (auth, rate limit, network, aborts) would
          // fail again and the corrective prompt below does not apply.
          const isTransportError =
            typeof e?.status === 'number' ||
            e?.name === 'AbortError' ||
            e?.code === 'ECONNABORTED' ||
            e?.code === 'ETIMEDOUT';
          if (attempt === retries || isTransportError) {
            throw e;
          }
          if (Array.isArray(messages)) {
            try {
              const HumanMessage = requireFromApp('@langchain/core/messages').HumanMessage;
              // Clone to avoid mutating the caller's original message array
              messages = [
                ...messages,
                new HumanMessage(
                  `Your previous response was invalid. Please correct it and try again. Error: ${e.message}`,
                ),
              ];
              args[0] = messages;
            } catch (err) {
              // Ignore errors if HumanMessage fails to import
            }
          }
        }
      }
    };
  }

  // Patch stream — covers streaming path.
  // NOTE: Tool-call retry is NOT applied here because stream() returns an
  // async iterable — errors from bad tool-call JSON surface during iteration
  // (downstream in ai-employee.ts), not during stream creation. Retry at this
  // level would require buffering the entire stream, which defeats the purpose.
  // Auto tool-call retry is only effective via the invoke() path.
  const origStream = runnable.stream?.bind(runnable);
  if (origStream) {
    runnable.stream = async function (...args: any[]) {
      let messages = args[0];
      if (options?.enableTokenTruncation && options.maxContextTokens) {
        messages = truncateMessages(messages, options.maxContextTokens);
        args[0] = messages;
      }

      const iter = await origStream(...args);
      const safeIter = streamWithKeepAliveAndChunking(iter, {
        keepAlive: options?.streamKeepAlive ?? false,
        intervalMs: normalizePositiveNumber(options?.keepAliveIntervalMs, DEFAULT_KEEPALIVE_INTERVAL_MS),
        chunkSize: normalizePositiveNumber(options?.streamChunkSize, DEFAULT_STREAM_CHUNK_SIZE),
      });

      return (async function* () {
        for await (const value of safeIter) {
          if (value) {
            sanitizeAIMessageToolCalls(value);
          }
          yield value;
        }
      })();
    };
  }

  // Patch bindTools — combined wrapper that fixes empty tool properties (Gemini
  // rejects `properties: {}`) AND propagates sanitization to derived runnables.
  const PLACEHOLDER_PROP = {
    _placeholder: { type: 'string', description: 'No parameters required' },
  };

  /**
   * Recursively fix empty properties in a JSON Schema-like object.
   * Handles: top-level properties, function.parameters.properties,
   * and nested anyOf/oneOf/allOf schemas.
   */
  function fixPropertiesInSchema(schema: any): void {
    if (!schema || typeof schema !== 'object') return;
    if (schema.properties && typeof schema.properties === 'object' && Object.keys(schema.properties).length === 0) {
      schema.properties = { ...PLACEHOLDER_PROP };
    }
    for (const key of ['anyOf', 'oneOf', 'allOf']) {
      if (Array.isArray(schema[key])) {
        schema[key].forEach((sub: any) => fixPropertiesInSchema(sub));
      }
    }
  }

  const origBindTools = runnable.bindTools?.bind(runnable);
  if (origBindTools) {
    runnable.bindTools = function (tools: any[], kwargs?: any) {
      // Phase 1: Pre-conversion fix for raw JSON Schema tool definitions
      const fixedTools = tools.map((tool: any) => {
        if (!tool || typeof tool !== 'object') return tool;
        // Skip Zod schema tools — they'll be handled post-conversion
        if (typeof tool.schema?.safeParse === 'function') return tool;
        const schema = tool.schema;
        if (schema && typeof schema === 'object' && !schema.safeParse) {
          const props = schema.properties;
          if (props && typeof props === 'object' && Object.keys(props).length === 0) {
            return { ...tool, schema: { ...schema, properties: { ...PLACEHOLDER_PROP } } };
          }
        }
        const funcParams = tool.function?.parameters;
        if (funcParams?.properties) {
          if (typeof funcParams.properties === 'object' && Object.keys(funcParams.properties).length === 0) {
            return {
              ...tool,
              function: { ...tool.function, parameters: { ...funcParams, properties: { ...PLACEHOLDER_PROP } } },
            };
          }
        }
        return tool;
      });
      const bound = origBindTools(fixedTools, kwargs);
      // Phase 2: Post-conversion fix — patch the converted tools in the result
      try {
        const config = bound?.kwargs ?? bound?.defaultOptions;
        if (config?.tools && Array.isArray(config.tools)) {
          for (const tool of config.tools) {
            if (tool?.function?.parameters) fixPropertiesInSchema(tool.function.parameters);
            if (tool?.parameters) fixPropertiesInSchema(tool.parameters);
          }
        }
      } catch {
        /* don't break tool binding if post-fix inspection fails */
      }
      return patchRunnableForSanitization(bound, options);
    };
  }

  // Patch bind — bindTools internally calls bind(), some runnables use it directly
  const origBind = runnable.bind?.bind(runnable);
  if (origBind) {
    runnable.bind = function (...args: any[]) {
      const bound = origBind(...args);
      return patchRunnableForSanitization(bound, options);
    };
  }

  return runnable;
}

export class CustomLLMProvider extends LLMProvider {
  get baseURL() {
    return null;
  }

  private get requestConfig() {
    return safeParseJSON(this.serviceOptions?.requestConfig, 'requestConfig');
  }

  private get responseConfig() {
    return safeParseJSON(this.serviceOptions?.responseConfig, 'responseConfig');
  }

  createModel() {
    const { apiKey, disableStream, timeout, streamKeepAlive, keepAliveIntervalMs, enableReasoning } =
      this.serviceOptions || {};
    // baseURL comes from core's options.baseURL field
    const baseURL = this.serviceOptions?.baseURL;
    const {
      responseFormat,
      jsonSchemaDefinition,
      enableTokenTruncation,
      maxContextTokens,
      enableToolRetry,
      maxToolRetries,
    } = this.modelOptions || {};
    const reqConfig = this.requestConfig;
    const resConfig = this.responseConfig;

    const modelKwargs: Record<string, any> = {
      ...(reqConfig.modelKwargs || {}),
    };

    // NOTE: response_format is NOT set in modelKwargs here.
    // It is applied lazily in prepareChain() so that withStructuredOutput()
    // (from AI Employee context) takes priority when both are configured.
    // See: prepareChain() override below.

    if (reqConfig.extraBody && typeof reqConfig.extraBody === 'object') {
      Object.assign(modelKwargs, reqConfig.extraBody);
    }

    // Issue #4: Use ReasoningChatOpenAI when enableReasoning is set.
    // This ensures reasoning_content is preserved and patched back into
    // assistant messages during tool call round-trips (required by DeepSeek-R1, etc.)
    const ChatClass = enableReasoning ? createReasoningChatClass() : getChatOpenAI();
    // Exclude plugin-specific fields from the LangChain config spread.
    // Fields like maxToolRetries, enableToolRetry, enableVision etc. are not
    // LangChain constructor params and could cause collisions (e.g. maxRetries).
    const {
      responseFormat: _rf,
      jsonSchemaDefinition: _jsd,
      enableTokenTruncation: _ett,
      maxContextTokens: _mct,
      enableToolRetry: _etr,
      maxToolRetries: _mtr,
      enableVision: _ev,
      ...langchainModelOptions
    } = this.modelOptions || {};
    const config: Record<string, any> = {
      apiKey,
      ...langchainModelOptions,
      modelKwargs,
      configuration: {
        baseURL,
      },
      verbose: false,
    };

    // Disable streaming for models with long thinking phases
    // that return empty stream values causing processing to terminate
    if (disableStream) {
      config.streaming = false;
    }

    // Apply a long request timeout for slow-thinking models. The OpenAI SDK
    // default is 10 minutes; custom providers can exceed that before headers.
    const timeoutMs = timeout && Number(timeout) > 0 ? Number(timeout) : 0;
    const effectiveTimeoutMs = Math.max(timeoutMs, DEFAULT_REQUEST_TIMEOUT_MS);
    config.timeout = effectiveTimeoutMs;
    config.configuration.timeout = effectiveTimeoutMs;

    // Apply extra headers
    if (reqConfig.extraHeaders && typeof reqConfig.extraHeaders === 'object') {
      config.configuration.defaultHeaders = reqConfig.extraHeaders;
    }

    // Apply response mapping via custom fetch — pass timeout for fetch-level protection (Issue #7)
    if (resConfig.responseMapping) {
      config.configuration.fetch = createMappingFetch(resConfig.responseMapping, reqConfig.retry);
    }

    let model = new ChatClass(config);

    // Sanitize Gemini's __thought__<base64> suffixes in tool call IDs.
    // Patches invoke/stream/bindTools/bind at the public API level so that
    // ALL code paths (including langgraph's internal model calls via
    // RunnableBinding after bindTools) return clean IDs.
    model = patchRunnableForSanitization(model, {
      enableTokenTruncation,
      maxContextTokens,
      enableToolRetry,
      maxToolRetries,
      streamKeepAlive: streamKeepAlive !== false,
      keepAliveIntervalMs: normalizePositiveNumber(keepAliveIntervalMs, DEFAULT_KEEPALIVE_INTERVAL_MS),
      streamChunkSize: normalizePositiveNumber(this.serviceOptions?.streamChunkSize, DEFAULT_STREAM_CHUNK_SIZE),
    });

    return model;
  }

  /**
   * Override listModels to support LiteLLM/custom proxies whose model-listing
   * endpoint, auth scheme, or response shape deviates from OpenAI's GET /models.
   *
   * Configured via the `modelsConfig` service option:
   * {
   *   "path": "models",        // endpoint appended to baseURL
   *   "auth": "bearer",        // "bearer" | "api-key" | "none"
   *   "headers": {},           // extra headers
   *   "dataPath": "data",      // dot-path to the model array in the response
   *   "idPath": "id"           // dot-path to the id within each item
   * }
   */
  async listModels(): Promise<{ models?: { id: string }[]; code?: number; errMsg?: string }> {
    const options = this.serviceOptions || {};
    const modelsConfig = safeParseJSON(options.modelsConfig, 'modelsConfig');
    const pathName = typeof modelsConfig.path === 'string' && modelsConfig.path ? modelsConfig.path : 'models';
    const auth = modelsConfig.auth === 'api-key' || modelsConfig.auth === 'none' ? modelsConfig.auth : 'bearer';
    const apiKey = options.apiKey;

    let url: string;
    try {
      url = this.buildRequestURL(pathName);
    } catch (e) {
      return { code: 400, errMsg: e instanceof Error ? e.message : String(e) };
    }

    if (auth !== 'none' && !apiKey) {
      return { code: 400, errMsg: 'API Key required' };
    }

    const headers: Record<string, string> = {
      ...(modelsConfig.headers && typeof modelsConfig.headers === 'object' ? modelsConfig.headers : {}),
    };
    if (auth === 'bearer') {
      headers.Authorization = `Bearer ${apiKey}`;
    } else if (auth === 'api-key') {
      headers['x-api-key'] = String(apiKey);
    }

    try {
      const response = await axios.get(url, { headers, timeout: 30_000 });
      const body = response.data;
      const rawItems = getByPath(body, modelsConfig.dataPath || 'data');

      if (rawItems == null) {
        return { models: [] };
      }

      const items = Array.isArray(rawItems) ? rawItems : [];
      const models = items.map((item: any) => {
        if (modelsConfig.idPath) {
          return { id: String(getByPath(item, modelsConfig.idPath) ?? '') };
        }
        if (typeof item === 'string') {
          return { id: item };
        }
        return { id: String(item?.id ?? item?.model ?? '') };
      });

      return { models };
    } catch (e) {
      const err = e as { response?: { status?: number; data?: any; statusText?: string }; message?: string };
      const status = err.response?.status || 500;
      const data = err.response?.data;
      const errorMsg =
        data?.error?.message ??
        data?.message ??
        (typeof data?.error === 'string' ? data.error : undefined) ??
        (typeof data === 'string' ? data : undefined) ??
        err.response?.statusText ??
        err.message ??
        'Failed to list models';
      return { code: status, errMsg: errorMsg };
    }
  }

  /**
   * Override prepareChain to apply response_format lazily.
   *
   * When `context.structuredOutput` is present (set by AI Employee),
   * the base class's `withStructuredOutput()` takes full priority and
   * response_format from model settings is NOT applied — avoiding the
   * double-schema conflict that some providers reject.
   *
   * When structuredOutput is NOT present, the user's response_format
   * (json_object, json_schema) is applied via a bound model kwargs.
   */
  prepareChain(context: any) {
    // Let the base class handle tools + structuredOutput
    let chain = super.prepareChain(context);

    // Only apply response_format when structuredOutput is NOT active
    if (!context?.structuredOutput) {
      const { responseFormat, jsonSchemaDefinition } = this.modelOptions || {};

      if (responseFormat === 'json_schema' && jsonSchemaDefinition) {
        try {
          const parsedSchema = JSON.parse(jsonSchemaDefinition);
          chain = chain.bind({
            response_format: {
              type: 'json_schema',
              json_schema: {
                name: 'custom_response',
                strict: true,
                schema: parsedSchema,
              },
            },
          });
        } catch (e) {
          console.warn('[CustomLLM] Failed to parse jsonSchemaDefinition', e);
        }
      } else if (responseFormat && responseFormat !== 'text') {
        // json_object mode — don't send { type: 'text' } as it breaks non-OpenAI providers
        chain = chain.bind({
          response_format: { type: responseFormat },
        });
      }
    }

    return chain;
  }

  parseResponseChunk(chunk: any): string | null {
    const resConfig = this.responseConfig;
    const text = extractTextContent(chunk, resConfig.contentPath);

    // Return keepalive prefix as-is so protocol.content() emits SSE event.
    // The zero-width spaces are invisible in the client UI but keep
    // proxy/gateway connections alive during long model thinking phases.
    if (isKeepAlive(text)) {
      return KEEPALIVE_PREFIX;
    }

    return stripToolCallTags(text);
  }

  parseResponseMessage(message: Model) {
    const { content: rawContent, messageId, metadata, role, toolCalls, attachments, workContext } = message;
    const content: Record<string, any> = {
      ...(rawContent ?? {}),
      messageId,
      metadata,
      attachments,
      workContext,
    };

    if (toolCalls) {
      content.tool_calls = Array.isArray(toolCalls)
        ? toolCalls.map((tc: any) => ({ ...tc, id: stripGeminiThoughtSuffix(tc.id) }))
        : toolCalls;
    }

    if (Array.isArray(content.content)) {
      const textBlocks = content.content.filter((block: any) => block.type === 'text');
      content.content = textBlocks.map((block: any) => block.text).join('') || '';
    }

    if (typeof content.content === 'string') {
      // Issue #2: Strip keepalive markers safely — use simple replaceAll instead of
      // greedy regex that could accidentally eat real content between two markers.
      content.content = content.content.replaceAll(KEEPALIVE_PREFIX, '');
      content.content = stripToolCallTags(content.content);
    }

    // Clean internal keepalive flag from persisted additional_kwargs
    if (content.metadata?.additional_kwargs?.__keepalive !== undefined) {
      const { __keepalive, ...cleanKwargs } = content.metadata.additional_kwargs;
      content.metadata = { ...content.metadata, additional_kwargs: cleanKwargs };
    }

    return {
      key: messageId,
      content,
      role,
    };
  }

  parseReasoningContent(chunk: any) {
    const resConfig = this.responseConfig;
    const reasoningKey = resConfig.reasoningKey || 'reasoning_content';

    // Check multiple paths — different providers/chunk formats nest reasoning differently
    const reasoning = chunk?.additional_kwargs?.[reasoningKey] ?? chunk?.kwargs?.additional_kwargs?.[reasoningKey];

    if (reasoning && typeof reasoning === 'string') {
      return { status: 'streaming', content: reasoning };
    }
    return null as any;
  }

  /**
   * Extract response metadata from LLM output for post-save enrichment.
   * Sanitizes overly long message IDs from Gemini or other providers.
   */
  parseResponseMetadata(output: any): any {
    try {
      const generation = output?.generations?.[0]?.[0];
      if (!generation) return [null, null];

      const message = generation.message;
      let id = message?.id;
      if (!id) return [null, null];

      // Sanitize overly long IDs (Gemini can return very long chatcmpl-xxx or run-xxx IDs)
      if (typeof id === 'string' && id.length > 128) {
        id = id.substring(0, 128);
      }

      const metadata: Record<string, any> = {};
      if (message?.response_metadata) {
        metadata.finish_reason = message.response_metadata.finish_reason;
        metadata.system_fingerprint = message.response_metadata.system_fingerprint;
      }
      if (message?.usage_metadata) {
        metadata.usage_metadata = message.usage_metadata;
      }

      return Object.keys(metadata).length > 0 ? [id, metadata] : [null, null];
    } catch {
      return [null, null];
    }
  }

  parseResponseError(err: any) {
    return err?.message ?? 'Unexpected LLM service error';
  }

  /**
   * Self-contained file reading that correctly handles the APP_PUBLIC_PATH prefix.
   *
   * plugin-ai's encodeLocalFile does path.join(cwd, url) without stripping
   * APP_PUBLIC_PATH, so when the app is deployed under a sub-path (e.g. /my-app)
   * the resolved path becomes '{cwd}/my-app/storage/uploads/…' which does not exist.
   * We cannot fix that in plugin-ai (core), so we re-implement file reading here
   * with the prefix stripped before the cwd join.
   */
  /**
   * Reads the attachment and returns its base64-encoded content plus, when the
   * file lives on the local filesystem, the resolved absolute path so callers
   * can hand that path directly to tools like DocPixie and avoid a second
   * write-to-disk round-trip.
   */
  private async readFileData(ctx: Context, attachment: any): Promise<{ base64: string; absPath?: string }> {
    const fileManager = this.app.pm.get('file-manager') as any;
    const rawUrl: string = await fileManager.getFileURL(attachment);
    const url = decodeURIComponent(rawUrl);

    if (url.startsWith('http://') || url.startsWith('https://')) {
      const referer = ctx.get('referer') || '';
      const ua = ctx.get('user-agent') || '';
      const response = await axios.get(url, {
        responseType: 'arraybuffer',
        timeout: 30_000,
        headers: { referer, 'User-Agent': ua },
      });
      return { base64: Buffer.from(response.data).toString('base64') };
    }

    // Internal API stream URL (e.g. s3-private-storage proxy) — read directly via fileManager
    if (url.includes('/api/attachments:stream')) {
      const rawStorageId =
        attachment.storageId ||
        (typeof attachment.get === 'function' ? attachment.get('storageId') : attachment.storageId);
      let matchedKey = null;
      if (rawStorageId) {
        const strId = String(rawStorageId);
        for (const key of fileManager.storagesCache.keys()) {
          if (String(key) === strId) {
            matchedKey = key;
            break;
          }
        }
      }

      const attachmentObj = typeof attachment.toJSON === 'function' ? attachment.toJSON() : { ...attachment };
      if (matchedKey !== null) {
        attachmentObj.storageId = matchedKey;
      }

      const { stream } = await fileManager.getFileStream(attachmentObj);
      const chunks: Buffer[] = [];
      for await (const chunk of stream) {
        chunks.push(typeof chunk === 'string' ? Buffer.from(chunk) : chunk);
      }
      return { base64: Buffer.concat(chunks).toString('base64') };
    }

    // Local file — strip APP_PUBLIC_PATH prefix before joining with cwd
    let localPath = url;
    const appPublicPath = (process.env.APP_PUBLIC_PATH || '/').replace(/\/+$/, '');
    if (appPublicPath && localPath.startsWith(appPublicPath + '/')) {
      localPath = localPath.slice(appPublicPath.length);
    }

    // Resolve and guard against path traversal
    const storageRoot = path.resolve(process.cwd());
    const absPath = path.resolve(storageRoot, localPath.replace(/^\//, ''));
    if (!absPath.startsWith(storageRoot + path.sep) && absPath !== storageRoot) {
      throw new Error(`Attachment path escapes storage root: ${localPath}`);
    }

    const data = await fs.readFile(absPath);
    // Return absPath so parseAttachment can pass it directly to DocPixie
    return { base64: Buffer.from(data).toString('base64'), absPath };
  }

  /**
   * Override parseAttachment to convert all attachments into formats that
   * generic OpenAI-compatible endpoints actually support:
   *
   *   - Images     → image_url block with base64 data URI (vision models)
   *   - Text files → text block with decoded UTF-8 content
   *   - Binary     → text block with base64 data URI (multi-modal or fallback)
   *
   * The base-class implementation returns a LangChain ContentBlock.Multimodal.File
   * (`type: 'file'`) for non-image attachments. LangChain serialises this as the
   * newer OpenAI Files API format which most custom/local endpoints do NOT understand,
   * causing file content to be silently dropped.
   *
   * This method is entirely self-contained — it does not call super — so it is
   * safe to use without modifying plugin-ai core.
   */
  /**
   * Try to extract text from an attachment using DocPixie (if available and
   * the file type is supported). Returns null if DocPixie is unavailable,
   * not ready, or the file type is not supported.
   */
  /**
   * Check whether the DocPixie skill (`docpixie.query.document`) is configured
   * on the AI employee that initiated this request.
   *
   * Reads `ctx.action.params.values.aiEmployee` (the employee username set by the
   * `sendMessages` action handler), then looks up the employee's `skillSettings`
   * from DB. Result is cached on `ctx.state._docPixieActive` for the request lifetime.
   */
  private async hasDocPixieSkill(ctx: Context): Promise<boolean> {
    if (ctx.state._docPixieActive !== undefined) return ctx.state._docPixieActive as boolean;

    try {
      // Issue #6: Try multiple sources for the AI employee username.
      // The field may be placed differently depending on whether the request
      // comes from sendMessages action, workflow invoke, or direct API call.
      const employeeUsername =
        ctx.action?.params?.values?.aiEmployee ?? ctx.action?.params?.aiEmployee ?? ctx.state?.currentAiEmployee;

      if (!employeeUsername) {
        ctx.state._docPixieActive = false;
        return false;
      }

      const employee = await ctx.db.getRepository('aiEmployees').findOne({
        filter: { username: String(employeeUsername) },
        fields: ['skillSettings'],
      });

      const skills: Array<{ name: string }> = (employee?.get?.('skillSettings') as any)?.skills ?? [];
      const has = skills.some((s) => s.name === 'docpixie.query.document');
      ctx.state._docPixieActive = has;
      return has;
    } catch {
      ctx.state._docPixieActive = false;
      return false;
    }
  }

  /**
   * Run the full DocPixie ingestion pipeline (extract pages → generate summary → index).
   * Returns a formatted `<processed_document>` context block the LLM can use immediately,
   * plus a clear instruction to call the RAG tool with the returned documentId for details.
   *
   * Prefers passing `absPath` directly for local-storage files to avoid a second
   * write-to-disk round-trip. Falls back to Buffer for remote / S3 files.
   *
   * Returns null if DocPixie is unavailable, not configured, or processing fails.
   */
  private async tryDocPixieFullProcess(
    fileData: { base64: string; absPath?: string },
    filename: string,
    ctx: Context,
  ): Promise<string | null> {
    try {
      const docpixie = this.app.pm.get('docpixie') as any;
      if (!docpixie?.service?.isReady?.()) return null;

      const userId: number | undefined = ctx.state?.currentUser?.id;
      let result: { documentId: number; summary: string; pageCount: number };

      if (fileData.absPath) {
        result = await docpixie.service.processDocumentFromPath(fileData.absPath, filename, { userId });
      } else {
        const buffer = Buffer.from(fileData.base64, 'base64');
        result = await docpixie.service.processDocumentFromBuffer(buffer, filename, { userId });
      }

      const { documentId, summary, pageCount } = result;
      const summaryText = summary?.trim() || 'No summary available.';

      return (
        `<processed_document id="${documentId}" filename="${filename}" pages="${pageCount}">\n` +
        `<summary>\n${summaryText}\n</summary>\n` +
        `<rag_instruction>This document is fully indexed. ` +
        `Call docpixie.query.document with documentId=${documentId} to retrieve specific details.</rag_instruction>\n` +
        `</processed_document>`
      );
    } catch {
      return null;
    }
  }

  /**
   * Try to extract text from an attachment using DocPixie (transient — no DB indexing).
   * When `absPath` is provided (local-storage file), DocPixie reads the file
   * directly — no Buffer decode/re-encode or extra temp-file write.
   * Falls back to `extractTextFromBuffer` for remote/S3 files.
   * Returns null if DocPixie is unavailable, not ready, or file type unsupported.
   */
  private async tryDocPixieExtract(
    fileData: { base64: string; absPath?: string },
    filename: string,
  ): Promise<string | null> {
    try {
      const docpixie = this.app.pm.get('docpixie') as any;
      if (!docpixie?.service) return null;
      let text: string;
      if (fileData.absPath) {
        text = await docpixie.service.extractTextFromPath(fileData.absPath, filename);
      } else {
        const buffer = Buffer.from(fileData.base64, 'base64');
        text = await docpixie.service.extractTextFromBuffer(buffer, filename);
      }
      return text || null;
    } catch {
      return null;
    }
  }

  async parseAttachment(ctx: Context, attachment: any) {
    const mimetype: string = attachment.mimetype || 'application/octet-stream';
    // basename only — raw filenames are interpolated into prompt XML below
    const filename: string = path.basename(String(attachment.filename || attachment.name || 'file'));
    const fileData = await this.readFileData(ctx, attachment);
    const { base64: data } = fileData;

    const { enableVision } = this.modelOptions || {};
    const isImage = mimetype.startsWith('image/');
    const isPdf = mimetype === 'application/pdf';
    const isDocPixieSupported = isPdf || isImage;

    // ── Early exit: images with vision disabled → skip all image processing ──
    // When enableVision is off, the model cannot process images at all,
    // so DocPixie OCR is also skipped (no point extracting text for a
    // model that can't contextualize it). PDFs remain unaffected.
    if (isImage && !enableVision) {
      return {
        placement: 'contentBlocks',
        content: {
          type: 'text',
          text: `<attachment filename="${filename}" type="${mimetype}">[Image: ${filename} — omitted because native Vision is disabled]</attachment>`,
        },
      };
    }

    // ── Path A: DocPixie skill active → full ingestion pipeline ──────────────
    // Runs processDocument (extract pages + generate summary + DB index) so the
    // LLM gets a rich summary + documentId it can pass to the RAG tool for specifics.
    if (isDocPixieSupported && (await this.hasDocPixieSkill(ctx))) {
      const contextBlock = await this.tryDocPixieFullProcess(fileData, filename, ctx);
      if (contextBlock) {
        return {
          placement: 'contentBlocks',
          content: { type: 'text', text: contextBlock },
        };
      }
      // DocPixie not configured / failed → fall through to Path B
    }

    // ── Path B: DocPixie skill absent → transient extraction (no DB) ─────────
    if (isPdf) {
      const extracted = await this.tryDocPixieExtract(fileData, filename);
      if (extracted) {
        return {
          placement: 'contentBlocks',
          content: {
            type: 'text',
            text: `<attachment filename="${filename}" type="${mimetype}">\n${extracted}\n</attachment>`,
          },
        };
      }
      // DocPixie unavailable — fall through to base64 data-URI
    }

    // ── Path C: Image with vision enabled ────────────────────────────────────
    if (isImage) {
      // Try DocPixie OCR first (e.g. scanned documents as images)
      const extracted = await this.tryDocPixieExtract(fileData, filename);
      if (extracted) {
        return {
          placement: 'contentBlocks',
          content: {
            type: 'text',
            text: `<attachment filename="${filename}" type="${mimetype}">\n${extracted}\n</attachment>`,
          },
        };
      }
      // Final fallback — send as image_url for vision-capable models
      return {
        placement: 'contentBlocks',
        content: {
          type: 'image_url',
          image_url: { url: `data:${mimetype};base64,${data}` },
        },
      };
    }

    let textContent: string;
    if (isTextMimetype(mimetype)) {
      // Decode to readable UTF-8 so the model can actually read the content
      const decoded = Buffer.from(data, 'base64').toString('utf-8');
      textContent = `<attachment filename="${filename}" type="${mimetype}">\n${decoded}\n</attachment>`;
    } else {
      // Binary non-image: embed as data-URI; multi-modal models may process it,
      // text-only models at minimum see the filename and type
      textContent = `<attachment filename="${filename}" type="${mimetype}">\ndata:${mimetype};base64,${data}\n</attachment>`;
    }

    return {
      placement: 'contentBlocks',
      content: { type: 'text', text: textContent },
    };
  }
}

export class CustomEmbeddingProvider extends EmbeddingProvider {
  protected getDefaultUrl(): string {
    return '';
  }

  createEmbedding(): EmbeddingsInterface {
    const { OpenAIEmbeddings } = requireFromApp('@langchain/openai');
    return new OpenAIEmbeddings({
      apiKey: this.apiKey,
      configuration: {
        baseURL: this.baseURL,
      },
      model: this.model,
    });
  }
}

export const customLLMProviderOptions: LLMProviderMeta = {
  title: 'Custom LLM (OpenAI Compatible)',
  provider: CustomLLMProvider,
  embedding: CustomEmbeddingProvider as any,
  supportedModel: [SupportedModel.LLM, SupportedModel.EMBEDDING] as any,
};