/**
 * Bloby pi harness — public surface (mirrors `harnesses/claude.ts`).
 *
 * The dispatcher in `supervisor/bloby-agent.ts` imports this module as
 * `* as pi` and selects it when `cfg.ai.provider === 'pi'`. The surface
 * matches the Claude harness so the dispatcher needs no provider-specific
 * code.
 *
 * Live conversations run the full tool loop (session.ts); one-shots are still
 * tool-less (audit Phase C will route them through createPiSession). The
 * non-blocking feel — user keeps typing while the model is still answering —
 * comes from the same `AsyncQueue` pattern Claude uses (one message per turn);
 * see `async-queue.ts` and PI-PARITY-AUDIT-2026-06-11.md.
 */
import { log } from '../../../shared/logger.js';
import { WORKSPACE_DIR } from '../../../shared/paths.js';
import type { SavedFile } from '../../file-saver.js';
import { assembleSystemPrompt } from '../../../worker/prompts/prompt-assembler.js';
import { buildAgents } from '../../agents/index.js';
import crypto from 'crypto';
import fs from 'fs';
import path from 'path';
import type {
  RecentMessage,
  AgentAttachment,
  AgentQueryRequest,
  AgentQueryResult,
} from '../types.js';
export type { RecentMessage, AgentAttachment };

import { buildSkillsIndex } from '../skills.js';
import {
  routeAttachment,
  buildSavedFilesNote,
  normalizeImageMediaType,
  approxBase64Bytes,
  MAX_INLINE_IMAGE_BYTES,
  INLINE_TEXT_PER_FILE_CHARS,
  INLINE_TEXT_TOTAL_CHARS,
} from '../attachment-policy.js';
import { createAsyncQueue, type AsyncQueue } from './async-queue.js';
import { createPiSession, type PiSessionEvent, type PiSessionAuth } from './session.js';
import { getPiSubProvider, getCatalogModel, type PiApiFlavor } from './sub-providers.js';
import { readPiAuth } from './auth-storage.js';
import type { PiMessage, PiContentBlock } from './providers/types.js';
import { toolDefsForProvider } from './tools/registry.js';
import type { PiTaskHost } from './tools/types.js';

// ── Live conversation state ────────────────────────────────────────────────

interface LiveConversation {
  id: string;
  inputQueue: AsyncQueue<PiMessage>;
  abortController: AbortController;
  onMessage: (type: string, data: any) => void;
  busy: boolean;
  /** Messages pushed but not yet completed (1 turn-complete per message) — mirrors
   *  claude.ts pendingCount. idle:true on turn-complete only when this hits 0, so
   *  the supervisor's session recycling never fires with a message still queued. */
  pendingCount: number;
  /** 60ms micro-batcher for bot:token — collapses per-delta WS frame floods. */
  batcher: TokenBatcher;
  /** Running background sub-agent tasks (Phase B). While non-empty, the
   *  conversation reports idle:false (recycling deferred) and counts as busy
   *  (backend restarts / self-updates deferred) so a task is never killed
   *  mid-flight by housekeeping. */
  tasks: Map<string, RunningTask>;
  /** Set when a completed background task used file tools — OR'd into the next
   *  bot:turn-complete (the continuation turn) so the backend restarts right
   *  after the user hears "Done!", mirroring claude's usedTools capture of
   *  sub-agent tool_use blocks. */
  taskUsedFileTools: boolean;
  /** Origin of each queued message, FIFO-aligned with inputQueue: 'user' for
   *  pushMessage, 'synthetic' for task-completion injections. Shifted on
   *  turn_started so the turn's events can be tagged synthetic — the channel
   *  routing FIFO must skip turns that never enqueued a routing target
   *  (Phase B review PI-B-1: an untagged continuation turn would steal a
   *  concurrently-queued channel message's route). */
  turnOrigins: ('user' | 'synthetic')[];
  /** True while the in-flight turn originated from pushSyntheticMessage. */
  currentTurnSynthetic: boolean;
  loopDone: Promise<void> | null;
}

interface RunningTask {
  id: string;
  description: string;
  subagentType: string;
  abortController: AbortController;
  /** True when stopped via user:stop-task or conversation teardown. */
  stopped: boolean;
  /** True when the wall-clock task watchdog aborted it. */
  timedOut: boolean;
  startedAt: number;
}

/** Hard wall-clock cap per background task. Bounds the housekeeping wedge
 *  (idle:false defers recycling; anyConversationBusy defers restarts and
 *  self-updates) if a child ever hangs in a way the stream timeouts and the
 *  Bash exit-settle can't catch. Generous: a 50-round coder task fits well
 *  inside it. */
const TASK_WALL_CLOCK_MS = 30 * 60_000;

const liveConversations = new Map<string, LiveConversation>();

/**
 * Micro-batch streamed deltas into ~60ms bot:token frames (house standard
 * from the codex parity pass — an order-of-magnitude WS frame reduction with
 * no visible change in streaming feel). Callers MUST flush() before emitting
 * any non-token event so ordering and the streamed-text == bot:response
 * contract are preserved; discard() on teardown drops post-abort stragglers.
 */
interface TokenBatcher {
  add(delta: string): void;
  flush(): void;
  discard(): void;
}

function createTokenBatcher(emit: (text: string) => void, intervalMs = 60): TokenBatcher {
  let buf = '';
  let timer: NodeJS.Timeout | null = null;
  const flush = () => {
    if (timer) { clearTimeout(timer); timer = null; }
    if (buf) { const out = buf; buf = ''; emit(out); }
  };
  return {
    add(delta: string) {
      buf += delta;
      if (!timer) timer = setTimeout(flush, intervalMs);
    },
    flush,
    discard() {
      if (timer) { clearTimeout(timer); timer = null; }
      buf = '';
    },
  };
}

export function hasConversation(conversationId: string): boolean {
  return liveConversations.has(conversationId);
}

export function endAllConversations(): void {
  for (const id of liveConversations.keys()) {
    log.info(`[pi/conversation] Ending conversation ${id} (bulk end)`);
    endConversation(id);
  }
}

// ── Helpers ─────────────────────────────────────────────────────────────────

function readMemoryFile(filename: string): string {
  try {
    const content = fs.readFileSync(path.join(WORKSPACE_DIR, filename), 'utf-8').trim();
    return content || '(empty)';
  } catch {
    return '(empty)';
  }
}

function readMemoryFiles() {
  return {
    myself: readMemoryFile('MYSELF.md'),
    myhuman: readMemoryFile('MYHUMAN.md'),
    memory: readMemoryFile('MEMORY.md'),
    pulse: readMemoryFile('PULSE.json'),
    crons: readMemoryFile('CRONS.json'),
  };
}

function formatConversationHistory(messages: RecentMessage[]): string {
  if (!messages.length) return '';
  return messages.map((m) => `${m.role}: ${m.content}`).join('\n\n');
}

