executor.ts

import { execSync, spawn } from 'node:child_process';
import fs from 'node:fs';
import os from 'node:os';
import path from 'node:path';
import { query } from '@anthropic-ai/claude-agent-sdk';
import type { SDKUserMessage, SpawnOptions, SpawnedProcess } from '@anthropic-ai/claude-agent-sdk';
import type { BotConfigBase } from '../../config.js';
import type { Logger } from '../../utils/logger.js';
import { AsyncQueue } from '../../utils/async-queue.js';

const isWindows = process.platform === 'win32';

/** Resolve the Claude Code binary path at module load time. */
function resolveClaudePath(): string {
  if (process.env.CLAUDE_EXECUTABLE_PATH) return process.env.CLAUDE_EXECUTABLE_PATH;
  try {
    const cmd = isWindows ? 'where claude' : 'which claude';
    return execSync(cmd, { encoding: 'utf-8' }).trim().split(/\r?\n/)[0];
  } catch {
    return isWindows ? 'claude' : '/usr/local/bin/claude';
  }
}

const CLAUDE_EXECUTABLE = resolveClaudePath();

/**
 * Env var prefixes to always strip from the inherited process environment.
 * CLAUDE*: prevents "nested session" errors from the SDK.
 */
const ALWAYS_FILTERED_PREFIXES = ['CLAUDE'];

/**
 * Specific CLAUDE_* env vars that are SAFE to pass through to the child
 * Claude Code process even though the broad CLAUDE* filter would normally
 * strip them. These are user-tunable feature flags / mode toggles, not
 * session-state vars (which are what the nested-session guard is for).
 *
 * Add a var here when you need MetaBot users to be able to enable a
 * Claude Code feature via .env or the host environment.
 */
const CLAUDE_ENV_PASSTHROUGH = new Set([
  'CLAUDE_CODE_EXPERIMENTAL_AGENT_TEAMS', // /agent teams (multi-instance coordination)
  'CLAUDE_CODE_DISABLE_AGENT_VIEW',       // disable claude agents / --bg / /background
  'CLAUDE_CODE_SIMPLE',                   // --bare equivalent
  'CLAUDE_CODE_DISABLE_AUTO_MEMORY',      // toggle auto-memory (project patterns/learnings)
]);

/**
 * Auth-related env vars that are only filtered when an explicit API key
 * is provided in bots.json OR when ~/.claude/.credentials.json exists.
 * This ensures users who rely solely on ANTHROPIC_API_KEY env var can
 * still authenticate without configuring bots.json.
 */
const AUTH_ENV_VARS = ['ANTHROPIC_API_KEY', 'ANTHROPIC_AUTH_TOKEN'];

/**
 * Check if Claude Code has credentials.json (OAuth login).
 */
function hasCredentialsFile(): boolean {
  const credPath = path.join(os.homedir(), '.claude', '.credentials.json');
  try {
    return fs.existsSync(credPath);
  } catch {
    return false;
  }
}

/**
 * Create a custom spawn function for cross-platform compatibility.
 * - Honors `options.command` from the SDK — for claude-agent-sdk >= 0.2.140
 *   the SDK spawns the native Claude binary directly, so we must NOT force
 *   `process.execPath` (node). Legacy JS entrypoints set `options.command`
 *   to the node executable themselves, so this works in both worlds.
 * - Always filters CLAUDE* env vars to prevent nested session errors.
 * - Filters ANTHROPIC auth env vars only when an explicit API key is provided
 *   or credentials.json exists (so env-var-only users can still authenticate).
 * - Merges process.env so child inherits system PATH, TEMP, etc.
 * - Optionally injects an explicit ANTHROPIC_API_KEY from bots.json config.
 */
