// P5 (SPLIT SKETCH): the generic AI-SDK adapter, split out of stream-phase.ts.
// This module is the v1.13.1-A streamText adapter and nothing else — it has NO
// SQL, broker, or BooCode persistence dependencies (its only `ctx` access is
// config + log), so it can be unit-tested without standing up a DB or broker.
// stream-phase.ts (the I/O layer) re-exports the public names below so existing
// importers (`./stream-phase.js`) are unchanged.

import type { FastifyBaseLogger } from 'fastify';
import type { Config } from '../../config.js';
import type { Agent, ToolCall } from '../../types/api.js';
import type { ToolJsonSchema } from '../tools.js';
import type { OpenAiMessage } from './payload.js';
import { extractToolCallBlocks } from './tool-call-parser.js';
import { ThinkSplitter } from './think-splitter.js';
import { classifyStreamError } from './stream-error-classifier.js';
import type { StreamResult } from './types.js';
import { resolveModelProvider, upstreamModel } from './provider.js';
import {
  jsonSchema,
  streamText,
  tool,
  type JSONValue,
  type ModelMessage,
  type ToolCallRepairFunction,
} from 'ai';

// The slice of InferenceContext the adapter actually needs. Narrowing it here
// (instead of taking the full InferenceContext) keeps the adapter free of the
// SQL/broker/publish surface. InferenceContext structurally satisfies this, so
// callers pass their ctx unchanged.
export interface StreamAdapterContext {
  config: Config;
  log: FastifyBaseLogger;
}

export interface StreamOptions {
  // null = omit tools entirely (compact phase); [] = caller stripped all tools
  // (rare; we still omit from the request body to avoid OpenAI 400).
  tools: ToolJsonSchema[] | null;
  temperature?: number;
  top_p?: number | null;
  top_k?: number | null;
  min_p?: number | null;
  presence_penalty?: number | null;
  // v2.6 sampling-streamjson-tokens (#11): llama.cpp sampler extensions. These
  // are NOT standard AI-SDK streamText options and are NOT serialized by the
  // openai-compatible provider's standardized-settings path (topK is even
  // explicitly dropped with an "unsupported feature: topK" warning). They reach
  // llama-server only via providerOptions.openaiCompatible (see buildSamplerProviderOptions).
  top_n_sigma?: number | null;
  dry_multiplier?: number | null;
  dry_base?: number | null;
  dry_allowed_length?: number | null;
  dry_penalty_last_n?: number | null;
  // vDeepSeek: thinking/reasoning effort. Maps to DeepSeek's reasoning_effort
  // API param for deepseek-v4-flash / deepseek-v4-pro models.
  reasoning_effort?: 'off' | 'low' | 'medium' | 'high' | 'xhigh' | 'max';
  // vDeepSeek: JSON output mode. When set, model outputs valid JSON.
  response_format?: { type: 'json_object' };
  // vDeepSeek: prefix content for chat prefix completion.
  prefix_content?: string;
}

// P5: the 10-field sampler-options literal that was copy-pasted at 4 sites
// (the three sentinel summaries + executeStreamPhase). Builds the StreamOptions
// sampler subset from an agent's frontmatter knobs. `temperature` is
// `agent?.temperature` (already number|undefined); the nullable fields strip
// null → undefined so they're omitted from the request body when unset. Keep
// this in lockstep with the StreamOptions sampler fields — a new sampler knob
// (the v2.7.3 dry_* family did this) is added here once instead of at 4 sites.
export type SamplerOpts = Omit<StreamOptions, 'tools'>;

