/**
 * Pi agent session — the live conversation loop.
 *
 * Mirrors the *shape* of the Claude harness loop in `harnesses/claude.ts`:
 *   - one long-lived session per conversation
 *   - user messages arrive via an `AsyncQueue<PiMessage>` input
 *   - the loop drains the queue ONE MESSAGE PER TURN — exactly like the Claude
 *     SDK's input queue: each pushed message gets its own turn, its own
 *     text_end, and its own turn_complete. (An earlier design folded mid-turn
 *     messages into the in-flight turn; that broke the channel manager's
 *     one-response-per-push routing FIFO — see PI-PARITY-AUDIT-2026-06-11.md
 *     D1-1 — so queued messages now simply wait for their own turn.)
 *   - each turn streams provider events back through a single `onEvent`
 *     callback the caller hooked up
 *
 * Each user turn is an inner loop — provider call → if the model asked for
 * tool calls, execute them and feed results back → call provider again — until
 * the model finishes without requesting more tools. Tokens stream live;
 * `text_end` only fires once at the very end of the turn so the UI doesn't
 * display half-answers between tool rounds.
 *
 * Error precedence matches claude (audit D6-2): streamed partial text is
 * always committed via `text_end` (the consumer persists it and the routing
 * FIFO consumes normally); the `error` event fires only when a failed turn
 * produced no text — except fatal kinds (auth / context-overflow), which are
 * surfaced even after partial text so the harness can tear the session down.
 *
 * Auth (key/model/base URL/flavor) is resolved via `getAuth()` on every
 * provider round (audit D6-8): fixing a revoked key or switching models in the
 * wizard applies on the very next round, with full history intact.
 *
 * Sub-agents are NOT spawned here — Bruno will add those later (Phase B).
 */
import { log } from '../../../shared/logger.js';
import type { PiApiFlavor } from './sub-providers.js';
import { streamProvider } from './providers/stream.js';
import type { PiMessage, PiStreamEvent, PiToolDef, PiContentBlock, PiUsage, PiErrorKind } from './providers/types.js';
import { sleep } from './providers/retry.js';
import type { AsyncQueue } from './async-queue.js';
import { findTool } from './tools/registry.js';
import type { PiTool, PiTaskHost } from './tools/types.js';

export type PiSessionEvent =
  | { type: 'turn_started' }
  | { type: 'text_delta'; delta: string }
  | { type: 'text_end'; text: string }
  /** Liveness pulse: the model is reasoning (thinking models) — no text attached. */
  | { type: 'thinking' }
  | { type: 'tool_use'; id: string; name: string; input: any }
  | { type: 'tool_result'; toolUseId: string; name: string; isError?: boolean }
  | {
      type: 'turn_complete';
      usedFileTools: boolean;
      usage?: PiUsage;
      contextWindow?: number;
      /** True when the turn ended on a provider error. NOTE: the `error` EVENT
       *  is suppressed when partial text streamed (D6-2 response-over-error
       *  precedence, designed for the watched parent stream) — these fields
       *  are how an UNwatched consumer (the task host) learns the turn failed,
       *  so a child that streamed text then died isn't reported 'completed'. */
      errored?: boolean;
      errorKind?: PiErrorKind;
      errorMsg?: string;
      /** True when the turn was cut off by the tool-round budget mid-task. */
      roundCapHit?: boolean;
    }
  | { type: 'error'; error: string; kind?: PiErrorKind };

/** Everything the providers need that can change while a session is alive. */
export interface PiSessionAuth {
  flavor: PiApiFlavor;
  modelId: string;
  baseUrl: string;
  apiKey: string;
  /** Per-model output cap from the catalog; providers fall back to safe defaults. */
  maxOutputTokens?: number;
  /** openai-completions only: which field carries the output cap (C-2). */
  maxTokensField?: 'max_tokens' | 'max_completion_tokens';
  /** openai-completions only: false for strict-schema vendors that 422 on stream_options. */
  includeStreamUsage?: boolean;
  /** Model context window from the catalog — reported on turn_complete for the recycler. */
  contextWindow?: number;
  /** False when the catalog says the model is text-only — image blocks are
   *  downgraded to placeholders on send so one screenshot can't 400-poison
   *  the session (audit C-8). Undefined (dynamic models) ⇒ assume vision. */
  supportsImages?: boolean;
}