/**
 * Live-conversation pacing hint. The Claude Agent SDK trains its model to do
 * this natively; non-Anthropic models (Gemini especially) tend to go silent
 * during tool loops and never report progress. This nudge makes the
 * conversation feel alive — quick acknowledgement before long tasks, short
 * status notes between tool calls, and inline answers if the user types
 * something while the agent is mid-task.
 */
const LIVE_CONVERSATION_HINT = `

---
# Live-conversation pacing

You are running in a streaming chat where the user can keep typing while you work. Make the conversation feel alive:

- Before kicking off a multi-step task, say one short line acknowledging it ("On it, looking at the widget now.").
- Between tool calls on long tasks, drop a brief progress note ("Found the file, checking the layout next.") so the user knows you're still working.
- Messages the user sends while you're working are queued and delivered to you one at a time after the current task finishes — each gets its own answer, so never assume you missed one.
- Final answers should be concise and concrete.`;

async function buildSystemPrompt(
  names?: { botName: string; humanName: string },
  recentMessages?: RecentMessage[],
): Promise<string> {
  const memoryFiles = readMemoryFiles();
  const basePrompt = await assembleSystemPrompt(names?.botName, names?.humanName, 'pi');
  let systemPrompt = basePrompt;
  systemPrompt += LIVE_CONVERSATION_HINT;
  // Pi has no native skill machinery (Claude's SDK and Codex discover skills
  // themselves), so inject the name+description index here — the agent reads
  // skills/<name>/SKILL.md on demand. Customer-facing supportPrompt runs skip
  // this builder entirely, so they never see the index.
  systemPrompt += buildSkillsIndex();
  systemPrompt += `\n\n---\n# Your Memory Files\n\n## MYSELF.md\n${memoryFiles.myself}\n\n## MYHUMAN.md\n${memoryFiles.myhuman}\n\n## MEMORY.md\n${memoryFiles.memory}\n\n---\n# Your Config Files\n\n## PULSE.json\n${memoryFiles.pulse}\n\n## CRONS.json\n${memoryFiles.crons}`;

  try {
    const { loadConfig: loadCfg } = await import('../../../shared/config.js');
    const cfg = loadCfg();
    const channels = (cfg as any).channels;
    if (channels) {
      systemPrompt += `\n\n---\n# Channel Config\n\`\`\`json\n${JSON.stringify(channels, null, 2)}\n\`\`\``;
    }
  } catch {}

  if (recentMessages?.length) {
    systemPrompt += `\n\n---\n# Recent Conversation\n${formatConversationHistory(recentMessages)}`;
  }
  return systemPrompt;
}

/**
 * Resolve the full provider auth bundle from saved pi-auth.json: sub-provider
 * flavor, base url, api key, model id, plus catalog metadata (per-model output
 * cap, context window) and the sub-provider's max-tokens field quirk.
 *
 * Called at session/one-shot start AND re-called on every live provider round
 * via the session's getAuth thunk — so fixing a revoked key or switching
 * models in the wizard heals a live conversation on its very next round.
 */
function resolveAuth(): { ok: true; auth: PiSessionAuth } | { ok: false; error: string } {
  const saved = readPiAuth();
  if (!saved) return { ok: false, error: 'Bloby provider is not configured. Run the onboarding wizard.' };
  const sub = getPiSubProvider(saved.subProvider);
  if (!sub) return { ok: false, error: `Unknown sub-provider in pi-auth.json: ${saved.subProvider}` };
  const baseUrl = (saved.baseUrl || sub.baseUrl || '').replace(/\/+$/, '');
  if (!baseUrl) return { ok: false, error: `No base URL configured for ${sub.id}` };
  const modelId = saved.modelId || sub.defaultModel || '';
  if (!modelId) return { ok: false, error: `No model selected for ${sub.id}` };
  if (sub.needsApiKey && !saved.apiKey) return { ok: false, error: `Missing API key for ${sub.id}` };
  const catalog = getCatalogModel(sub.id, modelId);

  // Effective window reported to the supervisor's recycler. Two corrections
  // over the raw catalog figure (audit review F1):
  // 1. Anthropic catalog windows can reflect the 1M-context beta; without the
  //    beta header (we don't send it) the real window is 200k.
  // 2. Since every request reserves max_tokens of output budget, providers
  //    enforce input + max_tokens <= window — the usable INPUT ceiling is
  //    window - maxOutputTokens. Reporting the raw window would put the 70%
  //    recycle threshold ABOVE that ceiling (e.g. 140k > 200k-64k=136k on
  //    claude-haiku-4-5) and the recycler could never preempt the wall.
  let contextWindow = catalog?.contextWindow;
  if (contextWindow && sub.flavor === 'anthropic-messages') {
    contextWindow = Math.min(contextWindow, 200_000);
  }
  if (contextWindow && catalog?.maxOutputTokens) {
    contextWindow = Math.max(0, contextWindow - catalog.maxOutputTokens);
  }

  return {
    ok: true,
    auth: {
      flavor: sub.flavor,
      modelId,
      baseUrl,
      apiKey: saved.apiKey || '',
      maxOutputTokens: catalog?.maxOutputTokens,
      maxTokensField: sub.maxTokensField,
      includeStreamUsage: sub.noStreamUsage ? false : undefined,
      contextWindow,
      // Text-only models 400 on image blocks AND the stuck image re-fails
      // every later message (audit C-8) — the session downgrades images to
      // placeholders when the catalog says no vision. Unknown (dynamic
      // sub-providers) ⇒ undefined ⇒ assume vision.
      supportsImages: catalog?.input ? catalog.input.includes('image') : undefined,
    },
  };
}

// ── Background sub-agents (Phase B — audit D4-1) ───────────────────────────

/** Inject a system-originated message into the parent's queue (task completion).
 *  Mirrors the Claude SDK's self-prompted continuation turn, with one
 *  improvement: the resulting turn's events are tagged `synthetic: true`
 *  (origin tracked via conv.turnOrigins) so the channel routing FIFO ignores
 *  them entirely — a continuation enqueues no routing target, and an untagged
 *  bot:response would steal a concurrently-queued channel message's route
 *  (review PI-B-1). Synthetic turns are dashboard-broadcast + DB-persist only.
 *  pendingCount/busy are maintained so idle stays accurate and the recycler
 *  can't fire mid-continuation. No bot:typing (claude parity). */
function pushSyntheticMessage(conv: LiveConversation, text: string): void {
  conv.busy = true;
  conv.pendingCount += 1;
  conv.turnOrigins.push('synthetic');
  conv.inputQueue.push({ role: 'user', content: [{ type: 'text', text }] });
}

/** coder.txt advertises the claude toolset ("Read, Write, Edit, Bash, Glob,
 *  Grep") — swap in the child's REAL pi toolset so the sub-agent never chases
 *  tools it doesn't have (audit D4-4). claude keeps its richer line. */