export function samplerOptsFromAgent(agent: Agent | null): SamplerOpts {
  return {
    temperature: agent?.temperature,
    top_p: agent?.top_p ?? undefined,
    top_k: agent?.top_k ?? undefined,
    min_p: agent?.min_p ?? undefined,
    presence_penalty: agent?.presence_penalty ?? undefined,
    top_n_sigma: agent?.top_n_sigma ?? undefined,
    dry_multiplier: agent?.dry_multiplier ?? undefined,
    dry_base: agent?.dry_base ?? undefined,
    dry_allowed_length: agent?.dry_allowed_length ?? undefined,
    dry_penalty_last_n: agent?.dry_penalty_last_n ?? undefined,
    reasoning_effort: agent?.reasoning_effort ?? undefined,
    response_format: agent?.response_format ?? undefined,
    prefix_content: agent?.prefix_content ?? undefined,
  };
}

// v2.6 #11: build the providerOptions.openaiCompatible extraBody object for the
// llama.cpp sampler extensions. @ai-sdk/openai-compatible (2.0.47) merges every
// non-reserved key under providerOptions.openaiCompatible straight into the
// chat-completion request body (see its getArgs: the Object.fromEntries spread
// filtered against openaiCompatibleLanguageModelChatOptions.shape). This is the
// ONLY working passthrough for these params:
//   - top_k / min_p were latently dropped before this: top_k was passed as the
//     AI-SDK `topK` setting which the openai-compatible provider rejects as
//     unsupported; min_p was never passed to streamText at all.
//   - top_n_sigma + the dry_* family have no AI-SDK equivalent.
// Keys use llama-server's snake_case body names so they land verbatim.
function buildSamplerProviderOptions(opts: StreamOptions): Record<string, number> | undefined {
  const body: Record<string, number> = {};
  if (typeof opts.top_k === 'number') body.top_k = opts.top_k;
  if (typeof opts.min_p === 'number') body.min_p = opts.min_p;
  if (typeof opts.top_n_sigma === 'number') body.top_n_sigma = opts.top_n_sigma;
  if (typeof opts.dry_multiplier === 'number') body.dry_multiplier = opts.dry_multiplier;
  if (typeof opts.dry_base === 'number') body.dry_base = opts.dry_base;
  if (typeof opts.dry_allowed_length === 'number') body.dry_allowed_length = opts.dry_allowed_length;
  if (typeof opts.dry_penalty_last_n === 'number') body.dry_penalty_last_n = opts.dry_penalty_last_n;
  return Object.keys(body).length > 0 ? body : undefined;
}