export interface PiSessionInit {
  /**
   * Resolved on EVERY provider round (not captured once) so wizard-side
   * key/model fixes heal a live conversation on the next round.
   */
  getAuth: () => PiSessionAuth;
  systemPrompt: string;
  /** Pre-loaded history before the first new user turn. */
  initialMessages?: PiMessage[];
  /** Tools the model can call this session. Empty array ⇒ chat-only. */
  tools?: PiToolDef[];
  /** Resolved every time a tool fires (registry → run). */
  cwd: string;
  /**
   * Background sub-agent host (Phase B). Set only on PARENT live sessions —
   * threaded into PiToolContext so the Task tool can spawn; child sessions
   * leave it unset (no grandchildren, Claude SDK parity).
   */
  taskHost?: PiTaskHost;
  /**
   * Per-turn tool-round budget. Parents keep the default; sub-agent children
   * get their agent config's maxTurns (e.g. coder: 50).
   */
  maxToolRounds?: number;
  /** Used to interrupt in-flight provider calls when the session ends. */
  abortController: AbortController;
  /** Caller's event sink — translated to bloby's `bot:*` events one layer up. */
  onEvent: (evt: PiSessionEvent) => void;
}

export interface PiSession {
  /** Resolves when the loop exits (queue closed or aborted). */
  run(input: AsyncQueue<PiMessage>): Promise<void>;
  /** Cumulative history including prefilled context and live turns. */
  getMessages(): PiMessage[];
}

/** Transform-on-send for text-only models (audit C-8): image blocks become
 *  placeholders in the REQUEST only — the stored history keeps the images, so
 *  switching to a vision model later restores them. */
function downgradeImages(messages: PiMessage[]): PiMessage[] {
  let any = false;
  const out = messages.map((m) => {
    if (!m.content.some((b) => b.type === 'image')) return m;
    any = true;
    return {
      ...m,
      content: m.content.map((b): PiContentBlock =>
        b.type === 'image'
          ? { type: 'text', text: '[An image was attached here, but the current model cannot view images. Tell the user to switch to a vision-capable model if the image matters.]' }
          : b,
      ),
    };
  });
  return any ? out : messages;
}

/** Emergency in-turn context relief (audit D2-6): when occupancy crosses the
 *  threshold MID-turn (recycling only acts between idle turns), stub out the
 *  oldest large tool_result payloads — never user/assistant text, never the
 *  protected tail (the current round's results). Cruder than real compaction,
 *  but the turn finishes instead of 400ing on the context wall. */
function trimOldToolResults(messages: PiMessage[], charsToFree: number, protectTail: number): number {
  let freed = 0;
  const limit = Math.max(0, messages.length - protectTail);
  for (let i = 0; i < limit && freed < charsToFree; i++) {
    const m = messages[i];
    if (m.role !== 'user') continue;
    for (const b of m.content) {
      if (b.type === 'tool_result' && typeof b.content === 'string' && b.content.length > 2048) {
        freed += b.content.length;
        b.content = `[tool output trimmed to fit the context window — ~${Math.round(b.content.length / 1024)} KB removed]`;
        if (freed >= charsToFree) break;
      }
    }
  }
  return freed;
}

const ROUND_CAP_NOTICE =
  '[System: the tool budget for this turn is exhausted. Stop working now. In 2-3 sentences, summarize what you completed, what remains, and the exact next step.]';

const FILE_TOOL_NAMES = new Set(['Write', 'Edit', 'MultiEdit', 'NotebookEdit', 'write', 'edit', 'multiEdit', 'notebookEdit']);
const MAX_TOOL_ROUNDS = 25;
/** Transparent re-runs of a failed round that produced nothing (audit D6-1). */
const MAX_ROUND_RETRIES = 2;

