/**
 * NanoClaw Agent Runner
 * Runs inside a container, receives config via stdin, outputs result to stdout
 *
 * Input protocol:
 *   Stdin: Full ContainerInput JSON (read until EOF, like before)
 *   IPC:   Follow-up messages written as JSON files to /workspace/ipc/input/
 *          Files: {type:"message", text:"..."}.json — polled and consumed
 *          Sentinel: /workspace/ipc/input/_close — signals session end
 *
 * Stdout protocol:
 *   Each result is wrapped in OUTPUT_START_MARKER / OUTPUT_END_MARKER pairs.
 *   Multiple results may be emitted (one per agent teams result).
 *   Final marker after loop ends signals completion.
 */

import fs from 'fs';
import path from 'path';
import os from 'os';
import { execFileSync } from 'child_process';
import { query, HookCallback, PreCompactHookInput, PreToolUseHookInput } from '@anthropic-ai/claude-agent-sdk';
import { Codex, ThreadEvent, ThreadOptions, UserInput } from '@openai/codex-sdk';
import { fileURLToPath } from 'url';
import { WorkspaceRuntimePromptAdapter } from './runtime-adapter.js';
import { createAgentSession, createCodingTools, AuthStorage, ModelRegistry } from '@mariozechner/pi-coding-agent';
import { getOAuthProvider } from '@mariozechner/pi-ai/oauth';

interface Attachment {
  kind: 'image' | 'document' | 'text' | 'audio' | 'file';
  data?: string;
  mediaType: string;
  name: string;
  mountedPath?: string;
}

interface ContainerInput {
  provider?: 'claude' | 'codex' | 'pi';
  pi_backend?: string; // LLM backend for pi: 'anthropic' | 'openai' | 'gemini'
  prompt: string;
  sessionId?: string;
  groupFolder: string;
  chatJid: string;
  isMain: boolean;
  mainGroupFolder?: string;
  isScheduledTask?: boolean;
  assistantName?: string;
  secrets?: Record<string, string>;
  model?: string;
  /** Start in warm-standby mode — skip initial query, wait for IPC */
  warmOnly?: boolean;
  attachments?: Attachment[];
}

interface ContainerOutput {
  status: 'success' | 'error';
  result: string | null;
  newSessionId?: string;
  error?: string;
  progress?: string;
  /** Streamed assistant prose text */
  assistantText?: string;
}

type RuntimeProvider = 'claude' | 'codex' | 'pi';

interface SessionEntry {
  sessionId: string;
  fullPath: string;
  summary: string;
  firstPrompt: string;
}

interface SessionsIndex {
  entries: SessionEntry[];
}

type ContentBlock =
  | { type: 'text'; text: string }
  | { type: 'image'; source: { type: 'base64'; media_type: 'image/jpeg' | 'image/png' | 'image/gif' | 'image/webp'; data: string } }
  | { type: 'document'; source: { type: 'base64'; media_type: string; data: string } };

interface SDKUserMessage {
  type: 'user';
  message: { role: 'user'; content: string | ContentBlock[] };
  parent_tool_use_id: null;
  session_id: string;
}

interface PreparedMessageInput {
  text: string;
  attachments?: Attachment[];
}

const IPC_INPUT_DIR = '/workspace/ipc/input';
const IPC_INPUT_CLOSE_SENTINEL = path.join(IPC_INPUT_DIR, '_close');
const IPC_POLL_MS = 500;
const TOKEN_FILE_PATH = '/workspace/ipc/.token.json';
const DEFAULT_WHISPER_MODEL_PATH = '/models/whisper/ggml-base.en.bin';

/**
 * Re-read the token file and update sdkEnv if the token has changed.
 * The host writes this file when it refreshes OAuth credentials,
 * so long-running containers pick up new tokens without restarting.
 */
function refreshSdkEnvFromTokenFile(sdkEnv: Record<string, string | undefined>): void {
  try {
    if (!fs.existsSync(TOKEN_FILE_PATH)) return;
    const raw = fs.readFileSync(TOKEN_FILE_PATH, 'utf-8');
    const data = JSON.parse(raw) as Record<string, string>;
    let updated = false;
    for (const key of ['CLAUDE_CODE_OAUTH_TOKEN', 'ANTHROPIC_API_KEY', 'ANTHROPIC_BASE_URL', 'ANTHROPIC_AUTH_TOKEN', 'OPENAI_API_KEY', 'OPENAI_BASE_URL', 'CODEX_AUTH_JSON', 'PI_AUTH_JSON']) {
      if (data[key] && data[key] !== sdkEnv[key]) {
        sdkEnv[key] = data[key];
        updated = true;
      }
    }
    if (updated) {
      log(`Refreshed credentials from token file (updated_at: ${data._updatedAt || 'unknown'})`);
    }
  } catch {
    // Ignore read errors (file being written, corrupted, etc.)
  }
}

function prepareCodexAuthFiles(
  sdkEnv: Record<string, string | undefined>,
  model?: string,
): void {
  const authJson = sdkEnv.CODEX_AUTH_JSON;
  if (!authJson) return;

  const codexDir = '/home/node/.codex';
  fs.mkdirSync(codexDir, { recursive: true });
  fs.writeFileSync(path.join(codexDir, 'auth.json'), authJson, { mode: 0o600 });
  if (model) {
    fs.writeFileSync(path.join(codexDir, 'config.toml'), `model = "${model}"\n`, {
      mode: 0o600,
    });
  }
  delete sdkEnv.CODEX_AUTH_JSON;
}

/**
 * Push-based async iterable for streaming user messages to the SDK.
 * Keeps the iterable alive until end() is called, preventing isSingleUserTurn.
 */
class MessageStream {
  private queue: SDKUserMessage[] = [];
  private waiting: (() => void) | null = null;
  private done = false;

  push(text: string): void {
    this.queue.push({
      type: 'user',
      message: { role: 'user', content: text },
      parent_tool_use_id: null,
      session_id: '',
    });
    this.waiting?.();
  }

  pushMultimodal(text: string, attachments: Attachment[]): void {
    const extraText: string[] = [];
    const contentBlocks: ContentBlock[] = [];
    for (const att of attachments) {
      if (att.kind === 'image') {
        contentBlocks.push({
          type: 'image',
          source: { type: 'base64', media_type: att.mediaType as 'image/jpeg' | 'image/png' | 'image/gif' | 'image/webp', data: att.data || '' },
        });
        if (att.mountedPath) {
          extraText.push(`[Image "${att.name}" is saved at ${att.mountedPath} — use this path to read or process the file programmatically]`);
        }
      } else if (att.kind === 'document') {
        contentBlocks.push({
          type: 'document',
          source: { type: 'base64', media_type: att.mediaType, data: att.data || '' },
        });
        if (att.mountedPath) {
          extraText.push(`[Document "${att.name}" is saved at ${att.mountedPath} — use this path to read or process the file programmatically]`);
        }
      } else if (att.kind === 'text') {
        extraText.push(`\n--- File: ${att.name} ---\n${att.data || ''}\n--- End: ${att.name} ---`);
      } else if (att.kind === 'file') {
        if (att.mountedPath) {
          extraText.push(`[File "${att.name}" is available at ${att.mountedPath} — use this path to read or process the file programmatically]`);
        }
      }
    }
    const fullText = text + extraText.join('\n');
    const content: ContentBlock[] = [
      ...contentBlocks,
      { type: 'text' as const, text: fullText },
    ];
    this.queue.push({
      type: 'user',
      message: { role: 'user', content },
      parent_tool_use_id: null,
      session_id: '',
    });
    this.waiting?.();
  }