function rewriteToolAccessLine(prompt: string, toolNames: string[]): string {
  return prompt.replace(/You have full tool access:[^\n]*/i, `You have full tool access: ${toolNames.join(', ')}.`);
}

/** Compact human-readable descriptor of a child tool call for bot:task-progress. */
function toolCallSummary(name: string, input: any): string {
  const tail = (p: any) => (typeof p === 'string' ? p.split('/').slice(-2).join('/') : '');
  switch (name.toLowerCase()) {
    case 'bash': return `Bash: ${String(input?.description || input?.command || '').slice(0, 80)}`;
    case 'read': return `Reading ${tail(input?.file_path)}`;
    case 'write': return `Writing ${tail(input?.file_path)}`;
    case 'edit': return `Editing ${tail(input?.file_path)}`;
    default: return name;
  }
}

/**
 * Per-conversation task host: spawns an in-process child `createPiSession`
 * per Task call, translates child events into the `bot:task-*` vocabulary
 * (payload fields exactly as claude.ts:443-484 emits them), and injects the
 * completion back into the parent's queue for the "Done!" continuation turn.
 */
function createTaskHost(conv: LiveConversation, getAuth: () => PiSessionAuth): PiTaskHost {
  return {
    spawn(req) {
      const agents = buildAgents();
      const cfg = agents[req.subagentType];
      if (!cfg) {
        return {
          ok: false,
          error: `Unknown subagent_type "${req.subagentType}". Available: ${Object.keys(agents).join(', ') || 'none'}.`,
        };
      }

      const taskId = crypto.randomUUID().slice(0, 8);
      const abortController = new AbortController();
      const task: RunningTask = {
        id: taskId,
        description: req.description,
        subagentType: req.subagentType,
        abortController,
        stopped: false,
        timedOut: false,
        startedAt: Date.now(),
      };
      conv.tasks.set(taskId, task);

      // Wall-clock backstop: a hung child would otherwise pin idle:false and
      // anyConversationBusy forever, deferring recycling/restarts/self-updates
      // indefinitely (review PI-B lifecycle finding). Cleared in the finally.
      const watchdog = setTimeout(() => {
        if (!conv.tasks.has(taskId)) return;
        log.warn(`[pi/task] Task ${taskId} hit the ${TASK_WALL_CLOCK_MS / 60_000}-minute wall clock — aborting`);
        task.timedOut = true;
        abortController.abort();
      }, TASK_WALL_CLOCK_MS);

      // Honor the agent config's tool restrictions (claude applies these via
      // the SDK's tools/disallowedTools options — e.g. a future researcher
      // agent with disallowedTools: ['Write','Edit']).
      let childTools = toolDefsForProvider();
      if (Array.isArray(cfg.tools) && cfg.tools.length > 0) {
        childTools = childTools.filter((t) => cfg.tools.includes(t.name));
      }
      if (Array.isArray(cfg.disallowedTools) && cfg.disallowedTools.length > 0) {
        childTools = childTools.filter((t) => !cfg.disallowedTools.includes(t.name));
      }
      const systemPrompt = rewriteToolAccessLine(String(cfg.prompt || ''), childTools.map((t) => t.name));

      let summaryText = '';
      let errorText = '';
      let usedFileTools = false;
      let toolUses = 0;
      // Outcome from the session's turn_complete — the error EVENT is
      // suppressed when partial text streamed (D6-2 precedence), so these are
      // how the host learns a child that streamed text still failed
      // (review PI-B-CHILD-1: such tasks must not be reported 'completed').
      let childErrored = false;
      let childErrorMsg = '';
      let childCapHit = false;
      let lastUsage: { inputTokens?: number; outputTokens?: number; cacheReadTokens?: number; cacheCreationTokens?: number } | undefined;

      const session = createPiSession({
        getAuth,
        systemPrompt,
        tools: childTools,
        cwd: WORKSPACE_DIR,
        abortController,
        maxToolRounds: typeof cfg.maxTurns === 'number' ? cfg.maxTurns : 50,
        onEvent: (evt: PiSessionEvent) => {
          switch (evt.type) {
            case 'tool_use':
              toolUses += 1;
              conv.batcher.flush();
              conv.onMessage('bot:task-progress', {
                conversationId: conv.id,
                taskId,
                summary: toolCallSummary(evt.name, evt.input),
                lastTool: evt.name,
                usage: { tool_uses: toolUses, duration_ms: Date.now() - task.startedAt },
              });
              break;
            case 'text_end':
              summaryText = evt.text;
              break;
            case 'error':
              errorText = evt.error;
              break;
            case 'turn_complete':
              usedFileTools = usedFileTools || evt.usedFileTools;
              if (evt.usage) lastUsage = evt.usage;
              if (evt.errored) {
                childErrored = true;
                childErrorMsg = evt.errorMsg || childErrorMsg;
              }
              if (evt.roundCapHit) childCapHit = true;
              break;
          }
        },
      });

      const queue = createAsyncQueue<PiMessage>();
      queue.push({ role: 'user', content: [{ type: 'text', text: req.prompt }] });
      queue.end();

      log.info(`[pi/task] ──── SUB-AGENT STARTED ──── id=${taskId} type=${req.subagentType} "${req.description}"`);
      // Task events bypass translateAndEmit, so flush the token batcher first —
      // bot:task-created COMMITS the dashboard stream buffer (useBlobyChat),
      // and a batch flushed after it would mis-slice committedTextLength.
      conv.batcher.flush();
      conv.onMessage('bot:task-created', {
        conversationId: conv.id,
        taskId,
        description: req.description,
        type: req.subagentType,
      });

      void (async () => {
        try {
          await session.run(queue);
        } catch (err: any) {
          errorText = errorText || err?.message || String(err);
        } finally {
          clearTimeout(watchdog);
          conv.tasks.delete(taskId);
          // Honest status (review PI-B-CHILD-1): an error or round-cap exit is
          // 'failed' even when partial text streamed — the parent must not
          // relay half-done work as success.
          const failed = task.timedOut || childErrored || childCapHit || !!errorText || !summaryText;
          const status = task.stopped ? 'stopped' : failed ? 'failed' : 'completed';
          let summary = summaryText || errorText || '(the agent produced no output)';
          if (task.timedOut) {
            summary += `\n\n[The task was aborted after ${TASK_WALL_CLOCK_MS / 60_000} minutes — work is incomplete.]`;
          } else if (childCapHit) {
            summary += '\n\n[The task hit its tool-round limit before finishing — work may be incomplete.]';
          } else if (childErrored && summaryText) {
            summary += `\n\n[The task hit an error before finishing: ${childErrorMsg || errorText || 'unknown error'}]`;
          }
          const u = lastUsage;
          const totalTokens = u
            ? (u.inputTokens || 0) + (u.outputTokens || 0) + (u.cacheReadTokens || 0) + (u.cacheCreationTokens || 0)
            : 0;
          log.info(
            `[pi/task] ──── SUB-AGENT ${status.toUpperCase()} ──── id=${taskId} ` +
            `tools=${toolUses} ${Math.round((Date.now() - task.startedAt) / 1000)}s summary=${summary.slice(0, 160)}`,
          );
          conv.batcher.flush();
          conv.onMessage('bot:task-done', {
            conversationId: conv.id,
            taskId,
            status,
            summary,
            usage: { tool_uses: toolUses, duration_ms: Date.now() - task.startedAt, total_tokens: totalTokens },
          });
          if (usedFileTools) conv.taskUsedFileTools = true;

          // Drive the user-facing continuation turn — unless the conversation
          // itself is gone (ended/recycled), in which case the report dies with
          // it (claude parity: the SDK subprocess dies too).
          if (liveConversations.get(conv.id) === conv && !conv.abortController.signal.aborted) {
            const note = task.stopped
              ? `[System: the background task "${req.description}" was stopped by the user. Acknowledge that briefly in your own voice — never mention agents, tasks, or system messages.]`
              : `[System: background task "${req.description}" ${status}.]\n\nResult summary:\n${summary}\n\nRelay the outcome to the user concisely in your own voice (never mention agents, tasks, ids, or system messages). If it failed, say what went wrong and offer a next step.`;
            pushSyntheticMessage(conv, note);
          }
        }
      })();

      return { ok: true, taskId };
    },
  };
}

