/**
 * Backend-agnostic stream state accumulator.
 *
 * Every backend consumes a token-by-token stream and accumulates the same
 * pieces of information: text-so-far, tool calls, trailing prose,
 * delivered-text norms, token counts, an end-of-turn flag. This module
 * owns the shape of that accumulator and the helpers that mutate it.
 *
 * Backend adapters are responsible for translating their SDK's native
 * event types into calls against this state — `appendText`,
 * `recordToolUse`, `recordToolCompletion`, `markTurnTerminated`. The
 * shared handler runtime reads from it after the loop ends.
 *
 * Why a hand-managed object instead of an EventEmitter or RxJS pipeline:
 *   - The state is mutated from exactly one place (the backend's stream
 *     loop) and read from exactly one place (the post-loop accounting).
 *     A bag-of-fields object is simpler and easier to test than a stream.
 *   - Backends with different event models (Claude SDK async iterator,
 *     Kilo SSE subscription) can both call the same mutators.
 */

import { captureDeliveredText } from "./delivered-text.js";
import { isTurnTerminator, stripMcpPrefix } from "../../core/tools/index.js";
import { updateLiveTurn } from "../../storage/sessions.js";

// ── State shape ─────────────────────────────────────────────────────────────

/** Mutable state accumulated while iterating a backend's stream. */
export interface StreamState {
  // ── Live-stats binding ────────────────────────────────────────────────────
  /**
   * Chat this stream belongs to. When set, token mutators mirror the
   * accumulated counts into the chat's live-turn overlay (see
   * `storage/sessions.ts: updateLiveTurn`) so /status reflects the
   * in-progress turn in real time. Undefined keeps the state pure
   * (tests, contexts with no session).
   */
  chatId?: string;

  // ── Text accumulation ─────────────────────────────────────────────────────
  /** Text in the current pre-tool segment (not yet emitted as progress). */
  currentBlockText: string;
  /** All text accumulated this turn (concatenation of every segment). */
  allResponseText: string;
  /** Text *after* the last tool call (or the entire response if no tools). */
  lastTrailingText: string;
  /**
   * How much of `allResponseText` has already reached the user as a
   * mid-turn progress message.
   *
   * The remote-server backends flush the pending segment through
   * `onTextBlock` at each tool boundary, but `closeCurrentSegment` also
   * folds that segment into `allResponseText` — so the end-of-turn
   * delivery would ship every narration line a second time, concatenated.
   * Delivery ships only `allResponseText.slice(progressDeliveredLen)`.
   *
   * Advanced only after a progress send actually succeeds, so a flush that
   * throws (e.g. Telegram's 4096-char limit) leaves its text pending and
   * the end-of-turn delivery still carries it.
   */
  progressDeliveredLen: number;

  // ── Session ───────────────────────────────────────────────────────────────
  /** Provider-assigned session id, if one was announced during the stream. */
  newSessionId: string | undefined;

  // ── Tool tracking ─────────────────────────────────────────────────────────
  /** Total number of tool_use blocks observed this turn. */
  toolCalls: number;
  /** Set when a turn-terminator (`end_turn` / `send` / `react`) fired. */
  turnTerminated: boolean;
  /** Normalized text args captured from delivery tools — used for dedup. */
  deliveredTextNorms: string[];
  /**
   * True when the model called any bridge-delivering tool this turn:
   * `end_turn(...)` or `send(...)` of any type (text/photo/poll/voice/...).
   * `deliveredTextNorms` only tracks `text`-bearing variants — needed for
   * dedup against assistant prose — so it misses e.g. `send(type="photo")`
   * which still puts a message in chat. The handler uses this flag to
   * suppress an additional text-part delivery when the bridge already
   * shipped something, preventing the doubled-message symptom.
   */
  hadBridgeDelivery: boolean;

  // ── Token accounting ──────────────────────────────────────────────────────
  /** Effective input tokens charged this turn. */
  sdkInputTokens: number;
  /** Output tokens generated by the model this turn. */
  sdkOutputTokens: number;
  /** Cache-read tokens (input pulled from the prompt-cache). */
  sdkCacheRead: number;
  /** Cache-creation/write tokens (new prompt-cache entries this turn). */
  sdkCacheWrite: number;