function createSpawnFn(explicitApiKey?: string): (options: SpawnOptions) => SpawnedProcess {
  // Force-use-env mode: pass ANTHROPIC_AUTH_TOKEN / ANTHROPIC_API_KEY /
  // ANTHROPIC_BASE_URL through to the Claude Code subprocess instead of
  // filtering them out. Triggered by either:
  //   (a) METABOT_PREFER_ENV_AUTH=true (explicit opt-in flag), or
  //   (b) presence of ANTHROPIC_AUTH_TOKEN / ANTHROPIC_API_KEY / ANTHROPIC_BASE_URL
  //       in the process env (auto-detect — user clearly wants env-based auth).
  // Use case: a bot points Claude Code at a third-party Anthropic-compatible
  // proxy while other bots on the same machine still use OAuth via
  // ~/.claude/.credentials.json (which can't be deleted).
  const preferEnvAuth =
    process.env.METABOT_PREFER_ENV_AUTH === 'true' ||
    !!(
      process.env.ANTHROPIC_AUTH_TOKEN ||
      process.env.ANTHROPIC_API_KEY ||
      process.env.ANTHROPIC_BASE_URL
    );

  // Decide once whether to filter auth env vars
  const filterAuthVars = !preferEnvAuth && !!(explicitApiKey || hasCredentialsFile());

  return (options: SpawnOptions): SpawnedProcess => {
    // Merge provided env with process.env for a complete environment
    const baseEnv = options.env && Object.keys(options.env).length > 0
      ? { ...process.env, ...options.env }
      : { ...process.env };

    // Filter out env vars that interfere with auth or cause nested session errors
    const env: Record<string, string> = {};
    for (const [key, value] of Object.entries(baseEnv)) {
      if (value === undefined) continue;
      // Safe-pass list takes precedence over the broad CLAUDE* strip — these
      // are feature flags users opt into (agent teams, disable agent view, etc.)
      if (CLAUDE_ENV_PASSTHROUGH.has(key)) {
        env[key] = value;
        continue;
      }
      if (ALWAYS_FILTERED_PREFIXES.some(p => key.startsWith(p))) continue;
      if (filterAuthVars && AUTH_ENV_VARS.some(v => key.startsWith(v))) continue;
      env[key] = value;
    }

    // Inject explicit API key from bots.json (after filtering, so it takes effect)
    if (explicitApiKey) {
      env.ANTHROPIC_API_KEY = explicitApiKey;
    }

    // Default-enable Claude Code Agent Teams. Without a real terminal there's
    // no tmux/iTerm2, so teammates must run in-process (controlled via the
    // `teammateMode` setting passed in queryOptions). Users can disable by
    // setting CLAUDE_CODE_EXPERIMENTAL_AGENT_TEAMS=0 in MetaBot's parent env.
    if (env.CLAUDE_CODE_EXPERIMENTAL_AGENT_TEAMS === undefined) {
      env.CLAUDE_CODE_EXPERIMENTAL_AGENT_TEAMS = '1';
    }

    // Default-enable Claude Code auto-memory so Claude can write project
    // patterns / preferences / decisions to ~/.claude/projects/<projDir>/memory/
    // across sessions — the user-facing memory system the bot's skills
    // rely on. Users can disable by setting
    // CLAUDE_CODE_DISABLE_AUTO_MEMORY=1 in MetaBot's parent env.
    // Pinning to '0' here makes the feature immune to upstream default
    // changes; the user shouldn't need to keep a magic line in .env.
    if (env.CLAUDE_CODE_DISABLE_AUTO_MEMORY === undefined) {
      env.CLAUDE_CODE_DISABLE_AUTO_MEMORY = '0';
    }

    const child = spawn(options.command, options.args, {
      cwd: options.cwd,
      env,
      signal: options.signal,
      stdio: ['pipe', 'pipe', 'pipe'],
    });

    return child as unknown as SpawnedProcess;
  };
}

export interface ApiContext {
  botName: string;
  chatId: string;
  /** Group chat member names — enables inter-bot communication prompt. */
  groupMembers?: string[];
  /** Group ID — used to build grouptalk chatIds for inter-bot communication. */
  groupId?: string;
}

/**
 * Events surfaced by Claude Code's experimental Agent Teams hooks
 * (TaskCreated / TaskCompleted / TeammateIdle). Used to drive the
 * Feishu / Web team panel without requiring the user to switch panes.
 */
export type TeamEvent =
  | {
      kind: 'task_created';
      taskId: string;
      subject: string;
      description?: string;
      teammate?: string;
      teamName?: string;
    }
  | {
      kind: 'task_completed';
      taskId: string;
      subject: string;
      teammate?: string;
      teamName?: string;
    }
  | {
      kind: 'teammate_idle';
      teammate: string;
      teamName: string;
    };