export function createPiSession(init: PiSessionInit): PiSession {
  const messages: PiMessage[] = init.initialMessages ? [...init.initialMessages] : [];

  // Last provider-reported usage + window, session-scoped so even an errored
  // turn's turn_complete carries the most recent context occupancy (D2-1).
  let lastUsage: PiUsage | undefined;
  let lastContextWindow: number | undefined;

  // Self-healing vision (audit D rank 12): when a model the catalog couldn't
  // classify (dynamic/unknown sub-providers ⇒ supportsImages undefined) rejects
  // an image with an 'image-unsupported' error, latch this for the rest of the
  // session and downgrade images on every subsequent send. The IMAGE stays in
  // history (downgradeImages is transform-on-send only), so switching to a
  // vision-capable model later restores it.
  let visionDisabled = false;

  /** One stream round — collect the assistant blocks the model emits this pass. */
  interface RoundResult {
    text: string;
    toolUses: { id: string; name: string; input: any; thoughtSignature?: string }[];
    errored: boolean;
    /** Stashed, NOT emitted inline — the turn decides response-vs-error precedence (D6-2). */
    errorMsg?: string;
    errorKind?: PiErrorKind;
    /** True when re-sending the identical round can plausibly succeed (429/5xx/network). */
    retryable?: boolean;
  }

  async function runOneRound(emitSeparatorFirst: boolean, opts?: { wrapUp?: boolean }): Promise<RoundResult> {
    const result: RoundResult = { text: '', toolUses: [], errored: false };
    let firstDelta = true;
    try {
      const auth = init.getAuth();
      lastContextWindow = auth.contextWindow ?? lastContextWindow;
      const stream = streamProvider(auth.flavor, {
        modelId: auth.modelId,
        baseUrl: auth.baseUrl,
        apiKey: auth.apiKey,
        systemPrompt: init.systemPrompt,
        // Downgrade images when the catalog says text-only (supportsImages
        // false) OR a prior round in THIS session learned it the hard way via
        // an 'image-unsupported' error (visionDisabled). The stored history
        // keeps the image so a later vision-capable model still restores it.
        messages: auth.supportsImages === false || visionDisabled ? downgradeImages(messages) : messages,
        tools: init.tools,
        toolChoice: opts?.wrapUp ? 'none' : undefined,
        maxOutputTokens: auth.maxOutputTokens,
        maxTokensField: auth.maxTokensField,
        includeStreamUsage: auth.includeStreamUsage,
        signal: init.abortController.signal,
      });

      for await (const evt of stream as AsyncIterable<PiStreamEvent>) {
        if (init.abortController.signal.aborted) break;
        switch (evt.type) {
          case 'text_delta':
            // Round separator rides BEFORE the new round's first token —
            // claude.ts:374-379 ordering — so the streamed bytes stay a true
            // prefix of the final bot:response even when the dashboard commits
            // the buffer at a tool boundary mid-turn (audit D1-5/PI-SES-1).
            if (firstDelta && emitSeparatorFirst) {
              init.onEvent({ type: 'text_delta', delta: '\n\n' });
            }
            firstDelta = false;
            result.text += evt.delta;
            init.onEvent({ type: 'text_delta', delta: evt.delta });
            break;
          case 'text_end':
            // Sync up with the provider's authoritative concatenation in case
            // we missed a delta. Don't forward — we only emit text_end once
            // at the end of the whole turn so the UI doesn't show half-answers.
            result.text = evt.text;
            break;
          case 'thinking':
            init.onEvent({ type: 'thinking' });
            break;
          case 'tool_use':
            result.toolUses.push({
              id: evt.id,
              name: evt.name,
              input: evt.input,
              thoughtSignature: evt.thoughtSignature,
            });
            // Wrap-up rounds forbid tools (toolChoice 'none'); if a vendor
            // ignores that, swallow the phantom call silently — it is never
            // executed or persisted.
            if (!opts?.wrapUp) {
              init.onEvent({ type: 'tool_use', id: evt.id, name: evt.name, input: evt.input });
            }
            break;
          case 'error':
            result.errored = true;
            result.errorMsg = evt.error;
            result.errorKind = evt.kind;
            result.retryable = evt.retryable;
            break;
          case 'done':
            // Loop control is by tool_use presence, not stop reason — but the
            // usage rides here and feeds the supervisor's session recycling.
            if (evt.usage) lastUsage = evt.usage;
            break;
        }
      }
    } catch (err: any) {
      if (!init.abortController.signal.aborted) {
        result.errored = true;
        result.errorMsg = err?.message || String(err);
        // A throw mid-iteration is a network/stream failure — transient.
        result.errorKind = 'transient';
        result.retryable = true;
      }
    }
    return result;
  }

  async function executeTool(call: { id: string; name: string; input: any }): Promise<{ output: string; isError?: boolean }> {
    const tool: PiTool | undefined = findTool(call.name);
    if (!tool) {
      return {
        output: `Tool not found: ${call.name}. Available tools: ${(init.tools || []).map((t) => t.name).join(', ') || 'none'}.`,
        isError: true,
      };
    }
    try {
      return await tool.run(call.input, { cwd: init.cwd, signal: init.abortController.signal, tasks: init.taskHost });
    } catch (err: any) {
      return { output: `Tool ${call.name} threw: ${err?.message || err}`, isError: true };
    }
  }

  async function runOneTurn(userMsg: PiMessage): Promise<void> {
    if (init.abortController.signal.aborted) return;
    // ONE message per turn — queued messages wait for their own turn so each
    // push gets its own bot:response (routing-FIFO invariant, audit D1-1).
    messages.push(userMsg);
    init.onEvent({ type: 'turn_started' });

    let accumulatedText = '';
    const usedTools = new Set<string>();
    let turnErrored = false;
    let turnErrorMsg: string | undefined;
    let turnErrorKind: PiErrorKind | undefined;

    const maxRounds = Math.max(1, init.maxToolRounds ?? MAX_TOOL_ROUNDS);
    // True only when the for-loop runs out of rounds with the model still
    // mid-task — every intentional exit (done, errored, aborted) clears it.
    let roundCapHit = true;
    for (let round = 0; round < maxRounds; round++) {
      if (init.abortController.signal.aborted) { roundCapHit = false; break; }
      // The separator condition is decided BEFORE the round so the round can
      // emit it ahead of its first token (claude.ts ordering — see runOneRound).
      const needsSeparator = accumulatedText.length > 0 && !accumulatedText.endsWith('\n');
      let res = await runOneRound(needsSeparator);

      // Transparent round retry (D6-1): a transient failure that produced
      // NOTHING is safe to re-run — requests are stateless full-history
      // resends. Never retry a round that already streamed text or tool calls.
      for (
        let attempt = 0;
        attempt < MAX_ROUND_RETRIES &&
        res.errored && res.retryable && !res.text && res.toolUses.length === 0 &&
        !init.abortController.signal.aborted;
        attempt++
      ) {
        log.info(`[pi/session] transient round failure — retrying (${attempt + 1}/${MAX_ROUND_RETRIES}): ${res.errorMsg?.slice(0, 160)}`);
        try { await sleep(1000 * 2 ** attempt, init.abortController.signal); } catch { break; }
        res = await runOneRound(needsSeparator);
      }

      // Self-healing vision (audit D rank 12): a model the catalog couldn't
      // classify just 400/415/422'd on an attached image. Latch visionDisabled
      // and re-run the round ONCE — runOneRound now downgrades images on send,
      // so the resend succeeds. Guarded by !visionDisabled so it fires at most
      // once per session; an image rides every stateless resend, so without
      // this the whole conversation would keep re-400ing.
      if (
        res.errored && res.errorKind === 'image-unsupported' && !visionDisabled &&
        !init.abortController.signal.aborted
      ) {
        log.info('[pi/session] model rejected image — disabling vision for this session and retrying without it');
        visionDisabled = true;
        res = await runOneRound(needsSeparator);
      }

      const { text, toolUses, errored } = res;

      // Append whatever the model produced this round to history so subsequent
      // rounds (and the next user turn) see it.
      const assistantContent: PiContentBlock[] = [];
      if (text) {
        // Matches the separator runOneRound streamed before this round's
        // first delta — accumulatedText and the token stream stay byte-equal.
        if (needsSeparator) accumulatedText += '\n\n';
        accumulatedText += text;
        assistantContent.push({ type: 'text', text });
      }
      if (!errored) {
        // On an errored round, keep the text but DROP the round's tool_use
        // blocks: the turn ends before executing them, and a dangling
        // tool_use with no tool_result poisons the history (Anthropic and
        // Gemini reject the next request outright).
        for (const tu of toolUses) {
          assistantContent.push({
            type: 'tool_use',
            id: tu.id,
            name: tu.name,
            input: tu.input,
            // Forward Gemini's thoughtSignature unchanged so the next turn's
            // request echoes it back; without it the API rejects with 400.
            thoughtSignature: tu.thoughtSignature,
          });
        }
      }
      if (assistantContent.length > 0) {
        messages.push({ role: 'assistant', content: assistantContent });
      }

      if (errored) {
        turnErrored = true;
        turnErrorMsg = res.errorMsg;
        turnErrorKind = res.errorKind;
        roundCapHit = false;
        break;
      }

      // Run every tool the model asked for this round, then feed the results
      // back as a single user message Gemini accepts as a batch.
      const toolResultBlocks: PiContentBlock[] = [];
      for (const tu of toolUses) {
        usedTools.add(tu.name);
        if (init.abortController.signal.aborted) break;
        log.info(`[pi/session] tool call ${tu.name}(${JSON.stringify(tu.input).slice(0, 200)})`);
        const res2 = await executeTool(tu);
        init.onEvent({ type: 'tool_result', toolUseId: tu.id, name: tu.name, isError: !!res2.isError });
        toolResultBlocks.push({
          type: 'tool_result',
          toolUseId: tu.id,
          content: res2.output,
          isError: res2.isError,
        });
      }
      if (toolResultBlocks.length > 0) {
        messages.push({ role: 'user', content: toolResultBlocks });
      }

      // Emergency in-turn context relief (audit D2-6): recycling only acts
      // between idle turns, so a single heavy tool loop could cross the wall
      // mid-turn. Above 85% occupancy, stub the oldest large tool outputs to
      // bring the next request back toward 70%.
      if (lastContextWindow && lastUsage) {
        const occupancy =
          (lastUsage.inputTokens || 0) + (lastUsage.cacheReadTokens || 0) + (lastUsage.cacheCreationTokens || 0);
        if (occupancy > 0.85 * lastContextWindow) {
          const charsToFree = (occupancy - Math.floor(0.7 * lastContextWindow)) * 4; // ~4 chars/token
          const freed = trimOldToolResults(messages, charsToFree, 4);
          if (freed > 0) {
            log.info(`[pi/session] context at ${occupancy}/${lastContextWindow} tok mid-turn — trimmed ~${Math.round(freed / 1024)} KB of old tool output`);
          }
        }
      }

      // No tool calls ⇒ the model is done with this turn.
      if (toolUses.length === 0) { roundCapHit = false; break; }
    }

    // Round-cap wrap-up (audit D5-8): the budget ran out with the model still
    // mid-task. Run ONE final no-tools round so the turn ends with an honest
    // status summary instead of silent truncation. roundCapHit stays true on
    // turn_complete — consumers still know the work is incomplete.
    if (roundCapHit && !turnErrored && !init.abortController.signal.aborted) {
      log.info(`[pi/session] tool-round budget (${maxRounds}) exhausted — running a no-tools wrap-up round`);
      messages.push({ role: 'user', content: [{ type: 'text', text: ROUND_CAP_NOTICE }] });
      const needsSeparator = accumulatedText.length > 0 && !accumulatedText.endsWith('\n');
      const res = await runOneRound(needsSeparator, { wrapUp: true });
      if (res.text) {
        if (needsSeparator) accumulatedText += '\n\n';
        accumulatedText += res.text;
        messages.push({ role: 'assistant', content: [{ type: 'text', text: res.text }] });
      } else {
        // The notice was never answered — pop it so the NEXT turn doesn't
        // open under a stale "stop working now" instruction (review PI-D-1).
        const last = messages[messages.length - 1];
        if (last?.role === 'user' && last.content.length === 1 &&
            last.content[0].type === 'text' && last.content[0].text === ROUND_CAP_NOTICE) {
          messages.pop();
        }
      }
      // Fatal wrap-up failures (dead key / context wall) must still tear the
      // session down, and a cap-hit turn with NO text at all must not end in
      // total silence — claude surfaces error_max_turns and pi's one-shot
      // paths guard this state too (PI-C-2). Set the turn-error fields so the
      // standard emission below handles both (review PI-D-1).
      if (res.errored && (res.errorKind === 'auth' || res.errorKind === 'context-overflow')) {
        turnErrored = true;
        turnErrorMsg = res.errorMsg;
        turnErrorKind = res.errorKind;
      } else if (!accumulatedText) {
        turnErrored = true;
        turnErrorMsg = `I hit my tool budget for this turn (${maxRounds} rounds) before finishing — say "continue" and I'll pick up where I left off.`;
      }
    }

    // Turn-end emission order (audit D6-2, mirrors claude.ts:394-401):
    //   1. text_end whenever ANY text streamed — even on errored turns, so the
    //      partial the user watched is committed, persisted, and consumes its
    //      routing-FIFO entry (the frontend's bot:error handler would
    //      otherwise erase it).
    //   2. error only when the turn produced no text — EXCEPT fatal kinds
    //      (auth / context-overflow), which must surface regardless so the
    //      harness tears the poisoned session down.
    //   3. turn_complete ALWAYS on a non-aborted turn — including errored
    //      paths — so the supervisor clears agentQueryActive. Skipping it
    //      wedged the flag true historically. Aborted turns are torn down via
    //      bot:conversation-ended.
    if (!init.abortController.signal.aborted) {
      if (accumulatedText) {
        init.onEvent({ type: 'text_end', text: accumulatedText });
      }
      const fatal = turnErrorKind === 'auth' || turnErrorKind === 'context-overflow';
      if (turnErrored && (!accumulatedText || fatal)) {
        init.onEvent({ type: 'error', error: turnErrorMsg || 'Provider turn failed', kind: turnErrorKind });
      }
      const usedFileTools = Array.from(usedTools).some((t) => FILE_TOOL_NAMES.has(t));
      init.onEvent({
        type: 'turn_complete',
        usedFileTools,
        usage: lastUsage,
        contextWindow: lastContextWindow,
        errored: turnErrored || undefined,
        errorKind: turnErrorKind,
        errorMsg: turnErrorMsg,
        roundCapHit: roundCapHit || undefined,
      });
    }
  }

  return {
    async run(input) {
      for await (const userMsg of input) {
        if (init.abortController.signal.aborted) break;
        try {
          await runOneTurn(userMsg);
        } catch (err: any) {
          log.warn(`[pi/session] Turn failed: ${err?.message || err}`);
          init.onEvent({ type: 'error', error: err?.message || String(err) });
          // A thrown turn emitted no turn_complete either — clear agentQueryActive so auto-heal
          // and chat aren't wedged. Skip when aborting (teardown emits conversation-ended).
          // usedFileTools=false is the safe default (it only governs whether to auto-restart now).
          if (!init.abortController.signal.aborted) {
            init.onEvent({
              type: 'turn_complete',
              usedFileTools: false,
              usage: lastUsage,
              contextWindow: lastContextWindow,
              errored: true,
              errorMsg: err?.message || String(err),
            });
          }
        }
      }
    },
    getMessages() {
      return messages;
    },
  };
}