  end(): void {
    this.done = true;
    this.waiting?.();
  }

  async *[Symbol.asyncIterator](): AsyncGenerator<SDKUserMessage> {
    while (true) {
      while (this.queue.length > 0) {
        yield this.queue.shift()!;
      }
      if (this.done) return;
      await new Promise<void>(r => { this.waiting = r; });
      this.waiting = null;
    }
  }
}

async function readStdin(): Promise<string> {
  return new Promise((resolve, reject) => {
    let data = '';
    process.stdin.setEncoding('utf8');
    process.stdin.on('data', chunk => { data += chunk; });
    process.stdin.on('end', () => resolve(data));
    process.stdin.on('error', reject);
  });
}

const OUTPUT_START_MARKER = '---NANOCLAW_OUTPUT_START---';
const OUTPUT_END_MARKER = '---NANOCLAW_OUTPUT_END---';

// Prevent unhandled EPIPE crashes when the host closes the pipe
process.stdout.on('error', (err: NodeJS.ErrnoException) => {
  if (err.code === 'EPIPE') process.exit(0);
  throw err;
});

function writeOutput(output: ContainerOutput): void {
  console.log(OUTPUT_START_MARKER);
  console.log(JSON.stringify(output));
  console.log(OUTPUT_END_MARKER);
}

function log(message: string): void {
  console.error(`[agent-runner] ${message}`);
}

function buildAudioTranscriptBlock(name: string, transcript: string): string {
  return `\n\n[Audio transcript: ${name}]\n${transcript.trim()}\n[End audio transcript: ${name}]`;
}

export function transcribeAudioAttachment(
  attachment: Attachment,
  env: NodeJS.ProcessEnv = process.env,
): string {
  if (!attachment.mountedPath) {
    throw new Error(`Audio attachment ${attachment.name} is missing mountedPath`);
  }

  const ffmpegBin = env.NANOCLAW_FFMPEG_BIN || 'ffmpeg';
  const whisperBin = env.NANOCLAW_WHISPER_BIN || 'whisper-cli';
  const whisperModelPath = env.NANOCLAW_WHISPER_MODEL || DEFAULT_WHISPER_MODEL_PATH;
  if (!fs.existsSync(attachment.mountedPath)) {
    throw new Error(`Mounted audio file not found: ${attachment.mountedPath}`);
  }
  if (!fs.existsSync(whisperModelPath)) {
    throw new Error(`Whisper model not found: ${whisperModelPath}`);
  }

  const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), 'nanoclaw-stt-'));
  const wavPath = path.join(tempDir, 'input.wav');
  const transcriptBase = path.join(tempDir, 'transcript');
  const transcriptPath = `${transcriptBase}.txt`;

  try {
    execFileSync(
      ffmpegBin,
      ['-y', '-i', attachment.mountedPath, '-ac', '1', '-ar', '16000', '-c:a', 'pcm_s16le', wavPath],
      { stdio: 'pipe' },
    );
    execFileSync(
      whisperBin,
      ['-m', whisperModelPath, '-f', wavPath, '-of', transcriptBase, '-otxt', '-nt'],
      { stdio: 'pipe' },
    );
    return fs.readFileSync(transcriptPath, 'utf-8').trim();
  } finally {
    fs.rmSync(tempDir, { recursive: true, force: true });
  }
}

export function prepareMessageInput(
  text: string,
  attachments: Attachment[] | undefined,
  env: NodeJS.ProcessEnv = process.env,
): PreparedMessageInput {
  if (!attachments?.length) {
    return { text, attachments: undefined };
  }

  const remainingAttachments: Attachment[] = [];
  let preparedText = text;

  for (const attachment of attachments) {
    if (attachment.kind !== 'audio') {
      remainingAttachments.push(attachment);
      continue;
    }

    try {
      const transcript = transcribeAudioAttachment(attachment, env);
      preparedText += buildAudioTranscriptBlock(attachment.name, transcript);
      log(`Transcribed audio attachment ${attachment.name} (${transcript.length} chars)`);
    } catch (err) {
      const errorMessage = err instanceof Error ? err.message : String(err);
      preparedText += `\n\n[Audio attachment received: ${attachment.name}]\nAutomatic transcription failed: ${errorMessage}\n`;
      log(`Audio transcription failed for ${attachment.name}: ${errorMessage}`);
    }
  }

  return {
    text: preparedText,
    attachments: remainingAttachments.length > 0 ? remainingAttachments : undefined,
  };
}

function normalizeProvider(provider?: string): RuntimeProvider {
  if (provider === 'codex') return 'codex';
  if (provider === 'pi') return 'pi';
  return 'claude';
}

function getDefaultModel(provider: RuntimeProvider): string {
  if (provider === 'codex') return 'gpt-5.2';
  return 'claude-sonnet-4-6'; // pi also uses claude models
}

function getSessionSummary(sessionId: string, transcriptPath: string): string | null {
  const projectDir = path.dirname(transcriptPath);
  const indexPath = path.join(projectDir, 'sessions-index.json');

  if (!fs.existsSync(indexPath)) {
    log(`Sessions index not found at ${indexPath}`);
    return null;
  }

  try {
    const index: SessionsIndex = JSON.parse(fs.readFileSync(indexPath, 'utf-8'));
    const entry = index.entries.find(e => e.sessionId === sessionId);
    if (entry?.summary) {
      return entry.summary;
    }
  } catch (err) {
    log(`Failed to read sessions index: ${err instanceof Error ? err.message : String(err)}`);
  }

  return null;
}

/**
 * Archive the full transcript to conversations/ before compaction.
 */
function createPreCompactHook(assistantName?: string): HookCallback {
  return async (input, _toolUseId, _context) => {
    const preCompact = input as PreCompactHookInput;
    const transcriptPath = preCompact.transcript_path;
    const sessionId = preCompact.session_id;

    if (!transcriptPath || !fs.existsSync(transcriptPath)) {
      log('No transcript found for archiving');
      return {};
    }

    try {
      const content = fs.readFileSync(transcriptPath, 'utf-8');
      const messages = parseTranscript(content);

      if (messages.length === 0) {
        log('No messages to archive');
        return {};
      }

      const summary = getSessionSummary(sessionId, transcriptPath);
      const name = summary ? sanitizeFilename(summary) : generateFallbackName();

      const conversationsDir = '/workspace/group/conversations';
      fs.mkdirSync(conversationsDir, { recursive: true });

      const date = new Date().toISOString().split('T')[0];
      const filename = `${date}-${name}.md`;
      const filePath = path.join(conversationsDir, filename);

      const markdown = formatTranscriptMarkdown(messages, summary, assistantName);
      fs.writeFileSync(filePath, markdown);

      log(`Archived conversation to ${filePath}`);
    } catch (err) {
      log(`Failed to archive transcript: ${err instanceof Error ? err.message : String(err)}`);
    }

    // LCM: signal host to run compaction now that Claude is about to wipe in-session context
    try {
      const ipcMessagesDir = '/workspace/ipc/messages';
      fs.mkdirSync(ipcMessagesDir, { recursive: true });
      const fname = `${Date.now()}-lcm-compact.json`;
      const tmp = path.join(ipcMessagesDir, fname + '.tmp');
      fs.writeFileSync(tmp, JSON.stringify({ type: 'lcm-compact' }));
      fs.renameSync(tmp, path.join(ipcMessagesDir, fname));
    } catch {
      // best-effort
    }

    return {};
  };
}