export interface ExecutorOptions {
  prompt: string;
  cwd: string;
  sessionId?: string;
  abortController: AbortController;
  outputsDir?: string;
  apiContext?: ApiContext;
  /** Override maxTurns for this execution. */
  maxTurns?: number;
  /** Override model for this execution (e.g. faster model for voice calls). */
  model?: string;
  /** Override allowed tools for this execution (empty array = no tools). */
  allowedTools?: string[];
  /** Called whenever Claude Code fires a team coordination hook. */
  onTeamEvent?: (event: TeamEvent) => void;
}

export type SDKMessage = {
  type: string;
  subtype?: string;
  uuid?: string;
  session_id?: string;
  message?: {
    content?: Array<{
      type: string;
      text?: string;
      name?: string;
      id?: string;
      input?: unknown;
    }>;
  };
  // Result fields
  duration_ms?: number;
  duration_api_ms?: number;
  total_cost_usd?: number;
  result?: string;
  is_error?: boolean;
  num_turns?: number;
  errors?: string[];
  // Model usage from result message (per-model breakdown)
  modelUsage?: Record<string, { inputTokens: number; outputTokens: number; contextWindow: number; costUSD: number }>;
  // Stream event fields
  event?: {
    type: string;
    index?: number;
    delta?: {
      type: string;
      text?: string;
    };
    content_block?: {
      type: string;
      text?: string;
      name?: string;
      id?: string;
    };
  };
  parent_tool_use_id?: string | null;
};

export interface ExecutionHandle {
  stream: AsyncGenerator<SDKMessage>;
  sendAnswer(toolUseId: string, sessionId: string, answerText: string): void;
  /**
   * Resolve a pending AskUserQuestion PreToolUse hook with the user's answers.
   * Use this instead of sendAnswer when running in bypassPermissions mode —
   * sendAnswer enqueues a tool_result that never reaches the SDK because the
   * internal permission check short-circuits before auto-allow.
   */
  resolveQuestion(toolUseId: string, answers: Record<string, string>): void;
  finish(): void;
}

export class ClaudeExecutor {
  constructor(
    private config: BotConfigBase,
    private logger: Logger,
  ) {}