// v1.13.1-A: convert BooCode's OpenAI-shaped history into AI SDK
// ModelMessage[]. Tool result messages need a `toolName` field that the
// OpenAI shape doesn't carry; we look it up by scanning earlier assistant
// `tool_calls` entries for a matching id.
function toModelMessages(messages: OpenAiMessage[]): ModelMessage[] {
  const toolNameById = new Map<string, string>();
  for (const m of messages) {
    if (m.role === 'assistant' && m.tool_calls) {
      for (const tc of m.tool_calls) {
        toolNameById.set(tc.id, tc.function.name);
      }
    }
  }
  const out: ModelMessage[] = [];
  for (const m of messages) {
    if (m.role === 'system' || m.role === 'user') {
      out.push({ role: m.role, content: m.content ?? '' });
      continue;
    }
    if (m.role === 'assistant') {
      const hasTools = m.tool_calls && m.tool_calls.length > 0;
      // Anthropic signed thinking blocks (Phase 2) must be replayed verbatim per
      // block with their signature; other reasoning uses the joined string.
      const signedBlocks = (m.reasoning_blocks ?? []).filter((b) => b.signature);
      const hasReasoning =
        (typeof m.reasoning === 'string' && m.reasoning.length > 0) || signedBlocks.length > 0;
      if (!hasTools && !hasReasoning) {
        // Bare text assistant (string content). null content + no tool_calls
        // is degenerate but harmless to forward.
        out.push({ role: 'assistant', content: m.content ?? '' });
        continue;
      }
      // v1.13.1-C: AI SDK ReasoningPart precedes text + tool-calls in the
      // assistant content array. Reasoning models (qwen3.6) consume their
      // prior reasoning context to resume mid-thought across tool boundaries.
      const parts: Array<
        | { type: 'reasoning'; text: string; providerOptions?: Record<string, Record<string, JSONValue>> }
        | { type: 'text'; text: string }
        | { type: 'tool-call'; toolCallId: string; toolName: string; input: unknown }
      > = [];
      if (signedBlocks.length > 0) {
        for (const b of m.reasoning_blocks!) {
          parts.push({
            type: 'reasoning',
            text: b.text,
            ...(b.signature ? { providerOptions: { anthropic: { signature: b.signature } } } : {}),
          });
        }
      } else if (typeof m.reasoning === 'string' && m.reasoning.length > 0) {
        parts.push({ type: 'reasoning', text: m.reasoning });
      }
      if (m.content && m.content.length > 0) {
        parts.push({ type: 'text', text: m.content });
      }
      for (const tc of m.tool_calls ?? []) {
        let input: unknown = {};
        try {
          input = tc.function.arguments.length > 0 ? JSON.parse(tc.function.arguments) : {};
        } catch {
          // Malformed args from a prior turn: pass through as a raw blob so
          // the model sees the same shape it emitted. Wraps the string under
          // _raw to match the buildMessagesPayload upstream convention.
          input = { _raw: tc.function.arguments };
        }
        parts.push({ type: 'tool-call', toolCallId: tc.id, toolName: tc.function.name, input });
      }
      out.push({ role: 'assistant', content: parts });
      continue;
    }
    if (m.role === 'tool') {
      const toolCallId = m.tool_call_id ?? '';
      const toolName = toolNameById.get(toolCallId) ?? 'unknown';
      const raw = m.content ?? '';
      let output: { type: 'text'; value: string } | { type: 'json'; value: JSONValue };
      try {
        // JSON.parse returns `any`; cast to JSONValue since the upstream
        // tool_results column is already JSON-serializable by construction.
        output = { type: 'json', value: JSON.parse(raw) as JSONValue };
      } catch {
        output = { type: 'text', value: raw };
      }
      out.push({
        role: 'tool',
        content: [{ type: 'tool-result', toolCallId, toolName, output }],
      });
      continue;
    }
  }
  return out;
}

// Build the AI SDK tools record from BooCode's JSON-schema tool definitions.
// No `execute` field: BooCode runs tools itself in tool-phase.ts; streamText
// surfaces the tool-call parts via fullStream and we capture them for the
// outer loop to dispatch.
function buildAiTools(schemas: ToolJsonSchema[], strict?: boolean): Record<string, ReturnType<typeof tool>> {
  const out: Record<string, ReturnType<typeof tool>> = {};
  for (const s of schemas) {
    out[s.function.name] = tool({
      description: s.function.description,
      inputSchema: jsonSchema(s.function.parameters),
      ...(strict ? { strict } : {}),
    });
  }
  return out;
}

// F6: per-chunk stall deadline. Exported so tests can advance fake timers by
// exactly this value without hardcoding a magic number.
export const STALL_TIMEOUT_MS = 90_000;