// Secrets to strip from Bash tool subprocess environments.
// These are needed by claude-code for API auth but should never
// be visible to commands Kit runs.
const SECRET_ENV_VARS = ['ANTHROPIC_API_KEY', 'CLAUDE_CODE_OAUTH_TOKEN'];

function createSanitizeBashHook(): HookCallback {
  return async (input, _toolUseId, _context) => {
    const preInput = input as PreToolUseHookInput;
    const command = (preInput.tool_input as { command?: string })?.command;
    if (!command) return {};

    const unsetPrefix = `unset ${SECRET_ENV_VARS.join(' ')} 2>/dev/null; `;
    return {
      hookSpecificOutput: {
        hookEventName: 'PreToolUse',
        updatedInput: {
          ...(preInput.tool_input as Record<string, unknown>),
          command: unsetPrefix + command,
        },
      },
    };
  };
}

function sanitizeFilename(summary: string): string {
  return summary
    .toLowerCase()
    .replace(/[^a-z0-9]+/g, '-')
    .replace(/^-+|-+$/g, '')
    .slice(0, 50);
}

function generateFallbackName(): string {
  const time = new Date();
  return `conversation-${time.getHours().toString().padStart(2, '0')}${time.getMinutes().toString().padStart(2, '0')}`;
}

interface ParsedMessage {
  role: 'user' | 'assistant';
  content: string;
}

function parseTranscript(content: string): ParsedMessage[] {
  const messages: ParsedMessage[] = [];

  for (const line of content.split('\n')) {
    if (!line.trim()) continue;
    try {
      const entry = JSON.parse(line);
      if (entry.type === 'user' && entry.message?.content) {
        const text = typeof entry.message.content === 'string'
          ? entry.message.content
          : entry.message.content.map((c: { text?: string }) => c.text || '').join('');
        if (text) messages.push({ role: 'user', content: text });
      } else if (entry.type === 'assistant' && entry.message?.content) {
        const textParts = entry.message.content
          .filter((c: { type: string }) => c.type === 'text')
          .map((c: { text: string }) => c.text);
        const text = textParts.join('');
        if (text) messages.push({ role: 'assistant', content: text });
      }
    } catch {
    }
  }

  return messages;
}

function formatTranscriptMarkdown(messages: ParsedMessage[], title?: string | null, assistantName?: string): string {
  const now = new Date();
  const formatDateTime = (d: Date) => d.toLocaleString('en-US', {
    month: 'short',
    day: 'numeric',
    hour: 'numeric',
    minute: '2-digit',
    hour12: true
  });

  const lines: string[] = [];
  lines.push(`# ${title || 'Conversation'}`);
  lines.push('');
  lines.push(`Archived: ${formatDateTime(now)}`);
  lines.push('');
  lines.push('---');
  lines.push('');

  for (const msg of messages) {
    const sender = msg.role === 'user' ? 'User' : (assistantName || 'Assistant');
    const content = msg.content.length > 2000
      ? msg.content.slice(0, 2000) + '...'
      : msg.content;
    lines.push(`**${sender}**: ${content}`);
    lines.push('');
  }

  return lines.join('\n');
}

/**
 * Check for _close sentinel.
 */
function shouldClose(): boolean {
  if (fs.existsSync(IPC_INPUT_CLOSE_SENTINEL)) {
    try { fs.unlinkSync(IPC_INPUT_CLOSE_SENTINEL); } catch { /* ignore */ }
    return true;
  }
  return false;
}

/**
 * Drain all pending IPC input messages.
 * Returns messages found, or empty array.
 */
interface IpcMessage { text: string; attachments?: Attachment[]; }

function drainIpcInput(): IpcMessage[] {
  try {
    fs.mkdirSync(IPC_INPUT_DIR, { recursive: true });
    const files = fs.readdirSync(IPC_INPUT_DIR)
      .filter(f => f.endsWith('.json'))
      .sort();

    const messages: IpcMessage[] = [];
    for (const file of files) {
      const filePath = path.join(IPC_INPUT_DIR, file);
      try {
        const data = JSON.parse(fs.readFileSync(filePath, 'utf-8'));
        fs.unlinkSync(filePath);
        if (data.type === 'message' && data.text) {
          messages.push({ text: data.text, attachments: data.attachments });
        }
      } catch (err) {
        log(`Failed to process input file ${file}: ${err instanceof Error ? err.message : String(err)}`);
        try { fs.unlinkSync(filePath); } catch { /* ignore */ }
      }
    }
    return messages;
  } catch (err) {
    log(`IPC drain error: ${err instanceof Error ? err.message : String(err)}`);
    return [];
  }
}

/**
 * Wait for a new IPC message or _close sentinel.
 * Returns the messages as a single string, or null if _close.
 */
function waitForIpcMessage(): Promise<IpcMessage | null> {
  return new Promise((resolve) => {
    const poll = () => {
      if (shouldClose()) {
        resolve(null);
        return;
      }
      const messages = drainIpcInput();
      if (messages.length > 0) {
        // Merge multiple messages: concatenate text, combine attachments
        const text = messages.map(m => m.text).join('\n');
        const attachments = messages.flatMap(m => m.attachments || []);
        resolve({ text, attachments: attachments.length > 0 ? attachments : undefined });
        return;
      }
      setTimeout(poll, IPC_POLL_MS);
    };
    poll();
  });
}

function buildRuntimePromptContext(containerInput: ContainerInput): {
  promptAppend?: string;
  injectedSummary?: string;
  extraDirs: string[];
} {
  const globalClaudeMdPath = '/workspace/global/CLAUDE.md';
  let globalPromptAppend: string | undefined;
  if (!containerInput.isMain && fs.existsSync(globalClaudeMdPath)) {
    globalPromptAppend = fs.readFileSync(globalClaudeMdPath, 'utf-8');
  }

  const runtimeAdapter = new WorkspaceRuntimePromptAdapter();
  const mainProfileDir = '/workspace/main-profile';
  const promptContext = runtimeAdapter.buildPromptContext({
    groupDir: '/workspace/group',
    globalPromptAppend,
  });

  const extraDirs: string[] = [];
  if (fs.existsSync('/workspace')) {
    extraDirs.push('/workspace');
  }
  if (!containerInput.isMain && fs.existsSync(mainProfileDir)) {
    extraDirs.push(mainProfileDir);
  }
  const extraBase = '/workspace/extra';
  if (fs.existsSync(extraBase)) {
    for (const entry of fs.readdirSync(extraBase)) {
      const fullPath = path.join(extraBase, entry);
      if (fs.statSync(fullPath).isDirectory()) {
        extraDirs.push(fullPath);
      }
    }
  }

  return {
    promptAppend: promptContext.systemPromptAppend,
    injectedSummary: promptContext.injectedSummary,
    extraDirs,
  };
}

function estimateTextMetrics(text: string): {
  chars: number;
  words: number;
  tokens: number;
} {
  const chars = text.length;
  const words = (text.trim().match(/\S+/g) || []).length;
  const tokens = Math.ceil(chars / 4);
  return { chars, words, tokens };
}

/**
 * Run a single query and stream results via writeOutput.
 * Uses MessageStream (AsyncIterable) to keep isSingleUserTurn=false,
 * allowing agent teams subagents to run to completion.
 * Also pipes IPC messages into the stream during the query.
 */