/** Convert a saved RecentMessage[] into the provider-neutral PiMessage[]. */
function recentToPiMessages(messages: RecentMessage[] | undefined): PiMessage[] {
  if (!messages?.length) return [];
  return messages.map((m) => ({
    role: m.role === 'assistant' ? 'assistant' : 'user',
    content: [{ type: 'text', text: m.content }],
  }));
}

/** Native PDF document blocks reach only the flavors that render them — the
 *  Anthropic Messages API and Gemini both ingest application/pdf inline
 *  (base64 document source / inlineData). openai-completions has no document
 *  type, so a PDF there falls back to the saved-files disk pointer. Matches the
 *  shared attachment-policy routing rule. */
function canNativeDocumentForFlavor(flavor: PiApiFlavor): boolean {
  return flavor === 'anthropic-messages' || flavor === 'google-gemini';
}

/** Build a PiContentBlock[] from raw text + attachments, MEDIA-FIRST then the
 *  prompt text last (parity with claude.ts and the other pi providers). Routing
 *  is delegated to the shared attachment-policy so all three harnesses ingest
 *  identically; canNativeDocument is the active provider's PDF capability. */
function buildAttachmentBlocks(
  text: string,
  canNativeDocument: boolean,
  attachments?: AgentAttachment[],
  savedFiles?: SavedFile[],
): PiContentBlock[] {
  const content: PiContentBlock[] = [];
  if (attachments?.length) {
    // Running budget so the cross-file inline-text total never exceeds the cap.
    let inlineTextBudget = INLINE_TEXT_TOTAL_CHARS;
    for (const att of attachments) {
      switch (routeAttachment(att, { canNativeDocument })) {
        case 'image': {
          // Drop the inline copy when it would bloat every stateless resend —
          // the file is on disk and buildSavedFilesNote points the tools at it.
          if (approxBase64Bytes(att.data) > MAX_INLINE_IMAGE_BYTES) break;
          content.push({ type: 'image', mediaType: normalizeImageMediaType(att.mediaType), data: att.data });
          break;
        }
        case 'native-document': {
          // PDF on a flavor that renders it natively (anthropic / gemini).
          content.push({ type: 'document', mediaType: 'application/pdf', data: att.data, name: att.name });
          break;
        }
        case 'inline-text': {
          if (inlineTextBudget <= 0) break;
          let decoded = '';
          try { decoded = Buffer.from(att.data, 'base64').toString('utf-8'); }
          catch { break; } // undecodable → rely on the saved-files note
          const cap = Math.min(INLINE_TEXT_PER_FILE_CHARS, inlineTextBudget);
          const slice = decoded.slice(0, cap);
          inlineTextBudget -= slice.length;
          content.push({ type: 'text', text: `--- ${att.name} ---\n${slice}` });
          break;
        }
        case 'reference-only':
        default:
          // Binary we can't inline (docx/xlsx/zip/…), a PDF on a flavor without
          // native documents, or an unexpected route — no provider block; the
          // saved-files note below carries the disk pointer. Never emit a
          // malformed block (defensive default, review PI-E).
          break;
      }
    }
  }

  let prompt = text || '(attached files)';
  if (savedFiles?.length) {
    const note = buildSavedFilesNote(savedFiles);
    if (note) prompt += `\n\n${note}`;
  }
  content.push({ type: 'text', text: prompt });
  return content;
}

/** Wrap a raw user input into a PiMessage with text + optional media blocks. */
function buildUserMessage(
  text: string,
  canNativeDocument: boolean,
  attachments?: AgentAttachment[],
  savedFiles?: SavedFile[],
): PiMessage {
  return { role: 'user', content: buildAttachmentBlocks(text, canNativeDocument, attachments, savedFiles) };
}

// ── Live Conversation API ──────────────────────────────────────────────────