// v1.10.5 Qwen-coder XML fallback. Some local models (notably qwen3-coder via
// llama-swap) emit tool calls as inline XML inside delta.content rather than
// the structured tool_calls field. We extract them out of the streamed text
// before flushing it to the client.
//
// Qwen shape:
//   <tool_call>
//   <function=NAME>
//   <parameter=KEY>VALUE</parameter>
//   ...
//   </function>
//   </tool_call>
//
// v1.13.16: also recognize Anthropic <invoke> markup that qwen3.6-35b-a3b-mxfp4
// drifts to (training-data residue from Claude Code documentation):
//   <invoke name="NAME">
//   <parameter name="KEY">VALUE</parameter>
//   </invoke>
// Both formats share the synthetic xml_call_${idx} ID space; the counter
// increments across whichever opener appears first. Multiple blocks may
// appear back-to-back in either format and they never nest.
export async function streamCompletion(
  ctx: StreamAdapterContext,
  model: string,
  messages: OpenAiMessage[],
  opts: StreamOptions,
  onDelta: (content: string) => void,
  onUsage: ((prompt: number | null, completion: number | null) => void) | undefined,
  signal?: AbortSignal,
  agent?: Agent | null,
): Promise<StreamResult> {
  const aiMessages = toModelMessages(messages);
  const hasTools = opts.tools !== null && opts.tools.length > 0;
  // DeepSeek detection via providerId (handles both bare "deepseek-*" and
  // composite "deepseek/model" — JD1 fix).
  const resolvedModel = resolveModelProvider(model, ctx.config);
  const isDsModel = resolvedModel.providerId === 'deepseek';
  // Anthropic's Messages API rejects the llama.cpp sampler extensions and
  // out-of-range temperature/top_p (Opus 4.x), so we omit all of them for the
  // anthropic route and let the model's defaults apply.
  const isAnthropic = resolvedModel.route === 'anthropic';
  // Extended thinking is opt-in per agent via reasoning_effort (same gate as
  // DeepSeek). Adaptive lets Claude choose depth; display:'summarized' surfaces
  // the reasoning to BooChat (Opus 4.7+ default 'omitted' = empty-text blocks).
  const anthropicThinkingEnabled = isAnthropic && !!opts.reasoning_effort && opts.reasoning_effort !== 'off';
  const aiTools = hasTools ? buildAiTools(opts.tools!, isDsModel) : undefined;

  const startedAt = Date.now();
  // v1.13.1-C: accumulate reasoning text across reasoning-delta parts.
  // qwen3.6 emits these on a separate channel from text content; we capture
  // them per stream so finalizeCompletion can dual-write a 'reasoning' part.
  // Replaces the v1.13.1-A counter-only diagnostic.
  let reasoningAccumulated = '';

  // Phase 2 (anthropic): reasoning grouped per thinking block (keyed by the
  // stream part id = content-block index), each carrying its signature. The
  // signature arrives as a reasoning-delta with empty text + providerMetadata.
  // Insertion order is preserved (Map), so replay order matches the model's.
  const reasoningBlockMap = new Map<string, { text: string; signature?: string }>();

  // Peel inline <think>...</think> reasoning out of the text-delta channel for
  // local models that don't use a structured reasoning channel. Arms only when
  // content starts with <think>; otherwise a verbatim pass-through (no-op for
  // models whose reasoning already arrives via reasoning-delta).
  const thinkSplitter = new ThinkSplitter();

  // v1.13.3: experimental_repairToolCall keeps the stream alive when the
  // model emits a malformed tool call (bad JSON args, unknown name, etc.).
  // Without a repair function streamText throws and the WHOLE stream dies;
  // with one, the SDK invokes us and we route the bad call through normally.
  // Strategy: pass through unmodified. executeToolPhase's existing error
  // path (unknown tool name → "unknown tool: X" result; zod-reject → tool
  // 'X' rejected — fieldname: required) already gives the model a clean
  // recovery surface on the next turn. Logging gives us visibility into
  // how often qwen3.6 actually emits broken calls.
  const repairToolCall: ToolCallRepairFunction<NonNullable<typeof aiTools>> = async ({
    toolCall,
    error,
  }) => {
    ctx.log.warn(
      {
        toolCallId: toolCall.toolCallId,
        toolName: toolCall.toolName,
        error: error.message,
      },
      'malformed tool call surfaced via repairToolCall',
    );
    return toolCall;
  };

  // v2.6 #11: llama.cpp sampler extensions (top_k, min_p, top_n_sigma, dry_*)
  // ride providerOptions.openaiCompatible — they are NOT standardized streamText
  // settings. NB: top_k used to be passed below as the AI-SDK `topK` setting;
  // the openai-compatible provider dropped it with an "unsupported feature: topK"
  // warning and min_p was never wired at all, so both were dead on the wire
  // before this. They now go through the same extraBody path as the new params.
  // Omit llama.cpp sampler extensions on the anthropic route (rejected there).
  const samplerBody = isAnthropic ? undefined : buildSamplerProviderOptions(opts);

  // vDeepSeek: build providerOptions.deepseek for DeepSeek V4 models.
  let deepseekProviderOptions:
    | { thinking: { type: 'enabled' | 'disabled' }; reasoningEffort?: 'low' | 'medium' | 'high' | 'xhigh' | 'max' }
    | undefined;
  if (isDsModel) {
    const dsEffort = opts.reasoning_effort;
    const thinkingEnabled = dsEffort && dsEffort !== 'off';
    deepseekProviderOptions = {
      thinking: { type: thinkingEnabled ? 'enabled' : 'disabled' },
      ...(thinkingEnabled ? { reasoningEffort: dsEffort } : {}),
    };
  }

  // F6: per-chunk stall deadline. If the model stops emitting chunks for
  // STALL_TIMEOUT_MS the stallAc fires through AbortSignal.any; the post-loop
  // abort check below then throws AbortError → handleAbortOrError writes
  // 'cancelled'. Timer is bumped on every chunk and cleared in the finally.
  // NO retry: partial-stream re-emit is non-idempotent at single-local-instance
  // scale; see stream-error-classifier.ts for the future retry seam.
  const stallAc = new AbortController();
  let stallTimer: ReturnType<typeof setTimeout> | null = null;
  const bumpStallTimer = () => {
    if (stallTimer !== null) clearTimeout(stallTimer);
    stallTimer = setTimeout(() => stallAc.abort(), STALL_TIMEOUT_MS);
  };
  const effectiveSignal = signal
    ? AbortSignal.any([signal, stallAc.signal])
    : stallAc.signal;

  // vDeepSeek: chat prefix completion (B2). When prefix_content is set and
  // the model is DeepSeek, inject an assistant message with prefix=true.
  // If the last message is already an assistant, replace its content to
  // avoid two consecutive assistant messages (AV5 fix).
  let prefixMessages = aiMessages;
  const useBeta = isDsModel && (hasTools || !!opts.prefix_content);
  if (opts.prefix_content && isDsModel) {
    const prefixMsg = { role: 'assistant' as const, content: opts.prefix_content };
    const last = prefixMessages[prefixMessages.length - 1];
    if (last && last.role === 'assistant') {
      prefixMessages = [...prefixMessages.slice(0, -1), prefixMsg];
    } else {
      prefixMessages = [...prefixMessages, prefixMsg];
    }
  }

  const result = streamText({
    model: upstreamModel(ctx.config, model, agent ?? null, 'boochat', useBeta),
    messages: prefixMessages,
    ...(aiTools
      ? { tools: aiTools, toolChoice: 'auto' as const, experimental_repairToolCall: repairToolCall }
      : {}),
    ...(!isAnthropic && typeof opts.temperature === 'number' ? { temperature: opts.temperature } : {}),
    ...(!isAnthropic && typeof opts.top_p === 'number' ? { topP: opts.top_p } : {}),
    ...(!isAnthropic && typeof opts.presence_penalty === 'number' ? { presencePenalty: opts.presence_penalty } : {}),
    ...(!isAnthropic && opts.response_format ? { responseFormat: { type: 'json_object' } } : {}),
    ...(samplerBody || deepseekProviderOptions || anthropicThinkingEnabled
      ? {
          providerOptions: {
            ...(samplerBody ? { openaiCompatible: samplerBody } : {}),
            ...(deepseekProviderOptions ? { deepseek: deepseekProviderOptions } : {}),
            ...(anthropicThinkingEnabled
              ? { anthropic: { thinking: { type: 'adaptive', display: 'summarized' } } }
              : {}),
          },
        }
      : {}),
    abortSignal: effectiveSignal,
  });

  let content = '';
  let pendingBuffer = '';
  let finishReason: string | null = null;
  // v1.13.1-A: AI SDK emits one `tool-call` part per fully-aggregated call,
  // so we no longer need the OpenAI-index reassembly map the manual SSE
  // parser used. XML tool calls extracted from text content go into the
  // same flat list and keep the v1.10.5 synthetic id convention.
  const toolCalls: ToolCall[] = [];

  bumpStallTimer();

  try {
  for await (const part of result.fullStream) {
    bumpStallTimer();
    switch (part.type) {
      case 'text-delta': {
        // Peel any inline <think> reasoning before tool-call extraction; the
        // reasoning span accumulates exactly like a structured reasoning-delta.
        const split = thinkSplitter.push(part.text);
        if (split.reasoning) reasoningAccumulated += split.reasoning;
        if (!split.text) break;
        pendingBuffer += split.text;
        // v1.13.16: unified extraction. The helper finds the earliest-opening
        // complete <tool_call> or <invoke> block, flushes prose between/around
        // them, holds any partial opener for the next chunk, and silently
        // drops blocks that fail to parse (matches pre-v1.13.16 behavior).
        const extracted = extractToolCallBlocks(pendingBuffer, ctx.log);
        if (extracted.flushed.length > 0) {
          content += extracted.flushed;
          onDelta(extracted.flushed);
        }
        for (const call of extracted.calls) {
          const synthIdx = toolCalls.length;
          toolCalls.push({
            id: `xml_call_${synthIdx}`,
            name: call.name,
            args: call.args,
          });
        }
        pendingBuffer = extracted.remaining;
        break;
      }
      case 'tool-call': {
        // AI SDK has already parsed the input into an object. Match the
        // ToolCall shape BooCode passes around in toolCallsBuffer downstream.
        toolCalls.push({
          id: part.toolCallId,
          name: part.toolName,
          args: (part.input ?? {}) as Record<string, unknown>,
        });
        break;
      }
      case 'reasoning-delta': {
        // v1.13.1-C: accumulate the joined string (compaction prose + non-
        // anthropic replay). Phase 2: also group per block id and capture the
        // Anthropic signature (arrives on a delta with empty text).
        if (typeof part.text === 'string') {
          reasoningAccumulated += part.text;
        }
        const blk = reasoningBlockMap.get(part.id) ?? { text: '' };
        if (typeof part.text === 'string') blk.text += part.text;
        const sig = (part.providerMetadata as Record<string, Record<string, unknown> | undefined> | undefined)
          ?.anthropic?.signature;
        if (typeof sig === 'string') blk.signature = sig;
        reasoningBlockMap.set(part.id, blk);
        break;
      }
      case 'finish': {
        if (typeof part.finishReason === 'string') {
          finishReason = part.finishReason;
        }
        break;
      }
      case 'error': {
        const err = part.error;
        const actualErr = err instanceof Error ? err : new Error(String(err));
        ctx.log.warn({ kind: classifyStreamError(actualErr) }, 'stream error part');
        throw actualErr;
      }
      // Intentional no-op: start, start-step, text-start, text-end,
      // reasoning-start, reasoning-end, source, file, tool-input-start,
      // tool-input-delta, tool-input-end, tool-result, tool-error,
      // finish-step, raw. We only care about the aggregated tool-call and
      // text-delta paths above; the rest are AI SDK lifecycle/streaming
      // breadcrumbs that don't change BooCode's persistence or WS contract.
      default:
        break;
    }
  }

  // Resolve any text the splitter was holding (an unterminated <think> block
  // becomes reasoning; a held partial opener becomes text).
  const splitTail = thinkSplitter.flush();
  if (splitTail.reasoning) reasoningAccumulated += splitTail.reasoning;
  if (splitTail.text) pendingBuffer += splitTail.text;

  // v1.13.1-A: drain any buffered partial XML opener as plain text. The
  // pre-AI-SDK path did this on stream end too — better to leak `<tool_c`
  // than vanish the text.
  if (pendingBuffer.length > 0) {
    content += pendingBuffer;
    onDelta(pendingBuffer);
    pendingBuffer = '';
  }

  // AI SDK v6 fullStream returns normally on abort; check signal explicitly.
  // Without this throw the row would land as status='complete' with partial
  // content instead of going through handleAbortOrError → status='cancelled'.
  // Smoke D caught this in v1.13.1-A — don't refactor it away.
  // F6: also catch the stall timeout arm (stallAc.signal.aborted).
  if (signal?.aborted || stallAc.signal.aborted) {
    const abortErr = new Error('aborted');
    abortErr.name = 'AbortError';
    throw abortErr;
  }

  // Usage lands as a promise on the result; awaiting after fullStream is
  // drained is safe. AI SDK v6 names: `inputTokens` / `outputTokens`.
  // Some providers (llama-swap via openai-compatible) return plain numbers;
  // others (deepseek via @ai-sdk/deepseek) return {total, cacheRead, noCache, ...}.
  let promptTokens: number | null = null;
  let completionTokens: number | null = null;
  let cacheReadTokens: number | null = null;
  let reasoningTokens: number | null = null;
  try {
    const usage = await result.usage;
    if (typeof usage.inputTokens === 'number') {
      promptTokens = usage.inputTokens;
    } else if (usage.inputTokens && typeof usage.inputTokens === 'object') {
      promptTokens = (usage.inputTokens as Record<string, number | undefined>).total ?? null;
      cacheReadTokens = (usage.inputTokens as Record<string, number | undefined>).cacheRead ?? null;
    }
    if (typeof usage.outputTokens === 'number') {
      completionTokens = usage.outputTokens;
    } else if (usage.outputTokens && typeof usage.outputTokens === 'object') {
      completionTokens = (usage.outputTokens as Record<string, number | undefined>).total ?? null;
      reasoningTokens = (usage.outputTokens as Record<string, number | undefined>).reasoning ?? null;
    }
  } catch {
    // Some providers omit usage on partial streams; leave both null.
  }

  if (onUsage && (promptTokens !== null || completionTokens !== null)) {
    onUsage(promptTokens, completionTokens);
  }

  if (reasoningAccumulated.length > 0) {
    ctx.log.debug(
      { reasoningChars: reasoningAccumulated.length, model, elapsed_ms: Date.now() - startedAt },
      'streamCompletion: captured reasoning',
    );
  }

  if (cacheReadTokens !== null || reasoningTokens !== null) {
    ctx.log.debug(
      { promptTokens, completionTokens, cacheReadTokens, reasoningTokens, model },
      'streamCompletion: deepseek usage breakdown',
    );
  }

  const reasoningBlocks = Array.from(reasoningBlockMap.values()).filter(
    (b) => b.text.length > 0 || b.signature,
  );
  // Only signed blocks need verbatim per-block replay; text-only reasoning uses
  // the joined string, so omit reasoningBlocks unless something was signed.
  const hasSignedReasoning = reasoningBlocks.some((b) => b.signature);

  return {
    finishReason,
    content,
    toolCalls,
    promptTokens,
    completionTokens,
    reasoning: reasoningAccumulated,
    ...(hasSignedReasoning ? { reasoningBlocks } : {}),
    // vDeepSeek: optional usage breakdown populated when the provider returns
    // structured usage (cache hit tokens, reasoning tokens).
    cacheReadTokens: cacheReadTokens ?? undefined,
    reasoningTokens: reasoningTokens ?? undefined,
  };
  } finally {
    // Clear the stall timer whether the stream completes normally, throws, or
    // is aborted — prevents a dangling timer from firing after the turn ends.
    if (stallTimer !== null) {
      clearTimeout(stallTimer);
      stallTimer = null;
    }
  }
}