async function runQuery(
  prompt: string,
  sessionId: string | undefined,
  containerInput: ContainerInput,
  sdkEnv: Record<string, string | undefined>,
  resumeAt?: string,
): Promise<{ newSessionId?: string; lastAssistantUuid?: string; closedDuringQuery: boolean; authError: boolean }> {
  const stream = new MessageStream();
  const preparedInitial = prepareMessageInput(prompt, containerInput.attachments);
  if (preparedInitial.attachments?.length) {
    stream.pushMultimodal(preparedInitial.text, preparedInitial.attachments);
  } else {
    stream.push(preparedInitial.text);
  }

  // Poll IPC for follow-up messages and _close sentinel during the query
  let ipcPolling = true;
  let closedDuringQuery = false;
  const pollIpcDuringQuery = () => {
    if (!ipcPolling) return;
    if (shouldClose()) {
      log('Close sentinel detected during query, ending stream');
      closedDuringQuery = true;
      stream.end();
      ipcPolling = false;
      return;
    }
    const messages = drainIpcInput();
    for (const msg of messages) {
      log(`Piping IPC message into active query (${msg.text.length} chars, ${msg.attachments?.length || 0} attachments)`);
      const preparedMessage = prepareMessageInput(msg.text, msg.attachments);
      if (preparedMessage.attachments?.length) {
        stream.pushMultimodal(preparedMessage.text, preparedMessage.attachments);
      } else {
        stream.push(preparedMessage.text);
      }
    }
    setTimeout(pollIpcDuringQuery, IPC_POLL_MS);
  };
  setTimeout(pollIpcDuringQuery, IPC_POLL_MS);

  let newSessionId: string | undefined;
  let lastAssistantUuid: string | undefined;
  let messageCount = 0;
  let resultCount = 0;
  let authError = false;

  const promptContext = buildRuntimePromptContext(containerInput);
  if (promptContext.injectedSummary) {
    log(`Workspace context injected (${promptContext.injectedSummary})`);
  }
  if (promptContext.promptAppend) {
    const metrics = estimateTextMetrics(promptContext.promptAppend);
    log(
      `System prompt append size: chars=${metrics.chars} words=${metrics.words} tokens~=${metrics.tokens}`,
    );
  }

  if (promptContext.extraDirs.length > 0) {
    log(`Additional directories: ${promptContext.extraDirs.join(', ')}`);
  }

  try {
    for await (const message of query({
      prompt: stream as any,
      options: {
        cwd: '/workspace/group',
        additionalDirectories: promptContext.extraDirs.length > 0 ? promptContext.extraDirs : undefined,
        resume: sessionId,
        resumeSessionAt: resumeAt,
        systemPrompt: promptContext.promptAppend
          ? {
              type: 'preset' as const,
              preset: 'claude_code' as const,
              append: promptContext.promptAppend,
            }
          : undefined,
        allowedTools: [
          'Bash',
          'Read', 'Write', 'Edit', 'Glob', 'Grep',
          'WebSearch', 'WebFetch',
          'Task', 'TaskOutput', 'TaskStop',
          'TeamCreate', 'TeamDelete', 'SendMessage',
          'TodoWrite', 'ToolSearch', 'Skill',
          'NotebookEdit',
        ],
        env: sdkEnv,
        permissionMode: 'bypassPermissions',
        allowDangerouslySkipPermissions: true,
        settingSources: ['project', 'user'],
        hooks: {
          PreCompact: [{ hooks: [createPreCompactHook(containerInput.assistantName)] }],
          PreToolUse: [{ matcher: 'Bash', hooks: [createSanitizeBashHook()] }],
        },
      }
    })) {
      messageCount++;
      const msgType = message.type === 'system' ? `system/${(message as { subtype?: string }).subtype}` : message.type;
      log(`[msg #${messageCount}] type=${msgType}`);

      if (message.type === 'assistant') {
        if ('uuid' in message) {
          lastAssistantUuid = (message as { uuid: string }).uuid;
        }

        // Emit assistant text and tool use progress to the host
        const msg = message as Record<string, unknown>;
        const msgInner = msg.message as { content?: unknown[] } | undefined;
        const content = msgInner?.content || (msg as { content?: unknown[] }).content;
        if (Array.isArray(content)) {
          for (const block of content) {
            const b = block as { type?: string; name?: string; text?: string; input?: Record<string, unknown> };
            if (b.type === 'text' && b.text) {
              // Suppress auth error text — the retry loop will handle recovery
              if (/API Error: 401|authentication_error|Invalid authentication credentials/.test(b.text)) {
                log(`AssistantText (suppressed auth error): ${b.text.slice(0, 120)}`);
              } else {
                log(`AssistantText: ${b.text.slice(0, 120)}`);
                writeOutput({ status: 'success', result: null, assistantText: b.text });
              }
            } else if (b.type === 'tool_use' && b.name) {
              const summary = formatToolProgress(b.name, b.input);
              if (summary) {
                log(`Progress: ${summary}`);
                writeOutput({ status: 'success', result: null, progress: summary });
              }
            }
          }
        } else {
          log(`Assistant message has no content array, keys: ${Object.keys(msg).join(',')}`);
        }
      }

      if (message.type === 'system' && message.subtype === 'init') {
        newSessionId = message.session_id;
        log(`Session initialized: ${newSessionId}`);
      }

      if (message.type === 'system' && (message as { subtype?: string }).subtype === 'task_notification') {
        const tn = message as { task_id: string; status: string; summary: string };
        log(`Task notification: task=${tn.task_id} status=${tn.status} summary=${tn.summary}`);
      }

      if (message.type === 'result') {
        resultCount++;
        const textResult = 'result' in message ? (message as { result?: string }).result : null;
        log(`Result #${resultCount}: subtype=${message.subtype}${textResult ? ` text=${textResult.slice(0, 200)}` : ''}`);

        // Detect authentication errors in result text — stop IPC polling
        // and end the stream so runQuery exits and the outer loop can retry
        // with a refreshed token from the token file.
        if (textResult && /API Error: 401|authentication_error|Invalid authentication credentials/.test(textResult)) {
          authError = true;
          ipcPolling = false;
          stream.end();
          log('Auth error detected in query result, ending stream for retry');
          // Don't emit auth error results to the user
        } else {
          writeOutput({
            status: 'success',
            result: textResult || null,
            newSessionId
          });
        }
      }
    }
  } catch (err) {
    // If we already detected an auth error and called stream.end(), the SDK
    // process may exit non-zero. Swallow this so the outer loop can retry.
    if (authError) {
      log(`SDK exited after auth error (expected): ${err instanceof Error ? err.message : err}`);
    } else {
      throw err;
    }
  }

  ipcPolling = false;
  log(`Query done. Messages: ${messageCount}, results: ${resultCount}, lastAssistantUuid: ${lastAssistantUuid || 'none'}, closedDuringQuery: ${closedDuringQuery}, authError: ${authError}`);
  return { newSessionId, lastAssistantUuid, closedDuringQuery, authError };
}