export async function startConversation(
  conversationId: string,
  _model: string,
  onMessage: (type: string, data: any) => void,
  names?: { botName: string; humanName: string },
  recentMessages?: RecentMessage[],
): Promise<boolean> {
  log.info(`[pi/conversation] ──── STARTING CONVERSATION ────`);
  log.info(`[pi/conversation] Conv ID: ${conversationId}`);

  if (liveConversations.has(conversationId)) {
    log.info(`[pi/conversation] Ending existing conversation ${conversationId} before starting new one`);
    endConversation(conversationId);
  }

  const resolved = resolveAuth();
  if (!resolved.ok) {
    log.warn(`[pi/conversation] Cannot start: ${resolved.error}`);
    onMessage('bot:error', { conversationId, error: resolved.error });
    return false;
  }

  log.info(`[pi/conversation] Sub-provider: ${resolved.auth.flavor} · model: ${resolved.auth.modelId}`);

  const systemPrompt = await buildSystemPrompt(names, recentMessages);
  log.info(`[pi/conversation] System prompt: ${systemPrompt.length} chars`);

  const inputQueue = createAsyncQueue<PiMessage>();
  const abortController = new AbortController();

  const conv: LiveConversation = {
    id: conversationId,
    inputQueue,
    abortController,
    onMessage,
    busy: false,
    pendingCount: 0,
    batcher: null as unknown as TokenBatcher, // set right below (needs conv for the synthetic tag)
    tasks: new Map(),
    taskUsedFileTools: false,
    turnOrigins: [],
    currentTurnSynthetic: false,
    loopDone: null,
  };
  conv.batcher = createTokenBatcher((text) =>
    onMessage('bot:token', {
      conversationId,
      token: text,
      ...(conv.currentTurnSynthetic ? { synthetic: true } : {}),
    }),
  );
  liveConversations.set(conversationId, conv);

  // Re-resolve auth on every provider round so a key/model fix in the wizard
  // applies to the next round with full history intact (audit D6-8). Falls
  // back to the last good bundle if pi-auth.json turns unreadable mid-session.
  let currentAuth: PiSessionAuth = resolved.auth;
  const getAuth = (): PiSessionAuth => {
    const fresh = resolveAuth();
    if (fresh.ok) currentAuth = fresh.auth;
    return currentAuth;
  };

  const session = createPiSession({
    getAuth,
    systemPrompt,
    tools: toolDefsForProvider({ withTask: true }),
    cwd: WORKSPACE_DIR,
    abortController,
    taskHost: createTaskHost(conv, getAuth),
    onEvent: (evt: PiSessionEvent) => {
      translateAndEmit(conv, evt);
    },
  });

  conv.loopDone = (async () => {
    try {
      await session.run(inputQueue);
      log.info(`[pi/conversation] ──── QUERY LOOP ENDED ────`);
    } catch (err: any) {
      if (!abortController.signal.aborted) {
        log.warn(`[pi/conversation] Loop error: ${err?.message || err}`);
        onMessage('bot:error', { conversationId, error: err?.message || String(err) });
      }
    } finally {
      log.info(`[pi/conversation] Cleaning up conversation ${conversationId}`);
      // Drop any unflushed token stragglers — at teardown the turn is either
      // complete (already flushed before turn_complete) or aborted (tokens
      // from an aborted stream must not surface after the fact).
      conv.batcher.discard();
      liveConversations.delete(conversationId);
      onMessage('bot:conversation-ended', { conversationId });
    }
  })();

  return true;
}

/** Map session-level events back into bloby's `bot:*` vocabulary. */
function translateAndEmit(conv: LiveConversation, evt: PiSessionEvent) {
  if (evt.type === 'text_delta') {
    conv.batcher.add(evt.delta);
    return;
  }
  // Any non-token event flushes the batch first — ordering (tokens before the
  // tool chip / final response) and the streamed-text == bot:response
  // invariant both depend on it.
  conv.batcher.flush();

  // Synthetic tag for every event of a continuation turn (origin shifted at
  // turn_started): the channel routing FIFO must treat these turns as
  // invisible — they enqueued no routing target, so consuming one would steal
  // a queued channel message's route (review PI-B-1).
  const syn = conv.currentTurnSynthetic ? { synthetic: true } : {};

  switch (evt.type) {
    case 'turn_started':
      conv.currentTurnSynthetic = conv.turnOrigins.shift() === 'synthetic';
      // No bloby event for this — `bot:typing` is already emitted by pushMessage().
      break;
    case 'text_end':
      conv.onMessage('bot:response', { conversationId: conv.id, content: evt.text, ...syn });
      break;
    case 'tool_use': {
      // House vocabulary: claude's delegation tool is named Task; the pi
      // prompt's 'Agent' alias resolves to the same tool — normalize the
      // event so consumers see one name.
      const toolName = evt.name === 'Agent' || evt.name === 'agent' ? 'Task' : evt.name;
      conv.onMessage('bot:tool', { conversationId: conv.id, name: toolName, input: evt.input, ...syn });
      break;
    }
    case 'thinking':
      // Reasoning-model liveness pulse (house standard, codex M1 analog) —
      // the UI dedups repeated name+running entries, channels get a chunk
      // flush opportunity. Reasoning TEXT is never forwarded.
      conv.onMessage('bot:tool', { conversationId: conv.id, name: 'thinking', status: 'running', ...syn });
      break;
    case 'tool_result':
      // Progress pulse between tool rounds (audit D1-7): claude punctuates
      // long tasks with tool_progress events; this is pi's equivalent —
      // commits dashboard bubbles and flushes channel chunks mid-task.
      conv.onMessage('bot:tool', { conversationId: conv.id, name: evt.name, status: 'running', ...syn });
      break;
    case 'turn_complete': {
      conv.busy = false;
      // One turn-complete per pushed message (D1-1 restored that invariant);
      // idle gates the supervisor's proactive recycling so it never fires with
      // a message still queued OR a background task still running — recycling
      // mid-task would kill the task (claude has the same teardown semantics,
      // but its idle flag doesn't guard tasks; this is strictly safer).
      conv.pendingCount = Math.max(0, conv.pendingCount - 1);
      const idle = conv.pendingCount === 0 && conv.tasks.size === 0;
      // A finished background task's file edits restart the backend on the
      // very next turn boundary (the continuation turn) — claude captures
      // sub-agent tool_use blocks into the parent's usedTools the same way.
      const usedFileTools = evt.usedFileTools || conv.taskUsedFileTools;
      conv.taskUsedFileTools = false;
      // Prompt occupancy of the last provider round — input + cache reads +
      // cache writes, exactly claude.ts's contextTokens math. Output tokens
      // are NOT added (claude doesn't either; the recycler's 70% threshold
      // absorbs the next-turn growth).
      const contextTokens = evt.usage
        ? (evt.usage.inputTokens || 0) + (evt.usage.cacheReadTokens || 0) + (evt.usage.cacheCreationTokens || 0)
        : 0;
      conv.onMessage('bot:turn-complete', {
        conversationId: conv.id,
        usedFileTools,
        contextTokens,
        contextWindow: evt.contextWindow || 0,
        idle,
        ...syn,
      });
      log.info(`[pi/conversation] ──── TURN COMPLETE ────  busy=false ctx=${contextTokens}/${evt.contextWindow || 'n/a'} idle=${idle} tasks=${conv.tasks.size}`);
      break;
    }
    case 'error': {
      // busy is NOT cleared here (audit D1-9): turn_complete is the single
      // busy=false site and the session guarantees it on every non-aborted
      // turn; an aborted/fatal path is torn down via bot:conversation-ended.
      const fatal = evt.kind === 'auth' || evt.kind === 'context-overflow';
      const remedy = evt.kind === 'context-overflow'
        ? ' Starting a fresh session — send your message again to continue.'
        : evt.kind === 'auth'
          ? ' I\'ll reconnect with the new key as soon as it\'s saved.'
          : '';
      conv.onMessage('bot:error', { conversationId: conv.id, error: `${evt.error}${remedy}`, ...syn });
      if (fatal) {
        // Unrecoverable for this session (audit D6-4): an over-window history
        // would re-fail on every future turn, and a dead key has no business
        // keeping the loop alive. Tear down — the finally emits
        // bot:conversation-ended (routes + flags clear) and the next user
        // message cold-starts a fresh session with re-injected history.
        log.warn(`[pi/conversation] Fatal provider error (${evt.kind}) — recycling session ${conv.id}`);
        endConversation(conv.id);
      }
      break;
    }
  }
}