  // ── Context window ────────────────────────────────────────────────────────
  /** Current context-fill estimate (last API iteration's total input). */
  contextTokens: number;
  /** Active model's full context window in tokens, if reported by the SDK. */
  contextWindow: number | undefined;
  /** API round-trips observed this turn (deltas for multi-tool flows). */
  numApiCalls: number;

  // ── Streaming bookkeeping ─────────────────────────────────────────────────
  /** Timestamp (ms) of the last delta callback — used to throttle UI updates. */
  lastStreamUpdate: number;

  // ── Diagnostics ───────────────────────────────────────────────────────────
  /**
   * Per-event-type counts observed on the backend's stream this turn (e.g.
   * Kilo SSE: `message.part.delta` × N). Backends increment this; the
   * handler logs a summary at end-of-turn so operators can diagnose
   * stuck/empty turns by inspecting which event types fired.
   */
  eventCounts: Record<string, number>;

  /**
   * Backend-generated error text peeled off the response (e.g. Kilo's
   * synthetic "model hit its output limit while reasoning" message).
   * The handler converts this into a user-friendly Talon error instead
   * of shipping the raw upstream string as a chat reply. Empty when the
   * turn produced no synthetic error.
   */
  syntheticError?: string;

  /**
   * partID → part type lookup populated from `message.part.updated` events.
   *
   * Kilo's `message.part.delta` events carry only `partID` and `field` —
   * they don't say what type of part the delta belongs to. A delta with
   * `field: "text"` could be filling the `text` field of a `TextPart`
   * (the actual user-facing reply) OR the `text` field of a
   * `ReasoningPart` (private scratchpad). Without the part-type lookup
   * the SSE consumer can't tell them apart and ends up treating
   * reasoning content as the reply.
   *
   * Backends populate this on every part.updated; the delta handler uses
   * it to classify each delta against its source part. Empty for
   * backends that don't have the same delta/part split.
   */
  partTypes: Map<string, string>;
}

// ── Factories ───────────────────────────────────────────────────────────────

export function createStreamState(chatId?: string): StreamState {
  return {
    chatId,
    currentBlockText: "",
    allResponseText: "",
    lastTrailingText: "",
    progressDeliveredLen: 0,
    newSessionId: undefined,
    toolCalls: 0,
    turnTerminated: false,
    deliveredTextNorms: [],
    hadBridgeDelivery: false,
    sdkInputTokens: 0,
    sdkOutputTokens: 0,
    sdkCacheRead: 0,
    sdkCacheWrite: 0,
    contextTokens: 0,
    contextWindow: undefined,
    numApiCalls: 0,
    lastStreamUpdate: 0,
    eventCounts: {},
    partTypes: new Map(),
  };
}

// ── Mutators ────────────────────────────────────────────────────────────────

/**
 * Append a text fragment to the current pre-tool segment and the
 * cumulative response. Used by streaming-delta handlers.
 */
export function appendText(state: StreamState, fragment: string): void {
  if (!fragment) return;
  state.currentBlockText += fragment;
  state.lastTrailingText += fragment;
}

/**
 * Close out the current pre-tool segment.
 *
 * Called when a tool call boundary is reached. Returns the segment text
 * (trimmed) so the caller can fire a progress-text callback, and resets
 * `currentBlockText` + `lastTrailingText` for the next segment.
 *
 * The cumulative `allResponseText` is updated with the pre-trim segment
 * so the post-turn full-text capture stays faithful to what the model
 * actually produced.
 */
export function closeCurrentSegment(state: StreamState): string {
  const raw = state.currentBlockText;
  const trimmed = raw.trim();
  if (raw) {
    state.allResponseText += raw;
    state.currentBlockText = "";
    state.lastTrailingText = "";
  }
  return trimmed;
}