function buildCodexInput(
  prompt: string,
  attachments: Attachment[] | undefined,
): { input: string | UserInput[]; cleanupPaths: string[] } {
  if (!attachments?.length) {
    return { input: prompt, cleanupPaths: [] };
  }

  const content: UserInput[] = [{ type: 'text', text: prompt }];
  const cleanupPaths: string[] = [];
  const inlineText: string[] = [];

  for (const attachment of attachments) {
    if (attachment.kind === 'image') {
      const safeName = attachment.name.replace(/[^a-zA-Z0-9._-]/g, '-');
      const imagePath = path.join('/tmp', `codex-${Date.now()}-${safeName}`);
      fs.writeFileSync(imagePath, Buffer.from(attachment.data || '', 'base64'));
      cleanupPaths.push(imagePath);
      content.push({ type: 'local_image', path: imagePath });
      continue;
    }

    if (attachment.kind === 'text') {
      inlineText.push(`\n--- File: ${attachment.name} ---\n${attachment.data || ''}\n--- End: ${attachment.name} ---`);
      continue;
    }

    inlineText.push(
      `\n[Attachment omitted: ${attachment.name} (${attachment.mediaType}). ` +
      'Codex image inputs are supported; non-image binary documents are summarized here only.]',
    );
  }

  if (inlineText.length > 0) {
    content[0] = {
      type: 'text',
      text: prompt + inlineText.join('\n'),
    };
  }

  return { input: content, cleanupPaths };
}

async function runCodexTurn(
  prompt: string,
  sessionId: string | undefined,
  containerInput: ContainerInput,
  sdkEnv: Record<string, string | undefined>,
): Promise<{ newSessionId?: string; closedDuringQuery: boolean; authError: boolean }> {
  const promptContext = buildRuntimePromptContext(containerInput);
  if (promptContext.injectedSummary) {
    log(`Workspace context injected (${promptContext.injectedSummary})`);
  }
  if (promptContext.promptAppend) {
    const metrics = estimateTextMetrics(promptContext.promptAppend);
    log(
      `System prompt append size: chars=${metrics.chars} words=${metrics.words} tokens~=${metrics.tokens}`,
    );
  }
  if (promptContext.extraDirs.length > 0) {
    log(`Additional directories: ${promptContext.extraDirs.join(', ')}`);
  }

  const codex = new Codex({
    apiKey: sdkEnv.OPENAI_API_KEY,
    baseUrl: sdkEnv.OPENAI_BASE_URL,
    env: Object.fromEntries(
      Object.entries(sdkEnv).filter((entry): entry is [string, string] => typeof entry[1] === 'string'),
    ),
    config: {
      approval_policy: 'never',
    },
  });

  const threadOptions: ThreadOptions = {
    model: containerInput.model,
    workingDirectory: '/workspace/group',
    skipGitRepoCheck: true,
    sandboxMode: process.env.GVISOR_ACTIVE === '1'
      ? 'danger-full-access'
      : 'workspace-write',
    approvalPolicy: 'never',
    networkAccessEnabled: true,
    additionalDirectories: promptContext.extraDirs.length > 0
      ? promptContext.extraDirs
      : undefined,
  };
  const thread = sessionId
    ? codex.resumeThread(sessionId, threadOptions)
    : codex.startThread(threadOptions);

  const fullPrompt = promptContext.promptAppend
    ? `${promptContext.promptAppend}\n\n${prompt}`
    : prompt;
  const preparedMessage = prepareMessageInput(fullPrompt, containerInput.attachments);
  const codexInput = buildCodexInput(preparedMessage.text, preparedMessage.attachments);

  let finalResponse: string | undefined;
  try {
    const { events } = await thread.runStreamed(codexInput.input);
    for await (const event of events) {
      const eventResult = handleCodexEvent(event, thread.id || undefined);
      if (eventResult) {
        finalResponse = eventResult;
      }
    }
  } finally {
    for (const cleanupPath of codexInput.cleanupPaths) {
      try { fs.unlinkSync(cleanupPath); } catch { /* ignore */ }
    }
  }

  writeOutput({
    status: 'success',
    result: finalResponse || null,
    newSessionId: thread.id || sessionId,
  });

  const authError = !!(finalResponse && /API Error: 401|authentication_error|Invalid authentication credentials/.test(finalResponse));
  return {
    newSessionId: thread.id || sessionId,
    closedDuringQuery: false,
    authError,
  };
}

function handleCodexEvent(
  event: ThreadEvent,
  threadId?: string,
): string | undefined {
  switch (event.type) {
    case 'thread.started':
      log(`Session initialized: ${event.thread_id}`);
      break;
    case 'item.started':
    case 'item.updated':
    case 'item.completed': {
      const item = event.item;
      if (item.type === 'agent_message' && item.text && event.type === 'item.completed') {
        writeOutput({
          status: 'success',
          result: null,
          newSessionId: threadId,
          assistantText: item.text,
        });
        return item.text;
      } else if (item.type === 'command_execution' && event.type === 'item.started') {
        const summary = formatToolProgress('Bash', { command: item.command });
        if (summary) {
          writeOutput({
            status: 'success',
            result: null,
            newSessionId: threadId,
            progress: summary,
          });
        }
      } else if (item.type === 'mcp_tool_call' && event.type === 'item.started') {
        const summary = formatToolProgress(`mcp__${item.server}__${item.tool}`, item.arguments as Record<string, unknown>);
        if (summary) {
          writeOutput({
            status: 'success',
            result: null,
            newSessionId: threadId,
            progress: summary,
          });
        }
      } else if (item.type === 'web_search' && event.type === 'item.started') {
        const summary = formatToolProgress('WebSearch', { query: item.query });
        if (summary) {
          writeOutput({
            status: 'success',
            result: null,
            newSessionId: threadId,
            progress: summary,
          });
        }
      } else if (item.type === 'error') {
        writeOutput({
          status: 'error',
          result: null,
          newSessionId: threadId,
          error: item.message,
        });
      }
      break;
    }
    case 'turn.completed':
      return undefined;
    case 'turn.failed':
      throw new Error(event.error.message);
    case 'error':
      throw new Error(event.message);
  }
}

/**
 * Format a tool use into a short human-readable progress string.
 * Returns null for tools that aren't worth reporting (e.g. ToolSearch).
 */
function formatToolProgress(toolName: string, input?: Record<string, unknown>): string | null {
  const filePath = String(input?.file_path || '').trim();
  switch (toolName) {
    case 'Bash': {
      const command = String(input?.command || '').trim();
      if (!command) return null;

      const maxLength = 500;
      if (command.length <= maxLength) return command;
      return `${command.slice(0, maxLength - 1)}…`;
    }
    case 'Read':
      if (!filePath) return null;
      if (
        filePath.endsWith('/MEMORY.md') ||
        filePath.endsWith('/BOOTSTRAP.md') ||
        filePath.endsWith('/AGENTS.md') ||
        filePath.endsWith('/IDENTITY.md') ||
        filePath.endsWith('/USER.md')
      ) {
        return null;
      }
      return `Reading \`${filePath}\``;
    case 'Write':
      return `Writing \`${input?.file_path || 'file'}\``;
    case 'Edit':
      return `Editing \`${input?.file_path || 'file'}\``;
    case 'Glob':
      return `Searching for \`${input?.pattern || 'files'}\``;
    case 'Grep':
      return `Searching for "${String(input?.pattern || '').slice(0, 60)}"`;
    case 'WebSearch':
      return `Searching the web for "${String(input?.query || '').slice(0, 80)}"`;
    case 'WebFetch':
      return `Fetching ${String(input?.url || 'URL').slice(0, 100)}`;
    case 'NotebookEdit':
      return `Editing notebook`;
    case 'TodoWrite':
      return null; // internal bookkeeping
    case 'ToolSearch':
      return null; // internal
    case 'Skill':
      return input?.name ? `Using skill: ${input.name}` : 'Working';
    case 'Task':
    case 'TeamCreate':
      return `Spawning subagent`;
    // Pi tool names (lowercase)
    case 'bash': {
      const _cmd = String(input?.command || '').trim();
      if (!_cmd) return null;
      return _cmd.length <= 500 ? _cmd : _cmd.slice(0, 499) + '…';
    }
    case 'read': {
      const _p = String(input?.path || '').trim();
      if (!_p) return null;
      const hidden = ['/MEMORY.md', '/BOOTSTRAP.md', '/AGENTS.md', '/IDENTITY.md', '/USER.md'];
      if (hidden.some(h => _p.endsWith(h))) return null;
      return `Reading \`${_p}\``;
    }
    case 'write':
      return `Writing \`${String(input?.path || 'file')}\``;
    case 'edit':
      return `Editing \`${String(input?.path || 'file')}\``;
    case 'grep':
      return `Searching for "${String(input?.pattern || '').slice(0, 60)}"`;
    case 'ls':
      return `Listing \`${String(input?.path || '.')}\``;
    default:
      if (toolName.startsWith('mcp__')) {
        const parts = toolName.split('__');
        return `Using tool`;
      }
      return `Using tool: ${toolName}`;
  }
}