  private buildQueryOptions(cwd: string, sessionId: string | undefined, abortController: AbortController, outputsDir?: string, apiContext?: ApiContext): Record<string, unknown> {
    const isRoot = process.getuid?.() === 0;
    const queryOptions: Record<string, unknown> = {
      permissionMode: isRoot ? 'auto' : ('bypassPermissions' as const),
      ...(isRoot ? {} : { allowDangerouslySkipPermissions: true }),
      cwd,
      abortController,
      includePartialMessages: true,
      // Load MCP servers and settings from user/project config files
      settingSources: ['user', 'project'],
      // Custom spawn filters CLAUDE* env vars (prevents nested session errors)
      // and injects an explicit ANTHROPIC_API_KEY when configured. The SDK
      // (>= 0.2.140) supplies the correct command in spawn options — for the
      // native Claude binary that's the binary itself; for legacy JS
      // entrypoints it's the Node executable.
      spawnClaudeCodeProcess: createSpawnFn(this.config.claude.apiKey),
      pathToClaudeCodeExecutable: CLAUDE_EXECUTABLE,
      // MetaBot has no terminal — split-pane (tmux/iTerm2) teammate display
      // doesn't apply. Force in-process so teammates run inside the same
      // session and surface via SDK message origin / TeammateIdle hooks.
      settings: { teammateMode: 'in-process' },
      // Periodic AI summaries for foreground/background subagents. The SDK
      // emits these as `task_progress.summary`; StreamProcessor already
      // forwards task events into the card's "Background" panel, so enabling
      // this immediately makes subagent cards richer (Agent View parity).
      agentProgressSummaries: true,
    };

    // Build system prompt appendix from sections
    const appendSections: string[] = [];

    if (outputsDir) {
      appendSections.push(`## Output Files\nWhen producing output files for the user (images, PDFs, documents, archives, code files, etc.), copy them to: ${outputsDir}\nUse \`cp\` via the Bash tool. The bridge will automatically send files placed there to the user.`);
    }

    if (apiContext) {
      // botName and chatId are per-session — inject into system prompt to avoid
      // race conditions when multiple chats run concurrently.
      // Port and secret are already set as METABOT_* env vars in config.ts.
      appendSections.push(
        `## MetaBot API\nYou are running as bot "${apiContext.botName}" in chat "${apiContext.chatId}".\nUse the /metabot skill for full API documentation (agent bus, scheduling, bot management).`
      );

      // Agent Teams namespace guidance: the team config lives at
      // ~/.claude/teams/{name}/, which is shared across all bots and chats
      // on the same host. Tell the lead to namespace team names so concurrent
      // bots/chats don't collide.
      const teamNs = `${apiContext.botName}-${apiContext.chatId.slice(0, 8)}`;
      appendSections.push(
        [
          '## Agent Teams (experimental)',
          `When the user asks you to create an agent team, ALWAYS prefix the team name with \`${teamNs}-\` to avoid collisions with other MetaBot chats sharing this machine. For example: \`${teamNs}-research\`, \`${teamNs}-refactor\`.`,
          'Display mode is forced to `in-process` (no tmux/iTerm2 in MetaBot). Teammates show up in the user\'s Feishu card via TeammateIdle / TaskCreated / TaskCompleted events — you don\'t need to walk the user through Shift+Down navigation.',
          'Clean up the team yourself when work is done so resources don\'t leak (`Clean up the team`).',
        ].join('\n')
      );

      // Group chat — tell the bot who else is in the group and how to talk to them
      if (apiContext.groupMembers && apiContext.groupMembers.length > 0) {
        const others = apiContext.groupMembers.filter((m) => m !== apiContext.botName);
        const groupId = apiContext.groupId;
        if (groupId) {
          appendSections.push(
            `## Group Chat\nYou are in a group chat (group: ${groupId}) with these bots: ${others.join(', ')}.\nTo talk to another bot, use: \`mb talk <botName> grouptalk-${groupId}-<botName> "message"\`\nExample: \`mb talk ${others[0]} grouptalk-${groupId}-${others[0]} "hello"\`\nIMPORTANT: Always use the grouptalk-${groupId}-<botName> chatId pattern when talking to other bots in this group.`
          );
        } else {
          appendSections.push(
            `## Group Chat\nYou are in a group chat with these bots: ${others.join(', ')}.\nUse \`mb talk <botName> <chatId> "message"\` to communicate with other bots in the group.`
          );
        }
      }
    }

    if (appendSections.length > 0) {
      queryOptions.systemPrompt = {
        type: 'preset',
        preset: 'claude_code',
        append: '\n\n' + appendSections.join('\n\n'),
      };
    }

    if (this.config.claude.maxTurns !== undefined) {
      queryOptions.maxTurns = this.config.claude.maxTurns;
    }

    if (this.config.claude.maxBudgetUsd !== undefined) {
      queryOptions.maxBudgetUsd = this.config.claude.maxBudgetUsd;
    }

    if (this.config.claude.model) {
      queryOptions.model = this.config.claude.model;
    }

    if (sessionId) {
      queryOptions.resume = sessionId;
    }

    // 1M context window is opt-in via the `[1m]` model-name suffix
    // (`claude-opus-4-7[1m]`). The suffix is the canonical signal — the SDK
    // parses it directly on both OAuth and API-key auth paths.
    //
    // We *also* set the matching `betas` flag when the suffix is present,
    // belt-and-braces: harmless when the SDK already inferred it, and a
    // safety net if a future SDK rev only honors the explicit beta header
    // for some auth modes. Without the suffix, leave `betas` unset —
    // setting it unconditionally on API-key auth had the side-effect of
    // forcing every model (e.g. plain `opus-4-7`) into 1M context at 2×
    // the price, which the user never asked for.
    if (this.config.claude.model?.includes('[1m]')) {
      queryOptions.betas = ['context-1m-2025-08-07'];
    }

    return queryOptions;
  }