/**
 * Record a tool call observed in the stream.
 *
 * Updates `toolCalls`, captures any delivered-text norm, and flips
 * `turnTerminated` when the tool is a turn terminator (with the soft-
 * terminator opt-out — `react` with `end_turn: false` keeps the turn
 * alive).
 */
export function recordToolUse(
  state: StreamState,
  toolName: string,
  toolInput: Record<string, unknown>,
): void {
  state.toolCalls += 1;
  const norm = captureDeliveredText(toolName, toolInput);
  if (norm) state.deliveredTextNorms.push(norm);
  if (isBridgeDelivery(toolName)) {
    state.hadBridgeDelivery = true;
  }
  if (isTurnTerminator(toolName, toolInput)) {
    state.turnTerminated = true;
  }
}

// True for tools that ship content to the user via the bridge:
// end_turn and send (any type). Excludes react and the read_*/get_* family.
// Used to suppress a doubled text-part when the bridge already shipped
// content (e.g. send(type="photo") followed by an assistant text part).
function isBridgeDelivery(toolName: string): boolean {
  const bare = stripMcpPrefix(toolName);
  return bare === "end_turn" || bare === "send";
}

/**
 * Bulk-update token accounting for the turn.
 *
 * Backends call this at the end of the stream loop when their final
 * result message carries usage totals. Inputs that are `undefined` or
 * negative are coerced to 0 — defensive against partial SDK payloads.
 */
export function recordTokens(
  state: StreamState,
  tokens: {
    inputTokens?: number;
    outputTokens?: number;
    cacheRead?: number;
    cacheWrite?: number;
  },
): void {
  state.sdkInputTokens = Math.max(0, tokens.inputTokens ?? 0);
  state.sdkOutputTokens = Math.max(0, tokens.outputTokens ?? 0);
  state.sdkCacheRead = Math.max(0, tokens.cacheRead ?? 0);
  state.sdkCacheWrite = Math.max(0, tokens.cacheWrite ?? 0);
  pushLiveUsage(state);
}

/**
 * Mirror the stream state's current token/context counts into the
 * chat's live-turn overlay so /status updates mid-turn. No-op when the
 * state isn't bound to a chat. Backends whose streams report usage
 * incrementally (Kilo/OpenCode SSE, per-call usage events) get this for
 * free via `recordTokens`; backends with out-of-band signals (Codex
 * rollout polls) call it directly after mutating the state.
 */
export function pushLiveUsage(state: StreamState): void {
  if (!state.chatId) return;
  updateLiveTurn(state.chatId, {
    inputTokens: state.sdkInputTokens,
    outputTokens: state.sdkOutputTokens,
    cacheRead: state.sdkCacheRead,
    cacheWrite: state.sdkCacheWrite,
    contextTokens: state.contextTokens,
    contextWindow: state.contextWindow ?? 0,
    numApiCalls: state.numApiCalls,
  });
}

/**
 * Finalise the response text for the post-loop accounting.
 *
 * Concatenates the still-open `currentBlockText` onto `allResponseText`
 * (a tool-free turn never closed a segment via `closeCurrentSegment`
 * but its text still needs to land in the cumulative buffer).
 *
 * Returns the cumulative text (trimmed) so the caller can build the
 * handler's `QueryResult.text` field (see `handler-types.ts`).
 */
export function finalizeResponseText(state: StreamState): string {
  if (state.currentBlockText) {
    state.allResponseText += state.currentBlockText;
    state.currentBlockText = "";
  }
  return state.allResponseText.trim();
}

/**
 * Record that everything accumulated so far has been shipped to the user
 * as a progress message. Call only after the send succeeds.
 */
export function markProgressDelivered(state: StreamState): void {
  state.progressDeliveredLen = state.allResponseText.length;
}

/**
 * The portion of the turn's text that has NOT already been shipped as a
 * mid-turn progress message — i.e. what end-of-turn delivery still owes
 * the user.
 *
 * Backends that never flush progress leave `progressDeliveredLen` at 0,
 * so this is the whole response and their behaviour is unchanged.
 */
export function undeliveredResponseText(state: StreamState): string {
  return state.allResponseText.slice(state.progressDeliveredLen).trim();
}