/**
 * Anthropic Messages-API streaming provider.
 *
 * Different from the existing Claude harness: that one uses the Claude Agent
 * SDK + subscription OAuth, this one talks directly to `api.anthropic.com/v1/messages`
 * with a pay-per-token API key. Lets users bring their own Anthropic credentials
 * through the pi flow instead of (or alongside) the subscription path.
 *
 * Wire shape: SSE with typed events — `message_start`, `content_block_start`,
 * `content_block_delta`, `content_block_stop`, `message_delta`, `message_stop`,
 * `ping`, `error`. Each event's `data:` JSON carries a `type` field matching
 * the event name, so we ignore the SSE `event:` line and route off the JSON.
 */
import { log } from '../../../../shared/logger.js';
import type {
  PiStreamRequest,
  PiStreamEvent,
  PiMessage,
  PiContentBlock,
  PiStopReason,
  PiUsage,
} from './types.js';
import { fetchWithRetry, readWithIdleTimeout } from './retry.js';
import { classifyPiError, classifyPiNetworkError } from './humanize-error.js';

/* ── SSE parser (shares the LF/CRLF-tolerant pattern from the other providers) ── */

async function* parseSse(res: Response): AsyncIterable<any> {
  if (!res.body) return;
  const reader = res.body.getReader();
  const decoder = new TextDecoder();
  let buffer = '';
  try {
    while (true) {
      const { value, done } = await readWithIdleTimeout(reader, 'Anthropic');
      if (done) break;
      buffer += decoder.decode(value, { stream: true });
      let idx;
      while (
        (idx = (() => {
          const a = buffer.indexOf('\n\n');
          const b = buffer.indexOf('\r\n\r\n');
          if (a < 0) return b;
          if (b < 0) return a;
          return Math.min(a, b);
        })()) !== -1
      ) {
        const isCrlf = buffer.slice(idx, idx + 4) === '\r\n\r\n';
        const raw = buffer.slice(0, idx);
        buffer = buffer.slice(idx + (isCrlf ? 4 : 2));
        const parsed = parseEvent(raw);
        if (parsed !== undefined) yield parsed;
      }
    }
    buffer += decoder.decode();
    if (buffer.trim()) {
      const parsed = parseEvent(buffer);
      if (parsed !== undefined) yield parsed;
    }
  } finally {
    try { reader.releaseLock(); } catch {}
  }
}

function parseEvent(raw: string): any | undefined {
  const lines = raw.split(/\r?\n/);
  const dataLines = lines
    .filter((l) => l.startsWith('data:'))
    .map((l) => l.slice(5).trimStart());
  if (!dataLines.length) return undefined;
  const data = dataLines.join('\n');
  if (!data) return undefined;
  try { return JSON.parse(data); } catch { return undefined; }
}

/* ── Message conversion (pi → Anthropic) ── */

function toAnthropicContent(blocks: PiContentBlock[]): any[] {
  const out: any[] = [];
  for (const b of blocks) {
    if (b.type === 'text') {
      // The Messages API rejects empty/whitespace-only text blocks ("text
      // content blocks must be non-empty") — drop them; an all-empty message
      // is then filtered by the content-length guards in toAnthropicMessages.
      if (!b.text || !b.text.trim()) continue;
      out.push({ type: 'text', text: b.text });
    } else if (b.type === 'image') {
      out.push({
        type: 'image',
        source: { type: 'base64', media_type: b.mediaType, data: b.data },
      });
    } else if (b.type === 'document') {
      // Native PDF document block — the Messages API renders the pages and the
      // model reads them as vision. The base64 document source accepts ONLY
      // application/pdf (buildUserMessage gates it on canNativeDocument).
      out.push({
        type: 'document',
        source: { type: 'base64', media_type: b.mediaType, data: b.data },
      });
    } else if (b.type === 'tool_use') {
      out.push({
        type: 'tool_use',
        id: b.id,
        name: b.name,
        input: b.input || {},
      });
    } else if (b.type === 'tool_result') {
      out.push({
        type: 'tool_result',
        tool_use_id: b.toolUseId,
        content: b.content,
        is_error: b.isError || false,
      });
    }
  }
  return out;
}

function toAnthropicMessages(pi: PiMessage[]): any[] {
  const msgs = pi
    .filter((m) => m.content.length > 0)
    .map((m) => ({
      role: m.role === 'assistant' ? 'assistant' : 'user',
      content: toAnthropicContent(m.content),
    }))
    .filter((m) => m.content.length > 0);
  // The Messages API requires the first message to be user-role. Rolling
  // history windows (customer buffers) are trimmed user-first at the source
  // (channels/manager.ts trimCustomerBuffer), but defend here too — a leading
  // assistant message 400s the whole request (audit C-7).
  while (msgs.length > 0 && msgs[0].role !== 'user') msgs.shift();
  return msgs;
}

function toAnthropicTools(tools: { name: string; description: string; inputSchema: Record<string, any> }[]) {
  return tools.map((t) => ({
    name: t.name,
    description: t.description,
    input_schema: t.inputSchema,
  }));
}