  startExecution(options: ExecutorOptions): ExecutionHandle {
    const { prompt, cwd, sessionId, abortController, outputsDir, apiContext } = options;

    this.logger.info({ cwd, hasSession: !!sessionId, outputsDir }, 'Starting Claude execution (multi-turn)');

    const inputQueue = new AsyncQueue<SDKUserMessage>();

    // Push the initial user message
    const initialMessage: SDKUserMessage = {
      type: 'user',
      message: {
        role: 'user' as const,
        content: prompt,
      },
      parent_tool_use_id: null,
      session_id: sessionId || '',
    };
    inputQueue.enqueue(initialMessage);

    const queryOptions = this.buildQueryOptions(cwd, sessionId, abortController, outputsDir, apiContext);
    if (options.maxTurns !== undefined) {
      queryOptions.maxTurns = options.maxTurns;
    }
    if (options.model) {
      queryOptions.model = options.model;
    }
    if (options.allowedTools !== undefined) {
      queryOptions.allowedTools = options.allowedTools;
    }

    // AskUserQuestion PreToolUse hook: the SDK marks AskUserQuestion as
    // requiresUserInteraction=true, so in bypassPermissions mode it is denied
    // before auto-allow can fire. We intercept the PreToolUse event, pause until
    // the bridge collects the user's answers, then return them as updatedInput.
    // Providing updatedInput satisfies the interaction requirement and the SDK
    // resolves the tool call with {answers} filled in.
    const pendingQuestionResolvers = new Map<string, (answers: Record<string, string>) => void>();

    const askUserQuestionHook = async (
      input: { hook_event_name: string; tool_name: string; tool_input: unknown; tool_use_id: string },
      _toolUseId: string | undefined,
      { signal }: { signal: AbortSignal },
    ): Promise<Record<string, unknown>> => {
      const toolInput = input.tool_input as Record<string, unknown>;
      const id = input.tool_use_id;

      const answers = await new Promise<Record<string, string>>((resolve) => {
        pendingQuestionResolvers.set(id, resolve);

        // Safety timeout: auto-resolve with empty answers after 6 minutes
        // (slightly longer than bridge's 5-minute QUESTION_TIMEOUT_MS) to
        // prevent indefinite hang if the bridge fails to deliver an answer.
        const timeout = setTimeout(() => {
          if (pendingQuestionResolvers.delete(id)) {
            logger.warn({ toolUseId: id }, 'AskUserQuestion hook timed out after 6 minutes — returning empty answers');
            resolve({});
          }
        }, 6 * 60 * 1000);

        const onAbort = () => {
          clearTimeout(timeout);
          pendingQuestionResolvers.delete(id);
          resolve({});
        };
        signal.addEventListener('abort', onAbort, { once: true });
      });

      return {
        hookSpecificOutput: {
          hookEventName: 'PreToolUse',
          permissionDecision: 'allow',
          updatedInput: { ...toolInput, answers },
        },
      };
    };

    // Agent Teams observation hooks. These never block — they just tap the
    // event so we can re-render the team panel in the Feishu / Web card.
    // Returning {} (no decision) lets the underlying action proceed.
    const onTeamEvent = options.onTeamEvent;
    const teamObserverHook = (kind: TeamEvent['kind']) => {
      return async (input: any): Promise<Record<string, unknown>> => {
        if (!onTeamEvent) return {};
        try {
          if (kind === 'task_created') {
            onTeamEvent({
              kind: 'task_created',
              taskId: input.task_id,
              subject: input.task_subject,
              description: input.task_description,
              teammate: input.teammate_name,
              teamName: input.team_name,
            });
          } else if (kind === 'task_completed') {
            onTeamEvent({
              kind: 'task_completed',
              taskId: input.task_id,
              subject: input.task_subject,
              teammate: input.teammate_name,
              teamName: input.team_name,
            });
          } else if (kind === 'teammate_idle') {
            onTeamEvent({
              kind: 'teammate_idle',
              teammate: input.teammate_name,
              teamName: input.team_name,
            });
          }
        } catch (err) {
          this.logger.warn({ err, kind }, 'Team observer hook callback threw');
        }
        return {};
      };
    };

    queryOptions.hooks = {
      PreToolUse: [{
        matcher: 'AskUserQuestion',
        hooks: [askUserQuestionHook as any],
      }],
      TaskCreated: [{ hooks: [teamObserverHook('task_created') as any] }],
      TaskCompleted: [{ hooks: [teamObserverHook('task_completed') as any] }],
      TeammateIdle: [{ hooks: [teamObserverHook('teammate_idle') as any] }],
    };

    const stream = query({
      prompt: inputQueue,
      options: queryOptions as any,
    });

    const logger = this.logger;

    async function* wrapStream(): AsyncGenerator<SDKMessage> {
      // Race each stream.next() against the abort signal so we exit immediately on /stop
      const abortPromise = new Promise<never>((_, reject) => {
        if (abortController.signal.aborted) {
          reject(new DOMException('Aborted', 'AbortError'));
          return;
        }
        abortController.signal.addEventListener('abort', () => {
          reject(new DOMException('Aborted', 'AbortError'));
        }, { once: true });
      });

      const iterator = stream[Symbol.asyncIterator]();

      try {
        while (true) {
          const result = await Promise.race([
            iterator.next(),
            abortPromise,
          ]);
          if (result.done) break;
          yield result.value as SDKMessage;
        }
      } catch (err: any) {
        if (err.name === 'AbortError' || abortController.signal.aborted) {
          logger.info('Claude execution aborted');
          // Clean up the underlying iterator (non-blocking)
          try { iterator.return?.(undefined); } catch { /* ignore */ }
          return;
        }
        throw err;
      }
    }

    return {
      stream: wrapStream(),
      sendAnswer: (toolUseId: string, sid: string, answerText: string) => {
        logger.info({ toolUseId }, 'Sending answer to Claude');
        const answerMessage: SDKUserMessage = {
          type: 'user',
          message: {
            role: 'user' as const,
            content: [
              {
                type: 'tool_result',
                tool_use_id: toolUseId,
                content: answerText,
              },
            ],
          },
          parent_tool_use_id: null,
          session_id: sid,
        };
        inputQueue.enqueue(answerMessage);
      },
      resolveQuestion: (toolUseId: string, answers: Record<string, string>) => {
        const resolver = pendingQuestionResolvers.get(toolUseId);
        if (resolver) {
          pendingQuestionResolvers.delete(toolUseId);
          logger.info({ toolUseId, answerCount: Object.keys(answers).length }, 'Resolving AskUserQuestion hook');
          resolver(answers);
        } else {
          // Fallback: enqueue tool_result via inputQueue. Used if the hook
          // didn't capture this toolUseId (e.g., legacy sendAnswer path) or
          // the SDK version differs.
          logger.warn({ toolUseId }, 'No pending AskUserQuestion resolver — falling back to sendAnswer path');
          const answerMessage: SDKUserMessage = {
            type: 'user',
            message: {
              role: 'user' as const,
              content: [{ type: 'tool_result', tool_use_id: toolUseId, content: JSON.stringify({ answers }) }],
            },
            parent_tool_use_id: null,
            session_id: '',
          };
          inputQueue.enqueue(answerMessage);
        }
      },
      finish: () => {
        inputQueue.finish();
      },
    };
  }

  async *execute(options: ExecutorOptions): AsyncGenerator<SDKMessage> {
    const { prompt, cwd, sessionId, abortController, outputsDir } = options;

    this.logger.info({ cwd, hasSession: !!sessionId }, 'Starting Claude execution');

    const queryOptions = this.buildQueryOptions(cwd, sessionId, abortController, outputsDir);

    const stream = query({
      prompt,
      options: queryOptions as any,
    });

    const abortPromise = new Promise<never>((_, reject) => {
      if (abortController.signal.aborted) {
        reject(new DOMException('Aborted', 'AbortError'));
        return;
      }
      abortController.signal.addEventListener('abort', () => {
        reject(new DOMException('Aborted', 'AbortError'));
      }, { once: true });
    });

    const iterator = stream[Symbol.asyncIterator]();

    try {
      while (true) {
        const result = await Promise.race([
          iterator.next(),
          abortPromise,
        ]);
        if (result.done) break;
        yield result.value as SDKMessage;
      }
    } catch (err: any) {
      if (err.name === 'AbortError' || abortController.signal.aborted) {
        this.logger.info('Claude execution aborted');
        try { iterator.return?.(undefined); } catch { /* ignore */ }
        return;
      }
      throw err;
    }
  }
}