// ---------------------------------------------------------------------------
// Pi provider (pi-ai + pi-agent-core via @mariozechner/pi-coding-agent)
// ---------------------------------------------------------------------------

function fireLcmCompactSignal(): void {
  try {
    const ipcMessagesDir = '/workspace/ipc/messages';
    fs.mkdirSync(ipcMessagesDir, { recursive: true });
    const fname = `${Date.now()}-lcm-compact.json`;
    const tmp = path.join(ipcMessagesDir, fname + '.tmp');
    fs.writeFileSync(tmp, JSON.stringify({ type: 'lcm-compact' }));
    fs.renameSync(tmp, path.join(ipcMessagesDir, fname));
    log('Pi: LCM compact signal sent');
  } catch {
    // best-effort
  }
}

async function runPiSession(
  initialPrompt: string,
  containerInput: ContainerInput,
  sdkEnv: Record<string, string | undefined>,
): Promise<void> {
  const agentDir = '/home/node/.pi/agent';
  fs.mkdirSync(agentDir, { recursive: true });

  const piBackend = containerInput.pi_backend || 'anthropic';

  // Build an authStorage in empty/in-memory mode — the host is exclusively
  // responsible for OAuth token refresh. The container NEVER refreshes tokens.
  // Instead we inject pre-computed API keys via setRuntimeApiKey so pi never
  // reaches its internal OAuth refresh path.
  const authStorage = AuthStorage.create(path.join(agentDir, 'auth.json'));

  // Helper: extract the API key for an OAuth pi backend from PI_AUTH_JSON.
  // Calls pi-ai's getApiKey() so we compute the correct format (e.g. JSON
  // for google-gemini-cli) without triggering any network refresh.
  function extractPiOAuthKey(piAuthJsonStr: string | undefined, backend: string): string | undefined {
    if (!piAuthJsonStr) return undefined;
    try {
      const parsed = JSON.parse(piAuthJsonStr) as Record<string, unknown>;
      const cred = parsed[backend];
      if (!cred || typeof cred !== 'object') return undefined;
      const provider = getOAuthProvider(backend);
      if (!provider) return undefined;
      return provider.getApiKey(cred as any) ?? undefined;
    } catch {
      return undefined;
    }
  }

  // Inject credentials via runtime overrides — host-managed, never container-refreshed.
  if (piBackend === 'openai' || piBackend === 'azure-openai-responses') {
    // For OAuth (CODEX_AUTH_JSON): write the full auth.json so pi handles
    // OAuth internally with the correct scopes for the Responses API.
    // Extract access_token from Codex OAuth auth.json and inject as runtime API key.
    // Pi's 'openai' provider doesn't have its own OAuth handler — only 'openai-codex' does.
    // So we extract the bearer token and inject it directly.
    const codexAuthJson = sdkEnv['CODEX_AUTH_JSON'];
    if (codexAuthJson) {
      try {
        const parsed = JSON.parse(codexAuthJson);
        if (parsed.tokens?.access_token) {
          authStorage.setRuntimeApiKey(piBackend, parsed.tokens.access_token);
          log(`Pi: injected access_token from Codex OAuth for ${piBackend}`);
        }
      } catch (err) {
        log(`Pi: failed to extract Codex access_token: ${err}`);
      }
    }
    // Fallback: plain API key
    const openaiKey = sdkEnv['OPENAI_API_KEY'];
    if (!codexAuthJson && openaiKey) authStorage.setRuntimeApiKey(piBackend, openaiKey);
  } else if (piBackend === 'google' || piBackend === 'google-vertex' || piBackend === 'google-antigravity') {
    const geminiKey = sdkEnv['GEMINI_API_KEY'];
    if (geminiKey) authStorage.setRuntimeApiKey(piBackend, geminiKey);
  } else if (piBackend === 'google-gemini-cli' || piBackend === 'github-copilot' || piBackend === 'openai-codex') {
    // OAuth backends: host injects PI_AUTH_JSON or CODEX_AUTH_JSON.
    // Try PI_AUTH_JSON first (pi-native format), then CODEX_AUTH_JSON (codex CLI format).
    let oauthKey = extractPiOAuthKey(sdkEnv['PI_AUTH_JSON'], piBackend);
    // Fallback: extract from codex-format auth.json (has tokens.access_token at top level)
    if (!oauthKey && piBackend === 'openai-codex') {
      const codexJson = sdkEnv['CODEX_AUTH_JSON'] || sdkEnv['PI_AUTH_JSON'];
      if (codexJson) {
        try {
          const parsed = JSON.parse(codexJson);
          if (parsed.tokens?.access_token) oauthKey = parsed.tokens.access_token;
        } catch { /* ignore */ }
      }
    }
    if (oauthKey) {
      authStorage.setRuntimeApiKey(piBackend, oauthKey);
      // google-gemini-cli models are registered under 'google-antigravity' in pi's ModelRegistry
      if (piBackend === 'google-gemini-cli') authStorage.setRuntimeApiKey('google-antigravity', oauthKey);
      log(`Pi OAuth key injected for backend: ${piBackend}`);
    } else {
      log(`Pi OAuth key not found for backend: ${piBackend}`);
    }
  } else if (piBackend === 'groq') {
    const groqKey = sdkEnv['GROQ_API_KEY'];
    if (groqKey) authStorage.setRuntimeApiKey('groq', groqKey);
  } else if (piBackend === 'mistral') {
    const mistralKey = sdkEnv['MISTRAL_API_KEY'];
    if (mistralKey) authStorage.setRuntimeApiKey('mistral', mistralKey);
  } else if (piBackend === 'xai') {
    const xaiKey = sdkEnv['XAI_API_KEY'];
    if (xaiKey) authStorage.setRuntimeApiKey('xai', xaiKey);
  } else {
    // anthropic (default) and others
    const apiKey = sdkEnv['ANTHROPIC_API_KEY'] || sdkEnv['ANTHROPIC_AUTH_TOKEN'];
    const oauthToken = sdkEnv['CLAUDE_CODE_OAUTH_TOKEN'];
    if (apiKey) authStorage.setRuntimeApiKey('anthropic', apiKey);
    if (oauthToken && !apiKey) authStorage.setRuntimeApiKey('anthropic', oauthToken);
  }

  const modelRegistry = new ModelRegistry(authStorage);
  // piBackend is the pi ModelRegistry provider ID (e.g. 'anthropic', 'openai', 'google', 'groq', etc.)
  const defaultModelMap: Record<string, string> = {
    'anthropic': 'claude-sonnet-4-6',
    'openai': 'gpt-4o',
    'openai-codex': 'gpt-5.2',
    'google': 'gemini-2.5-pro',
    'google-vertex': 'gemini-2.5-pro',
    'groq': 'llama-3.3-70b-versatile',
    'mistral': 'mistral-large-latest',
    'xai': 'grok-3',
    'google-gemini-cli': 'gemini-2.0-flash',
    'google-antigravity': 'gemini-2.0-flash',
    'github-copilot': 'gpt-4o',
  };
  const defaultModel = defaultModelMap[piBackend] ?? 'claude-sonnet-4-6';
  const modelId = containerInput.model || defaultModel;
  // Model lookup: try the direct backend name first. For google-gemini-cli, also try
  // 'google-antigravity' as fallback since some models (e.g. claude-sonnet-4-6) are
  // registered under that group even when accessed via the gemini-cli backend.
  let model = modelRegistry.find(piBackend, modelId);
  if (!model && piBackend === 'google-gemini-cli') {
    model = modelRegistry.find('google-antigravity', modelId);
  }
  if (!model) {
    throw new Error(`Pi: model not found for backend ${piBackend} (model=${modelId})`);
  }

  log(`Pi session starting: model=${model.id} backend=${piBackend}`);

  const promptContext = buildRuntimePromptContext(containerInput);
  if (promptContext.injectedSummary) {
    log(`Pi workspace context: ${promptContext.injectedSummary}`);
  }

  const { session } = await createAgentSession({
    cwd: '/workspace/group',
    agentDir,
    model,
    authStorage,
    modelRegistry,
    tools: createCodingTools('/workspace/group'),
    extensionFactories: [
      (pi: any) => {
        pi.on('agent_end', async (_event: any, _ctx: any) => {
          fireLcmCompactSignal();
        });
      },
    ],
  } as any);

  // Pi only supports text prompts — materialize base64 attachments to disk
  // so the agent can read them with tools, and note them in the prompt.
  function materializeAttachments(text: string, attachments: Attachment[] | undefined): string {
    if (!attachments?.length) return text;
    const attachDir = '/workspace/group/.attachments';
    fs.mkdirSync(attachDir, { recursive: true });
    const fileNotes: string[] = [];
    for (const att of attachments) {
      if (!att.data) continue;
      const safeName = att.name.replace(/[^a-zA-Z0-9._-]/g, '_');
      const filePath = path.join(attachDir, `${Date.now()}-${safeName}`);
      fs.writeFileSync(filePath, Buffer.from(att.data, 'base64'));
      fileNotes.push(`- ${att.name} (${att.mediaType}): ${filePath}`);
      log(`Pi: materialized attachment ${att.name} to ${filePath}`);
    }
    if (fileNotes.length) {
      text += `\n\n[Attached files — read them to see their contents]\n${fileNotes.join('\n')}`;
    }
    return text;
  }

  const prepared = prepareMessageInput(initialPrompt, containerInput.attachments);
  const materializedPrompt = materializeAttachments(prepared.text, prepared.attachments);

  // Prepend workspace context to the first user message
  let firstPrompt = materializedPrompt;
  if (promptContext.promptAppend) {
    firstPrompt = `<system_context>\n${promptContext.promptAppend}\n</system_context>\n\n${materializedPrompt}`;
  }

  let assistantText = '';
  let flushedAssistantText = '';
  const flushAssistantText = () => {
    if (assistantText && assistantText !== flushedAssistantText) {
      const newText = assistantText.slice(flushedAssistantText.length);
      if (newText) {
        writeOutput({ status: 'success', result: null, assistantText: newText });
      }
      flushedAssistantText = assistantText;
    }
  };
  const unsubscribe = (session as any).subscribe((event: any) => {
    if (event.type === 'message_update') {
      // Streaming providers: text arrives as text_delta events
      const e = event.assistantMessageEvent;
      if (e?.type === 'text_delta' && e.delta) {
        assistantText += e.delta;
      }
    } else if (event.type === 'message_end') {
      const msg = event.message;
      if (msg?.role === 'assistant') {
        // For non-streaming providers: full text arrives here
        if (!assistantText) {
          const content = msg.content;
          let text = '';
          if (typeof content === 'string') {
            text = content;
          } else if (Array.isArray(content)) {
            text = content.filter((c: any) => c.type === 'text').map((c: any) => c.text).join('');
          }
          if (text) {
            assistantText = text;
          }
        }
        // Log error from model
        if (msg?.stopReason === 'error' && msg?.errorMessage) {
          log(`Pi model error: ${msg.errorMessage}`);
          assistantText = `[Model error: ${msg.errorMessage}]`;
        }
        // Flush any accumulated assistant text (streaming or non-streaming)
        flushAssistantText();
      }
    } else if (event.type === 'tool_execution_start' && event.toolName) {
      // Flush assistant text before reporting tool progress
      flushAssistantText();
      const summary = formatToolProgress(event.toolName, event.args);
      if (summary) writeOutput({ status: 'success', result: null, progress: summary });
    }
  });

  const PI_SESSION_MARKER = 'pi-session';

  try {
    let prompt = firstPrompt;

    while (true) {
      // Re-read token file in case the host refreshed the OAuth credential
      refreshSdkEnvFromTokenFile(sdkEnv);
      // For OAuth pi backends, update the runtime key if PI_AUTH_JSON changed
      if (piBackend === 'google-gemini-cli' || piBackend === 'github-copilot' || piBackend === 'openai-codex') {
        const refreshedKey = extractPiOAuthKey(sdkEnv['PI_AUTH_JSON'], piBackend);
        if (refreshedKey) {
          authStorage.setRuntimeApiKey(piBackend, refreshedKey);
          if (piBackend === 'google-gemini-cli') authStorage.setRuntimeApiKey('google-antigravity', refreshedKey);
        }
      }
      assistantText = '';
      flushedAssistantText = '';
      let closedDuringQuery = false;
      let ipcPolling = true;

      const pollIpc = () => {
        if (!ipcPolling) return;
        if (shouldClose()) {
          closedDuringQuery = true;
          (session as any).abort?.();
          ipcPolling = false;
          return;
        }
        const messages = drainIpcInput();
        for (const msg of messages) {
          log(`Pi: injecting IPC message via steer() (${msg.text.length} chars)`);
          void (session as any).steer(msg.text);
        }
        setTimeout(pollIpc, IPC_POLL_MS);
      };
      setTimeout(pollIpc, IPC_POLL_MS);

      try {
        await (session as any).prompt(prompt);
      } finally {
        ipcPolling = false;
      }

      if (closedDuringQuery) break;

      writeOutput({ status: 'success', result: assistantText || null, newSessionId: PI_SESSION_MARKER });

      log('Pi: turn complete, waiting for next IPC message...');
      const nextMessage = await waitForIpcMessage();
      if (nextMessage === null) break;

      log(`Pi: next message (${nextMessage.text.length} chars)`);
      containerInput.attachments = nextMessage.attachments;
      const nextPrepared = prepareMessageInput(nextMessage.text, nextMessage.attachments);
      prompt = materializeAttachments(nextPrepared.text, nextPrepared.attachments);
    }
  } finally {
    unsubscribe();
  }
}