export function pushMessage(
  conversationId: string,
  content: string,
  attachments?: AgentAttachment[],
  savedFiles?: SavedFile[],
): boolean {
  const conv = liveConversations.get(conversationId);
  if (!conv) {
    log.warn(`[pi/conversation] pushMessage — no live conversation ${conversationId}`);
    return false;
  }

  log.info(`[pi/conversation] ──── PUSH MESSAGE ──── busy=${conv.busy} pending=${conv.pendingCount + 1}`);
  conv.busy = true;
  conv.pendingCount += 1;
  conv.turnOrigins.push('user');
  // Resolve the active flavor at push time (the session re-resolves auth every
  // round, so a wizard provider switch mid-session is honored). Unreadable auth
  // ⇒ no native documents — the conservative route sends a PDF to the disk
  // pointer rather than emitting a block the provider can't render.
  const resolved = resolveAuth();
  const canNativeDocument = resolved.ok ? canNativeDocumentForFlavor(resolved.auth.flavor) : false;
  conv.inputQueue.push(buildUserMessage(content, canNativeDocument, attachments, savedFiles));
  conv.onMessage('bot:typing', { conversationId });
  return true;
}

export function endConversation(conversationId: string): void {
  const conv = liveConversations.get(conversationId);
  if (!conv) return;

  log.info(`[pi/conversation] ──── ENDING CONVERSATION ${conversationId} ────`);
  // Background tasks die with the conversation (claude parity — the SDK
  // subprocess takes its tasks down too). Their finallys still emit
  // bot:task-done {status:'stopped'} so dashboard task cards don't spin
  // forever; the completion injection is skipped (conv gone).
  for (const task of conv.tasks.values()) {
    task.stopped = true;
    task.abortController.abort();
  }
  conv.batcher.discard();
  conv.inputQueue.end();
  conv.abortController.abort();
  liveConversations.delete(conversationId);
}

export function isConversationBusy(conversationId: string): boolean {
  return liveConversations.get(conversationId)?.busy || false;
}

/** True if ANY live conversation in this harness is mid-turn OR has a background
 *  sub-agent running. Used by the supervisor to defer backend restarts and
 *  self-updates — a restart mid-task would kill the task's work in flight. */
export function anyConversationBusy(): boolean {
  for (const c of liveConversations.values()) {
    if (c.busy || c.tasks.size > 0) return true;
  }
  return false;
}

/** Stop a specific background sub-agent task (dashboard user:stop-task). The
 *  child's teardown emits bot:task-done {status:'stopped'} and injects a brief
 *  acknowledgement turn into the parent. */
export async function stopSubAgentTask(conversationId: string, taskId: string): Promise<void> {
  const conv = liveConversations.get(conversationId);
  const task = conv?.tasks.get(taskId);
  if (!task) {
    log.warn(`[pi/task] Cannot stop task ${taskId} — not running in conversation ${conversationId}`);
    return;
  }
  log.info(`[pi/task] Stopping sub-agent task ${taskId}`);
  task.stopped = true;
  task.abortController.abort();
}

/** Pi has no pre-warm step (no subprocess), but the interface requires this. */
export async function warmUpForLiveConversation(
  _model: string,
  _names?: { botName: string; humanName: string },
): Promise<void> {
  // no-op
}

// ── One-shot API (customer WhatsApp, scheduler, /api/agent/query) ──────────

const activeQueries = new Map<string, AbortController>();

/** True while any one-shot startBlobyAgentQuery is in flight (cleared in a finally). These don't
 *  register as live conversations, so anyConversationBusy() can't see them. */
export function anyOneShotActive(): boolean {
  return activeQueries.size > 0;
}

/**
 * One-shot agentic query — used by customer WhatsApp + scheduler (pulse/cron).
 *
 * Phase C (audit D5-1/D3-1): runs the SAME tool loop as the live path — a
 * single-message `createPiSession` — so pulse/cron runs can actually edit
 * files, run Bash, and read skills, and the tool-advertising system prompt is
 * finally true (a tool-less request under that prompt made Gemini emit
 * MALFORMED_FUNCTION_CALL — PI-HARNESS.md gotcha #3). No task host: background
 * sub-agents stay a live-conversation feature (claude parity), so the Task def
 * is excluded from the tool list and a hallucinated call fails gracefully.
 *
 * Guarantees preserved: finally-emitted bot:done, 5-min non-resetting
 * watchdog, activeQueries registration AFTER the awaited prompt build
 * (leak-ordering, claude.ts), supportPrompt bypasses the owner prompt +
 * skills index entirely.
 */