function mapStopReason(reason?: string): PiStopReason {
  switch (reason) {
    case 'end_turn':      return 'end_turn';
    case 'stop_sequence': return 'end_turn';
    case 'max_tokens':    return 'max_tokens';
    case 'tool_use':      return 'tool_use';
    case 'pause_turn':    return 'end_turn';
    case 'refusal':       return 'error';
    default:              return 'end_turn';
  }
}

/* ── Streaming entry point ── */

interface PartialBlock {
  kind: 'text' | 'tool_use' | 'other';
  text?: string;
  toolUseId?: string;
  toolName?: string;
  toolArgsBuf?: string;
}

export async function* streamAnthropic(req: PiStreamRequest): AsyncIterable<PiStreamEvent> {
  const url = `${req.baseUrl.replace(/\/+$/, '')}/messages`;

  const body: any = {
    model: req.modelId,
    messages: toAnthropicMessages(req.messages),
    max_tokens: req.maxOutputTokens ?? 8192,
    stream: true,
  };
  // Prompt caching (3 of the 4 allowed breakpoints). Without these, every tool
  // round re-prefills the full system prompt + history at full input price —
  // up to 25x per agentic turn. The request prefix is tools → system →
  // messages, so: last tool def caches the tool block, the system block caches
  // tools+system as one prefix, and the last history block caches the
  // conversation so far (Anthropic checks previous breakpoint positions for
  // the longest cached prefix as the marker moves forward each round).
  if (req.systemPrompt?.trim()) {
    body.system = [{ type: 'text', text: req.systemPrompt, cache_control: { type: 'ephemeral' } }];
  }
  if (req.tools && req.tools.length > 0) {
    body.tools = toAnthropicTools(req.tools);
    body.tools[body.tools.length - 1].cache_control = { type: 'ephemeral' };
    // Round-cap wrap-up: forbid further tool calls; tools stay declared so
    // tool_use/tool_result blocks in history remain valid.
    if (req.toolChoice === 'none') body.tool_choice = { type: 'none' };
  }
  if (Array.isArray(body.messages) && body.messages.length > 0) {
    const lastContent = body.messages[body.messages.length - 1].content;
    if (Array.isArray(lastContent) && lastContent.length > 0) {
      lastContent[lastContent.length - 1].cache_control = { type: 'ephemeral' };
    }
  }

  let res: Response;
  try {
    res = await fetchWithRetry(url, {
      method: 'POST',
      headers: {
        'content-type': 'application/json',
        'accept': 'text/event-stream',
        'x-api-key': req.apiKey,
        'anthropic-version': '2023-06-01',
      },
      body: JSON.stringify(body),
      signal: req.signal,
    });
  } catch (err: any) {
    if (err?.name === 'AbortError') {
      yield { type: 'done', stopReason: 'aborted' };
      return;
    }
    const cls = classifyPiNetworkError('Anthropic', err);
    yield { type: 'error', error: cls.message, kind: cls.kind, retryable: cls.retryable };
    return;
  }

  if (!res.ok) {
    let detail = '';
    try { detail = await res.text(); } catch {}
    const cls = classifyPiError('Anthropic', res.status, res.statusText, detail);
    yield { type: 'error', error: cls.message, status: cls.status, kind: cls.kind, retryable: cls.retryable };
    return;
  }

  // Anthropic streams content blocks by index. Track partial state per index
  // so deltas land on the right block.
  const blocks = new Map<number, PartialBlock>();
  let accumulated = '';
  let lastStop: string | undefined;
  let usage: PiUsage | undefined;
  let chunkCount = 0;
  let firstChunkSummary = '';
  let thinkingEmitted = false;

  try {
    for await (const evt of parseSse(res)) {
      chunkCount++;
      if (chunkCount === 1) {
        try { firstChunkSummary = JSON.stringify(evt).slice(0, 600); } catch {}
      }
      const type = evt?.type;
      switch (type) {
        case 'message_start': {
          const u = evt?.message?.usage;
          if (u) {
            usage = {
              inputTokens: u.input_tokens,
              outputTokens: u.output_tokens,
              // With prompt caching on, the bulk of the prompt is cache reads —
              // input_tokens alone would massively under-report occupancy and
              // the supervisor's recycler would never fire.
              cacheReadTokens: u.cache_read_input_tokens,
              cacheCreationTokens: u.cache_creation_input_tokens,
            };
          }
          break;
        }
        case 'content_block_start': {
          const idx = evt?.index ?? 0;
          const block = evt?.content_block || {};
          if (block.type === 'text') {
            blocks.set(idx, { kind: 'text', text: '' });
          } else if (block.type === 'tool_use') {
            blocks.set(idx, {
              kind: 'tool_use',
              toolUseId: block.id,
              toolName: block.name,
              toolArgsBuf: '',
            });
          } else {
            // Extended-thinking blocks (not requested today, future-proofed):
            // one liveness pulse, text never forwarded.
            if (block.type === 'thinking' && !thinkingEmitted) {
              thinkingEmitted = true;
              yield { type: 'thinking' };
            }
            blocks.set(idx, { kind: 'other' });
          }
          break;
        }
        case 'content_block_delta': {
          const idx = evt?.index ?? 0;
          const delta = evt?.delta || {};
          const slot = blocks.get(idx);
          if (!slot) break;
          if (delta.type === 'text_delta' && typeof delta.text === 'string') {
            slot.text = (slot.text || '') + delta.text;
            accumulated += delta.text;
            yield { type: 'text_delta', delta: delta.text };
          } else if (delta.type === 'input_json_delta' && typeof delta.partial_json === 'string') {
            slot.toolArgsBuf = (slot.toolArgsBuf || '') + delta.partial_json;
          }
          // `thinking_delta` (extended thinking) is ignored for now — the
          // pi harness doesn't surface reasoning to the user yet.
          break;
        }
        case 'content_block_stop': {
          const idx = evt?.index ?? 0;
          const slot = blocks.get(idx);
          if (!slot) break;
          if (slot.kind === 'tool_use' && slot.toolUseId && slot.toolName) {
            let input: any = {};
            if (slot.toolArgsBuf) {
              try { input = JSON.parse(slot.toolArgsBuf); }
              catch {
                // Truncated tool-call JSON (output cap hit mid-arguments).
                // Executing a fabricated {_raw} input sends the model into an
                // unwinnable retry loop — fail the round loudly instead.
                yield {
                  type: 'error',
                  error: `The model's ${slot.toolName} call was cut off by the output-token limit (${req.maxOutputTokens ?? 8192} tokens) — the arguments did not fit. Try a smaller change, or raise the model's output budget.`,
                  kind: 'other',
                  retryable: false,
                };
                yield { type: 'done', stopReason: 'error', usage };
                return;
              }
            }
            yield {
              type: 'tool_use',
              id: slot.toolUseId,
              name: slot.toolName,
              input,
            };
          }
          break;
        }
        case 'message_delta': {
          if (evt?.delta?.stop_reason) lastStop = evt.delta.stop_reason;
          const u = evt?.usage;
          if (u && (u.output_tokens !== undefined || u.input_tokens !== undefined)) {
            usage = {
              inputTokens: u.input_tokens ?? usage?.inputTokens,
              outputTokens: u.output_tokens ?? usage?.outputTokens,
              cacheReadTokens: u.cache_read_input_tokens ?? usage?.cacheReadTokens,
              cacheCreationTokens: u.cache_creation_input_tokens ?? usage?.cacheCreationTokens,
            };
          }
          break;
        }
        case 'error': {
          // In-stream error event (e.g. overloaded_error) — classify so the
          // session can retry transient ones and the user sees friendly text.
          const cls = classifyPiError('Anthropic', undefined, '', JSON.stringify(evt?.error ?? evt ?? {}));
          const isOverloaded = (evt?.error?.type || '') === 'overloaded_error';
          yield {
            type: 'error',
            error: cls.kind === 'other' && !isOverloaded
              ? `Anthropic stream error: ${evt?.error?.message || evt?.message || 'Unknown error'}`
              : (isOverloaded ? 'Anthropic is overloaded right now — try again in a moment.' : cls.message),
            kind: isOverloaded ? 'transient' : cls.kind,
            retryable: isOverloaded || cls.retryable,
          };
          yield { type: 'done', stopReason: 'error', usage };
          return;
        }
        case 'message_stop':
        case 'ping':
        default:
          // ping is keep-alive; message_stop is bookkeeping.
          break;
      }
    }
  } catch (err: any) {
    if (err?.name === 'AbortError') {
      yield { type: 'done', stopReason: 'aborted' };
      return;
    }
    const cls = classifyPiNetworkError('Anthropic', err);
    yield { type: 'error', error: cls.message, kind: cls.kind, retryable: cls.retryable };
    return;
  }

  const hadToolUse = Array.from(blocks.values()).some((b) => b.kind === 'tool_use');

  log.info(
    `[pi/anthropic] stream done — chunks=${chunkCount} text=${accumulated.length} ` +
    `toolCalls=${Array.from(blocks.values()).filter((b) => b.kind === 'tool_use').length} ` +
    `stopReason=${lastStop || 'none'} ` +
    `promptTok=${usage?.inputTokens ?? '?'} outTok=${usage?.outputTokens ?? '?'}`,
  );
  if (chunkCount === 0) {
    log.warn(`[pi/anthropic] zero chunks parsed — content-type=${res.headers.get('content-type') || '?'}`);
  } else if (!accumulated && !hadToolUse) {
    log.info(`[pi/anthropic] first chunk (truncated): ${firstChunkSummary}`);
  }

  if (accumulated) yield { type: 'text_end', text: accumulated };

  if (!accumulated && !hadToolUse) {
    yield {
      type: 'error',
      error: `Anthropic returned no output (stopReason=${lastStop || 'unknown'}).`,
    };
    yield { type: 'done', stopReason: 'error', usage };
    return;
  }

  yield {
    type: 'done',
    stopReason: hadToolUse ? 'tool_use' : mapStopReason(lastStop),
    usage,
  };
}