async function main(): Promise<void> {
  let containerInput: ContainerInput;

  try {
    const stdinData = await readStdin();
    containerInput = JSON.parse(stdinData);
    // Delete the temp file the entrypoint wrote — it contains secrets
    try { fs.unlinkSync('/tmp/input.json'); } catch { /* may not exist */ }
    log(`Received input for group: ${containerInput.groupFolder}`);
  } catch (err) {
    writeOutput({
      status: 'error',
      result: null,
      error: `Failed to parse input: ${err instanceof Error ? err.message : String(err)}`
    });
    process.exit(1);
  }

  // Build SDK env: merge secrets into process.env for the SDK only.
  // Secrets never touch process.env itself, so Bash subprocesses can't see them.
  const sdkEnv: Record<string, string | undefined> = { ...process.env };
  for (const [key, value] of Object.entries(containerInput.secrets || {})) {
    sdkEnv[key] = value;
  }
  const provider = normalizeProvider(containerInput.provider);

  // Per-tenant model selection
  containerInput.model = containerInput.model || getDefaultModel(provider);
  if (provider === 'claude') {
    sdkEnv['CLAUDE_MODEL'] = containerInput.model;
  } else if (provider !== 'pi') {
    prepareCodexAuthFiles(sdkEnv, containerInput.model);
  }

  const __dirname = path.dirname(fileURLToPath(import.meta.url));

  let sessionId = containerInput.sessionId;
  fs.mkdirSync(IPC_INPUT_DIR, { recursive: true });

  // Clean up stale _close sentinel from previous container runs
  try { fs.unlinkSync(IPC_INPUT_CLOSE_SENTINEL); } catch { /* ignore */ }

  // Warm-only mode: skip initial query, emit session marker, wait for IPC
  if (containerInput.warmOnly) {
    log('Warm-only mode: container ready, waiting for IPC messages...');
    writeOutput({ status: 'success', result: null, newSessionId: sessionId });

    const firstMessage = await waitForIpcMessage();
    if (firstMessage === null) {
      log('Close sentinel received during warm wait, exiting');
      return;
    }
    log(`Warm container received first message (${firstMessage.text.length} chars)`);
    // Fall through to the normal query loop with this message as the prompt
    containerInput.prompt = firstMessage.text;
    containerInput.attachments = firstMessage.attachments;
  }

  // Build initial prompt (drain any pending IPC messages too)
  let prompt = containerInput.prompt;
  if (containerInput.isScheduledTask) {
    prompt = `[SCHEDULED TASK - The following message was sent automatically and is not coming directly from the user or group.]\n\nIMPORTANT: Your text output will be automatically delivered to the user — do NOT also call \`tc send\` to deliver the same message. Only use \`tc send\` if you need to send to a DIFFERENT channel than the one that triggered this task.\n\n${prompt}`;
  }
  const pending = drainIpcInput();
  if (pending.length > 0) {
    log(`Draining ${pending.length} pending IPC messages into initial prompt`);
    prompt += '\n' + pending.map(m => m.text).join('\n');
    const pendingAtts = pending.flatMap(m => m.attachments || []);
    if (pendingAtts.length > 0 && !containerInput.attachments) {
      containerInput.attachments = pendingAtts;
    } else if (pendingAtts.length > 0 && containerInput.attachments) {
      containerInput.attachments = [...containerInput.attachments, ...pendingAtts];
    }
  }

  // Pi provider: delegates entirely to its own session management loop
  if (provider === 'pi') {
    await runPiSession(prompt, containerInput, sdkEnv);
    return;
  }

  // Query loop: run query → wait for IPC message → run new query → repeat
  let resumeAt: string | undefined;
  const MAX_AUTH_RETRIES = 3;
  const AUTH_RETRY_DELAYS_MS = [1000, 3000, 5000];
  let consecutiveAuthFailures = 0;
  try {
    while (true) {
      // Re-read token file in case the host refreshed the OAuth credential
      refreshSdkEnvFromTokenFile(sdkEnv);
      if (provider === 'codex') {
        prepareCodexAuthFiles(sdkEnv);
      }

      log(`Starting query (session: ${sessionId || 'new'}, resumeAt: ${resumeAt || 'latest'})...`);

      const queryResult = provider === 'codex'
        ? await runCodexTurn(prompt, sessionId, containerInput, sdkEnv)
        : await runQuery(prompt, sessionId, containerInput, sdkEnv, resumeAt);
      if (queryResult.newSessionId) {
        sessionId = queryResult.newSessionId;
      }
      if (provider === 'claude') {
        const claudeResult = queryResult as Awaited<ReturnType<typeof runQuery>>;
        resumeAt = claudeResult.lastAssistantUuid;
      }

      // Auth error recovery: re-read token file and retry with backoff
      if (queryResult.authError) {
        consecutiveAuthFailures++;
        log(`Auth failure ${consecutiveAuthFailures}/${MAX_AUTH_RETRIES}`);

        if (consecutiveAuthFailures > MAX_AUTH_RETRIES) {
          log('Max auth retries exceeded, exiting container');
          writeOutput({
            status: 'error',
            result: null,
            error: 'Authentication failed after multiple retries. The container will restart on your next message.',
          });
          process.exit(1);
        }

        // Write IPC task asking nanoclaw to do a fresh OAuth refresh for this group
        try {
          const refreshTaskPath = `/workspace/ipc/tasks/refresh_cred_${Date.now()}.json`;
          const tempPath = refreshTaskPath + ".tmp";
          fs.writeFileSync(tempPath, JSON.stringify({ type: "refresh_credentials" }));
          fs.renameSync(tempPath, refreshTaskPath);
          log(`Wrote refresh_credentials IPC task, waiting 5s for host to refresh token...`);
        } catch (writeErr) {
          log(`Failed to write refresh_credentials IPC task: ${writeErr}`);
        }
        // Wait 5s — nanoclaw polls every 2s, enough to refresh and write fresh token
        await new Promise(resolve => setTimeout(resolve, 5000));
        // re-read token file — nanoclaw should have written fresh token
        continue;
      }

      // Reset auth failure counter on success
      consecutiveAuthFailures = 0;

      // If _close was consumed during the query, exit immediately.
      // Don't emit a session-update marker (it would reset the host's
      // idle timer and cause a 30-min delay before the next _close).
      if (queryResult.closedDuringQuery) {
        log('Close sentinel consumed during query, exiting');
        break;
      }

      // Emit session update so host can track it
      writeOutput({ status: 'success', result: null, newSessionId: sessionId });

      log('Query ended, waiting for next IPC message...');

      // Wait for the next message or _close sentinel
      const nextMessage = await waitForIpcMessage();
      if (nextMessage === null) {
        log('Close sentinel received, exiting');
        break;
      }

      log(`Got new message (${nextMessage.text.length} chars, ${nextMessage.attachments?.length || 0} attachments), starting new query`);
      prompt = nextMessage.text;
      containerInput.attachments = nextMessage.attachments;
    }
  } catch (err) {
    const errorMessage = err instanceof Error ? err.message : String(err);
    log(`Agent error: ${errorMessage}`);
    writeOutput({
      status: 'error',
      result: null,
      newSessionId: sessionId,
      error: errorMessage
    });
    process.exit(1);
  }
}

const entrypointPath = process.argv[1] ? path.resolve(process.argv[1]) : '';
if (entrypointPath === fileURLToPath(import.meta.url)) {
  main();
}