export async function startBlobyAgentQuery(
  conversationId: string,
  prompt: string,
  _model: string,
  onMessage: (type: string, data: any) => void,
  attachments?: AgentAttachment[],
  savedFiles?: SavedFile[],
  names?: { botName: string; humanName: string },
  recentMessages?: RecentMessage[],
  supportPrompt?: string,
  maxTurns?: number,
): Promise<void> {
  const resolved = resolveAuth();
  if (!resolved.ok) {
    onMessage('bot:error', { conversationId, error: resolved.error });
    // bot:done frees the caller's slot (WhatsApp activeAgents / scheduler) — without it
    // each distinct customer hitting this path pins one of the 5 concurrent slots until
    // supervisor restart (audit D3-2; mirrors claude.ts:620).
    onMessage('bot:done', { conversationId, usedFileTools: false });
    return;
  }

  // Build the prompt BEFORE registering in activeQueries / arming the watchdog
  // (claude.ts ordering): if anything in here ever rejected after registration,
  // the entry would leak forever — anyOneShotActive() stuck true defers every
  // backend restart/self-update, and the caller's slot never frees.
  let systemPrompt: string;
  if (supportPrompt) {
    systemPrompt = supportPrompt;
  } else {
    // History rides ONLY as structured messages (initialMessages below).
    // Passing it here too duplicated every prior turn into the system prompt
    // (audit D3-6).
    systemPrompt = await buildSystemPrompt(names, undefined);
    // The base prompt routes heavy coding to the Agent tool, which only LIVE
    // conversations have (one-shots have no task host) — keep the model
    // honest so it doesn't chase a tool that isn't declared (review PI-C-4).
    systemPrompt +=
      '\n\n---\n# One-shot run\nThis is a scheduled/one-shot run: the Agent tool is NOT available here. ' +
      'Do any heavy work yourself, directly with Read, Write, Edit, and Bash.';
  }

  const abortController = new AbortController();
  activeQueries.set(conversationId, abortController);
  // Hard watchdog — a hung turn would otherwise pin this query forever (finally never
  // runs, bot:done never fires). Abort after 5 min; cleared in the finally on normal completion.
  const watchdog = setTimeout(() => {
    log.warn(`[pi/bloby-agent] one-shot timed out (5m) — aborting conv=${conversationId}`);
    abortController.abort();
  }, 300_000);

  onMessage('bot:typing', { conversationId });

  let usedFileTools = false;
  // Track tool names LIVE (not only via turn_complete): an aborted run never
  // emits turn_complete, and files written in earlier rounds must still flag
  // usedFileTools on bot:done or the backend serves stale code
  // (review PI-C-1; mirrors claude.ts:723-760 and runAgentQuery below).
  const usedTools = new Set<string>();
  let sawResponse = false;
  let capHit = false;
  const batcher = createTokenBatcher((text) => onMessage('bot:token', { conversationId, token: text }));

  // Re-resolve auth per round, same as the live path — a key/model fix in the
  // wizard applies to the next round of an in-flight pulse run too.
  let currentAuth: PiSessionAuth = resolved.auth;
  const getAuth = (): PiSessionAuth => {
    const fresh = resolveAuth();
    if (fresh.ok) currentAuth = fresh.auth;
    return currentAuth;
  };

  try {
    const session = createPiSession({
      getAuth,
      systemPrompt,
      initialMessages: recentToPiMessages(recentMessages),
      tools: toolDefsForProvider(), // no Task — one-shots have no task host
      cwd: WORKSPACE_DIR,
      abortController,
      maxToolRounds: maxTurns ?? 50, // claude one-shot default (claude.ts:677)
      onEvent: (evt: PiSessionEvent) => {
        switch (evt.type) {
          case 'text_delta':
            batcher.add(evt.delta);
            break;
          case 'text_end':
            // Session precedence (D6-2): emitted even on errored turns when
            // partial text streamed — the partial reaches the customer/pulse.
            batcher.flush();
            sawResponse = true;
            onMessage('bot:response', { conversationId, content: evt.text });
            break;
          case 'tool_use': {
            batcher.flush();
            usedTools.add(evt.name);
            const toolName = evt.name === 'Agent' || evt.name === 'agent' ? 'Task' : evt.name;
            onMessage('bot:tool', { conversationId, name: toolName, input: evt.input });
            break;
          }
          case 'thinking':
            batcher.flush();
            onMessage('bot:tool', { conversationId, name: 'thinking', status: 'running' });
            break;
          case 'tool_result':
            batcher.flush();
            onMessage('bot:tool', { conversationId, name: evt.name, status: 'running' });
            break;
          case 'error':
            // Fires only when the turn produced no text, or fatally (D6-2).
            batcher.flush();
            sawResponse = true; // the caller got a terminal signal for this turn
            onMessage('bot:error', { conversationId, error: evt.error });
            break;
          case 'turn_complete':
            usedFileTools = usedFileTools || evt.usedFileTools;
            if (evt.roundCapHit) capHit = true;
            break;
        }
      },
    });

    const queue = createAsyncQueue<PiMessage>();
    queue.push(buildUserMessage(prompt, canNativeDocumentForFlavor(resolved.auth.flavor), attachments, savedFiles));
    queue.end();
    await session.run(queue);

    // Round-cap exhaustion with no terminal signal: the model was still
    // mid-task when the budget ran out and no text streamed — without this the
    // customer/pulse gets dead silence (review PI-C-2; claude surfaces an
    // error_max_turns result on the same path).
    if (!abortController.signal.aborted && capHit && !sawResponse) {
      batcher.flush();
      onMessage('bot:error', {
        conversationId,
        error: `The run hit its ${maxTurns ?? 50}-round tool limit before producing a reply. Try a narrower request.`,
      });
    }
  } catch (err: any) {
    // session.run contains per-turn error handling; a throw here is unexpected.
    if (!abortController.signal.aborted) {
      log.warn(`[pi/bloby-agent] one-shot error: ${err?.message || err}`);
      batcher.flush();
      onMessage('bot:error', { conversationId, error: err?.message || String(err) });
    }
  } finally {
    // Aborted-run stragglers must not surface (audit D3-8) — discard, never flush.
    batcher.discard();
    clearTimeout(watchdog);
    activeQueries.delete(conversationId);
    // Live tool tracking covers aborted runs whose turn_complete never fired —
    // files already written must still trigger the backend restart (PI-C-1).
    const fileToolsUsed = usedFileTools || ['Write', 'Edit', 'write', 'edit'].some((t) => usedTools.has(t));
    onMessage('bot:done', { conversationId, usedFileTools: fileToolsUsed });
  }
}

export function stopBlobyAgentQuery(conversationId: string): void {
  const ctl = activeQueries.get(conversationId);
  if (ctl) {
    ctl.abort();
    activeQueries.delete(conversationId);
  }
}

// ── Workspace agent endpoint (POST /api/agent/query) ──────────────────────

/** Minimal coding-agent prompt for /api/agent/query when the caller supplies
 *  none — claude falls back to its native `claude_code` preset; pi's
 *  equivalent advertises ONLY the tools that actually exist, and never the
 *  Bloby owner persona (agent-API callers are workspace apps, not the bot). */
const PI_CODING_AGENT_PROMPT =
  'You are a coding agent operating non-interactively inside a project workspace. ' +
  'Complete the request fully using your tools, then reply with a concise summary of what you did. ' +
  'Tools: Read (file contents), Write (create/overwrite a file), Edit (exact string replacement), ' +
  'Bash (shell commands; cwd is the workspace root). Paths are relative to the workspace root. ' +
  'Do the work — never claim to have done something without actually using the tools.';

/** In-memory session store for the agent API (audit D2-7/D3-3). Process-
 *  lifetime only — AGENT-API.md documents that sessions die on supervisor
 *  restart, and claude's resume has the same practical bound. */
interface StoredAgentSession { messages: PiMessage[]; lastUsed: number }
const agentSessions = new Map<string, StoredAgentSession>();
const AGENT_SESSION_CAP = 50;
const AGENT_SESSION_TTL_MS = 24 * 60 * 60_000;
const AGENT_SESSION_MAX_MESSAGES = 40;

function sweepAgentSessions(): void {
  const now = Date.now();
  for (const [id, s] of agentSessions) {
    if (now - s.lastUsed > AGENT_SESSION_TTL_MS) agentSessions.delete(id);
  }
  if (agentSessions.size > AGENT_SESSION_CAP) {
    const byAge = [...agentSessions.entries()].sort((a, b) => a[1].lastUsed - b[1].lastUsed);
    for (const [id] of byAge.slice(0, agentSessions.size - AGENT_SESSION_CAP)) {
      agentSessions.delete(id);
    }
  }
}

/** Trim resumed history at a clean turn boundary: the window must start on a
 *  REAL user message (not a tool_result carrier) — an orphaned tool_result or
 *  a leading assistant message makes Anthropic/Gemini reject the request. */
function trimAgentHistory(messages: PiMessage[]): PiMessage[] {
  if (messages.length <= AGENT_SESSION_MAX_MESSAGES) return messages;
  const isRealUser = (m: PiMessage) =>
    m.role === 'user' && !m.content.some((b) => b.type === 'tool_result');
  const windowStart = messages.length - AGENT_SESSION_MAX_MESSAGES;
  for (let i = windowStart; i < messages.length; i++) {
    if (isRealUser(messages[i])) return messages.slice(i);
  }
  // No clean boundary inside the window — a single tool-heavy turn (each round
  // adds an assistant + a tool_result message) exceeds the cap by itself.
  // Fall back BACKWARD to that turn's own user message: the window overshoots
  // the cap (bounded by the turn's size) instead of silently wiping the whole
  // history to [] (review PI-C-3 — total session amnesia).
  for (let j = windowStart - 1; j >= 0; j--) {
    if (isRealUser(messages[j])) return messages.slice(j);
  }
  return messages;
}

/** Per-sessionId serialization (review PI-C-SESS-2): two concurrent resumes of
 *  the same session would both read the same stored history and last-write-win
 *  the store, silently erasing one call's turn. Chaining the second behind the
 *  first keeps the linear-history contract; each run is bounded by its own
 *  timeout (≤300s), so the wait is too. */
const agentSessionLocks = new Map<string, Promise<void>>();

export async function runAgentQuery(req: AgentQueryRequest): Promise<AgentQueryResult> {
  if (!req.sessionId) return runAgentQueryInner(req);
  const id = req.sessionId;
  const prev = agentSessionLocks.get(id) ?? Promise.resolve();
  let release!: () => void;
  const gate = new Promise<void>((r) => { release = r; });
  const chained = prev.then(() => gate);
  agentSessionLocks.set(id, chained);
  await prev;
  try {
    return await runAgentQueryInner(req);
  } finally {
    release();
    if (agentSessionLocks.get(id) === chained) agentSessionLocks.delete(id);
  }
}

async function runAgentQueryInner(req: AgentQueryRequest): Promise<AgentQueryResult> {
  const resolved = resolveAuth();
  if (!resolved.ok) return { ok: false, error: resolved.error };

  const timeout = Math.min(Math.max(req.timeout || 120_000, 5_000), 300_000);
  // Same clamp as claude.ts:781 — maxTurns maps onto the session's tool-round budget.
  const maxTurns = Math.min(Math.max(req.maxTurns || 25, 1), 50);
  const abortController = new AbortController();
  const timeoutHandle = setTimeout(() => abortController.abort(), timeout);

  const systemPrompt = req.systemPrompt?.trim() ? req.systemPrompt : PI_CODING_AGENT_PROMPT;

  sweepAgentSessions();
  const resumed = req.sessionId ? agentSessions.get(req.sessionId) : undefined;
  const sessionId = resumed ? req.sessionId! : crypto.randomUUID();
  if (resumed) resumed.lastUsed = Date.now();

  let fullText = '';
  const usedTools = new Set<string>();
  let errored = false;
  let errorMsg = '';
  let usedFileTools = false;
  let capHit = false;

  let currentAuth: PiSessionAuth = resolved.auth;
  const getAuth = (): PiSessionAuth => {
    const fresh = resolveAuth();
    if (fresh.ok) currentAuth = fresh.auth;
    return currentAuth;
  };

  const session = createPiSession({
    getAuth,
    systemPrompt,
    initialMessages: resumed ? trimAgentHistory(resumed.messages) : undefined,
    tools: toolDefsForProvider(), // no Task — no task host on this path
    cwd: WORKSPACE_DIR,
    abortController,
    maxToolRounds: maxTurns,
    onEvent: (evt: PiSessionEvent) => {
      switch (evt.type) {
        case 'text_end':
          fullText = evt.text;
          break;
        case 'tool_use':
          usedTools.add(evt.name);
          break;
        case 'error':
          errored = true;
          errorMsg = evt.error;
          break;
        case 'turn_complete':
          usedFileTools = usedFileTools || evt.usedFileTools;
          // The error EVENT is suppressed when partial text streamed (D6-2) —
          // read the outcome fields so a failed turn isn't reported clean.
          if (evt.errored) {
            errored = true;
            errorMsg = errorMsg || evt.errorMsg || '';
          }
          if (evt.roundCapHit) capHit = true;
          break;
      }
    },
  });

  try {
    log.info(`[pi/agent-api] Query: msg="${req.message.slice(0, 80)}..." maxTurns=${maxTurns} timeout=${timeout}ms resume=${resumed ? sessionId : 'none'}`);
    const queue = createAsyncQueue<PiMessage>();
    queue.push({ role: 'user', content: [{ type: 'text', text: req.message }] });
    queue.end();
    await session.run(queue);
  } catch (err: any) {
    if (abortController.signal.aborted) return { ok: false, error: 'Query timed out.', sessionId };
    return { ok: false, error: err?.message || String(err), sessionId };
  } finally {
    clearTimeout(timeoutHandle);
  }

  if (abortController.signal.aborted) {
    // Timed-out histories can hold a dangling tool_use (aborted mid-round) —
    // don't persist them for resume.
    return { ok: false, error: 'Query timed out.', sessionId };
  }

  // Round-cap exhaustion with no answer: the model was still mid-task when the
  // budget ran out (claude maps the same state to an error_max_turns result —
  // review PI-C-2; an ok:true empty response reads as a silent blank bubble in
  // the documented maxTurns:1 aichat pattern). Don't persist the half-done
  // turn either — a fresh retry beats resuming into unanswered tool results.
  if (capHit && !fullText) {
    return {
      ok: false,
      error: `Agent hit its turn limit (maxTurns=${maxTurns}) before producing a response — raise maxTurns or narrow the request.`,
      sessionId,
      toolsUsed: Array.from(usedTools),
    };
  }

  // Trim at store time too — otherwise a long-lived session's stored history
  // grows unboundedly across resumes (the resume-side trim only caps what the
  // provider sees, not what we keep in memory).
  agentSessions.set(sessionId, { messages: trimAgentHistory(session.getMessages()), lastUsed: Date.now() });

  // Partial-text precedence (claude parity, audit D6-2): if the model streamed
  // anything before failing, return it as a successful (truncated) response —
  // claude's runAgentQuery only reports the error when nothing streamed.
  if (errored && !fullText) {
    return { ok: false, error: errorMsg || 'Agent query failed', sessionId, toolsUsed: Array.from(usedTools) };
  }

  const fileToolsUsed = usedFileTools || ['Write', 'Edit', 'write', 'edit'].some((t) => usedTools.has(t));
  log.info(`[pi/agent-api] Done: ${fullText.length} chars, tools=[${Array.from(usedTools).join(',')}], session=${sessionId}`);
  return { ok: true, response: fullText, sessionId, toolsUsed: Array.from(usedTools), usedFileTools: fileToolsUsed };
}