runner.ts

import { EventEmitter } from "node:events";
import { createInterface } from "node:readline/promises";
import { stdin as input, stdout as output, stderr } from "node:process";
import { join } from "node:path";
import { type Agent } from "@intx/agent";
import { getLogger } from "@intx/log";
import { type Config } from "../config/index.js";
import {
  shellTimeoutFromSettings,
  toolWatchdogFromSettings,
  type MCPServerConfig,
} from "../config/settings.js";
import {
  codexProfileFromProviderName,
  isCodexProviderName,
} from "../config/codex-providers.js";
import { xaiProfileFromProviderName } from "../config/xai-providers.js";
import {
  peekSourceCredentialSecret,
  registerSourceCredential,
} from "../config/source-credentials.js";
import { formatDirectorSystemPrompt } from "../agent/directors/identity.js";
import { DIRECTOR_REGISTRY } from "../agent/directors/registry.js";
import type { DirectorId, DirectorPackage } from "../agent/directors/types.js";
import { submitOutputDefinition } from "../agent/director.js";
import { handleChatDirectorEvent } from "../agent/chat-event-subscribers.js";
import {
  shellDefinition,
  updatePlanDefinition,
} from "../agent/codex-tool-proxies.js";
import {
  CodexRefreshLockError,
  codexAuthFailureDiagnostic,
  getValidCodexToken,
} from "../auth/codex/session.js";
import { getValidXaiToken } from "../auth/xai/session.js";
import {
  type ActivatedToolTracker,
  type ToolAvailability,
} from "../agent/tool-search.js";
import { detectLanguageServerAvailable } from "../agent/lsp-availability.js";
import {
  resolveSessionMode,
  type SessionMode,
} from "../config/session-mode.js";
import {
  createSubAgentSessionStore,
  type SubAgentProvider,
  type SubAgentSessionStore,
} from "../subagent/index.js";
import { getProcessAdmissionQueue } from "../subagent/admission.js";
import { disposeExecRuntime, formatCaughtError } from "./dispose.js";
import {
  EXEC_MCP_CONNECT_WAIT_MS,
  EXEC_MCP_HANDSHAKE_TIMEOUT_MS,
  armExecMcpHandshakeAbort,
  awaitExecMcpThenResume,
  followExecMcpHandshake,
} from "./mcp-handshake.js";
import type {
  ContextStore,
  InferenceSource,
  InboundMessage,
} from "@intx/types/runtime";
import { OPERATOR_ORIGINATED_FLAG } from "../agent/message-provenance.js";
import { loadAgentProfiles } from "../agent/profiles.js";
import type {
  ApprovalOutcome,
  PermissionRequest,
} from "../permission/types.js";
import { savedSkipPermissionsWarning } from "../permission/saved-skip-warning.js";
import {
  createAgentToolset,
  type AgentToolset,
  type OperatorResult,
} from "../agent/tools.js";
import { liveTelemetry } from "../telemetry/singleton.js";
import {
  CREDENTIAL_FAILURE_USER_MESSAGE,
  isResolvedProviderFailureError,
  terminalProviderFailureMessage,
} from "../inference-error-message.js";
import type { InferenceErrorLike } from "../inference-gateway-error.js";
import {
  collectToolPlugins,
  resolveToolPlugins,
} from "../plugins/tool-plugins.js";
import {
  formatExpandSkip,
  type ExpandPluginPathSkip,
} from "../plugins/loader.js";
import { consumeStream } from "../session/stream-consumer.js";
import { COMPACTION_CONTINUATION_EVENT } from "../agent/compaction.js";
import {
  generateSessionId,
  initSessionDir,
  sessionContextDir,
  sessionDir,
} from "../session/index.js";
import {
  setActiveRun,
  syncRunStateHandle,
  type RunStateHandle,
} from "../session/active-run.js";
import { startRunHeartbeat } from "../session/run-liveness.js";
import {
  setActiveDisposeHost,
  clearActiveDisposeHost,
} from "../session/active-host.js";
import {
  finalizeRunState,
  loadState,
  saveState,
  type ConnectedMcpServer,
} from "../session/state.js";
import { resolveExecRunStatus, type RunSink } from "../session/run-sink.js";
import { createRunSummary } from "../session/hooks.js";
import {
  buildCompactionContinuationMessage,
  buildShellBackgroundMessage,
  buildSubAgentProvider,
  createContinuationGate,
  createSessionPruningCompactor,
  loadSessionChatPrompt,
  skillDirsFromEnabledPlugins,
} from "../session/runtime-assembly.js";
import {
  createApprovalResume,
  resolveParkedCallIdFromStore,
} from "../session/approval-resume.js";
import { createReactorAuthorize } from "../permission/reactor-authorize.js";
import {
  assembleChatAgent,
  assembleInferenceBase,
  assembleSessionGate,
  assembleSessionLifecycle,
  assembleSessionTrust,
  createAdvertisedToolset,
  loadSessionLocalSettings,
  resolveLiveSessionSources,
} from "../session/assemble-runtime.js";
import type { CompactionArchive } from "../session/compaction-archive.js";
import { tryReadPriorHandoffFile } from "../session/compaction-handoff.js";
import { emitPluginWarningSummary } from "../plugins/diagnostics.js";
import { createModelSummarizer } from "../session/summarizer.js";
import { ID_PREFIX, LOG_NAMESPACE_ROOT } from "../branding.js";
import {
  anthropicCacheWriteAt,
  resumeCacheWriteSeed,
} from "../provider/cache-ttl.js";
import type { ReactorEmittedEvent } from "@intx/inference";
import { setAgentSourceUnlessClosed } from "../tui/agent-source-sync.js";
import { ensureFreshInferenceSource } from "../subagent/refresh-inference-source.js";
import {
  MAX_TOOL_APPROVAL_PAUSE_MS,
  getToolApprovalBudget,
} from "../tui/tool-execution-watchdog.js";
import { createApprovalDeliverer } from "../tui/approval-delivery.js";
import { createCorrelationAcceptance } from "../tui/correlation-acceptance.js";
import { APPROVAL_TIMEOUT_RESULT_TEXT } from "../permission/decline-markers.js";
import { WorkflowHost } from "../workflows/host.js";

const logger = getLogger([LOG_NAMESPACE_ROOT, "exec"]);

const SELECTED_PROVIDER_FAILURE = "SelectedProviderFailure";

export function formatExecMcpTrustQuestion(server: MCPServerConfig): string {
  return (
    `Trust local MCP server "${server.name}" for this project?` +
    (server.command !== undefined
      ? `\nCommand: ${server.command}${(server.args ?? []).length > 0 ? ` ${(server.args ?? []).join(" ")}` : ""}`
      : server.url !== undefined
        ? `\nURL: ${server.url}`
        : "")
  );
}

export async function refreshSelectedProviderCredential<T>(
  refresh: () => Promise<T>,
): Promise<T> {
  try {
    return await refresh();
  } catch (cause) {
    const error = new Error(formatCaughtError(cause), { cause });
    error.name = SELECTED_PROVIDER_FAILURE;
    throw error;
  }
}

function execTerminalProviderFailureMessage(
  config: Config,
  diagnostic: InferenceErrorLike,
): string {
  const providerId = diagnostic.providerId ?? config.providerName;
  const displayLabel =
    providerId === config.providerName
      ? config.settings?.providers[config.providerName]?.name
      : undefined;
  return terminalProviderFailureMessage(providerId, diagnostic, displayLabel);
}

// Unwraps a Codex refresh-lock failure whether it arrives raw or as the
// cause of a SELECTED_PROVIDER_FAILURE wrapper.
function codexRefreshLockFailure(err: unknown): CodexRefreshLockError | null {
  if (err instanceof CodexRefreshLockError) return err;
  if (
    err instanceof Error &&
    err.name === SELECTED_PROVIDER_FAILURE &&
    err.cause instanceof CodexRefreshLockError
  )
    return err.cause;
  return null;
}

export function execUserFailureMessage(
  config: Config,
  err: unknown,
  providerFailureObserved: boolean,
  providerError?: InferenceErrorLike,
): string {
  // A Codex refresh blocked on the inter-process lock arrives raw (pre-send,
  // no SELECTED wrapper) or as the cause of a SELECTED_PROVIDER_FAILURE (the
  // first-inference refresh at exec start). Either way it keeps its own lock
  // message: mapping it to the generic re-login hint would send the operator
  // into a futile loop that never removes the lock file.
  const lockFailure = codexRefreshLockFailure(err);
  if (lockFailure !== null) return lockFailure.message;
  if (err instanceof Error && err.name === SELECTED_PROVIDER_FAILURE) {
    return CREDENTIAL_FAILURE_USER_MESSAGE;
  }
  // A pre-send Codex refresh throws the raw auth error (no SELECTED wrapper):
  // a failed refresh is still a credential failure with a re-login hint.
  if (codexAuthFailureDiagnostic(err) !== null) {
    return CREDENTIAL_FAILURE_USER_MESSAGE;
  }
  if (providerError === undefined && isResolvedProviderFailureError(err))
    return err.message;
  if (providerFailureObserved || isResolvedProviderFailureError(err)) {
    const diagnostic = providerError ?? {
      category: "fatal",
      message: err instanceof Error ? err.message : String(err),
    };
    return execTerminalProviderFailureMessage(config, diagnostic);
  }
  return formatCaughtError(err);
}

/**
 * Exec-primary director overlay. Omit / skywalker keep the product default
 * (`loadSessionChatPrompt` + advertised session tools). Any other closed-fleet
 * id uses the package prompt and allowlist. Worker effort/nudge are not applied.
 */
export interface ExecDirectorOverlay {
  /** Package system prompt; omitted on the skywalker default path. */
  systemPrompt?: string;
  /** `pkg.tools.allow` (fleet tools stripped when `maySpawn` is false). */
  advertisedAllow?: readonly string[];
  mountFleet: boolean;
}

export function resolveExecDirectorOverlay(
  director: DirectorId | undefined,
): ExecDirectorOverlay {
  if (director === undefined || director === "skywalker") {
    return { mountFleet: true };
  }
  return resolveExecDirectorOverlayForPackage(DIRECTOR_REGISTRY[director]);
}

/**
 * Single enforcement point for the exec allowlist. Everything the overlay
 * permits — tool_search results, promoter activation, onToolsActivate, the
 * call gate — flows through here, so a tool outside the allow can never
 * become callable.
 */
export function isExecOverlayToolAllowed(
  overlay: ExecDirectorOverlay,
  name: string,
): boolean {
  return (
    overlay.advertisedAllow === undefined ||
    overlay.advertisedAllow.includes(name)
  );
}

export function resolveExecDirectorOverlayForPackage(
  pkg: DirectorPackage,
): ExecDirectorOverlay {
  const allow = pkg.tools?.allow;
  const deny = pkg.tools?.deny ?? [];
  if ((allow === undefined || allow.length === 0) && deny.length > 0) {
    throw new Error(
      `Director package "${pkg.id}" sets tools.deny without tools.allow — ` +
        "exec overlays enforce a closed allow list, so a deny-only package " +
        "has no list to subtract from. Add tools.allow.",
    );
  }
  const allowed =
    allow !== undefined && allow.length > 0
      ? allow.filter((name) => !deny.includes(name))
      : undefined;
  if (allowed !== undefined && allowed.length === 0) {
    throw new Error(
      `Director package "${pkg.id}" tools.allow minus tools.deny is empty — ` +
        "exec overlays enforce a closed allow list, so no tool would be " +
        "advertised. Keep an allow entry outside tools.deny.",
    );
  }
  const advertisedAllow =
    allowed !== undefined
      ? pkg.spawn.maySpawn
        ? [
            ...allowed,
            // Exec mounts wait_agents beside the fleet verbs (mountWaitAgents),
            // so it stays advertised here even though the package allow omits
            // it for TUI/nested mailbox-mail collection.
            ...(!allowed.includes("wait_agents") ? ["wait_agents"] : []),
          ]
        : allowed.filter(
            (name) =>
              ![
                "search_agents",
                "spawn_agent",
                "wait_agents",
                "list_agents",
                "close_agent",
                "resume_agent",
                "interrupt_agent",
                "send_input",
                "read_agent_trace",
              ].includes(name),
          )
      : undefined;
  return {
    systemPrompt: formatDirectorSystemPrompt(pkg),
    ...(advertisedAllow !== undefined ? { advertisedAllow } : {}),
    mountFleet: pkg.spawn.maySpawn,
  };
}

/**
 * Build the inbound message for exec's one genuine operator input: the
 * initial task supplied on the command line. Carries
 * OPERATOR_ORIGINATED_FLAG so director.ts's loop-protection backstop can
 * tell this apart from system-originated sends.
 */
function operatorTaskMessage(task: string): InboundMessage {
  return {
    ref: { uid: 1, mailbox: "INBOX" },
    headers: {
      from: "user@local",
      to: ["agent@local"],
      date: new Date().toISOString(),
      messageId: `<${crypto.randomUUID()}@local>`,
      interchangeType: "conversation.message",
    },
    flags: [OPERATOR_ORIGINATED_FLAG],
    content: task,
    signatureStatus: "missing",
  };
}

export interface ExecResult {
  exitCode: number;
  sessionId: string;
  text: string;
  error?: string;
  /** Run status after send/close (done | failed | cancelled). */
  status?: "done" | "failed" | "cancelled";
  /** Wall time from runExec start to summary (ms). */
  durationMs?: number;
  turnsUsed?: number;
  toolCallCount?: number;
  tokenUsage?: {
    input: number;
    output: number;
    cacheRead: number;
    cacheWrite: number;
    thinking: number;
  };
  /** Provider/model the config resolved for this run. */
  provider?: string;
  model?: string;
}

export function createExecToolCallGate(
  isAdvertised: (name: string) => boolean,
  options: { isCodex: boolean },
): (name: string) => boolean {
  const unadvertisedCallable = new Set<string>([
    submitOutputDefinition.name,
    ...(options.isCodex
      ? [shellDefinition.name, updatePlanDefinition.name]
      : []),
  ]);
  return (name) => unadvertisedCallable.has(name) || isAdvertised(name);
}

export function createExecToolPromoter(args: {
  activate: (names: readonly string[]) => boolean;
  isAllowed: (name: string) => boolean;
  persist?: () => void;
  commitWire?: () => void;
}): (names: string[]) => void {
  return (names) => {
    if (!args.activate(names.filter((name) => args.isAllowed(name)))) return;
    args.commitWire?.();
    args.persist?.();
  };
}

/**
 * Product non-TUI agent path (`corbits exec "prompt"`).
 *
 * Shares the same ChatDirector, toolset, permission gate, session mode, and
 * sub-agent surface as the TUI — without Ink. Bootstrap consumes the shared
 * session assembly in src/session/assemble-runtime.ts; see
 * docs/ARCHITECTURE.md "Exec Runner" for the intentional deltas.
 *
 * Operator/permission prompts use stdin when a TTY is available; otherwise
 * they deny (fail closed) unless `--dangerously-skip-permissions` / auto
 * grants cover the action.
 */
export async function runExec(config: Config): Promise<ExecResult> {
  const task = config.task.trim();
  if (task.length === 0) {
    stderr.write('Usage: corbits exec "<prompt>"\n');
    return {
      exitCode: 2,
      sessionId: config.sessionId,
      text: "",
      error: "missing prompt",
      status: "failed",
      durationMs: 0,
      turnsUsed: 0,
      toolCallCount: 0,
      tokenUsage: {
        input: 0,
        output: 0,
        cacheRead: 0,
        cacheWrite: 0,
        thinking: 0,
      },
      provider: config.providerName,
      model: config.model,
    };
  }

  const sessionId =
    config.sessionId.length > 0 ? config.sessionId : generateSessionId();
  const startedAt = Date.now();
  const workdir = sessionContextDir(config.cwd, sessionId);
  await initSessionDir(config.cwd, sessionId);
  const prior =
    config.sessionId.length > 0
      ? await loadState(config.cwd, config.sessionId)
      : undefined;
  const priorState = prior?.kind === "ok" ? prior.state : undefined;

  let connectedMcp: ConnectedMcpServer[] = [];
  let agent: Agent | null = null;
  let toolset: AgentToolset | null = null;
  let subAgentSessions: SubAgentSessionStore | null = null;
  let textOut = "";
  let finalized = false;
  let turnsUsed = 0;
  let runSink: RunSink | null = null;
  let providerFailureObserved = false;
  let providerError: InferenceErrorLike | undefined;
  let result: ExecResult | undefined;
  // Assigned once the advertised toolset exists (below); persist reads it live
  // so a snapshot taken before that point still writes, just without the field.
  const activatedToolsRef: { current?: ActivatedToolTracker } = {};
  const activeRunHandle: RunStateHandle = {
    sessionId,
    cwd: config.cwd,
    task,
    startedAt,
    turnsUsed: 0,
    model: `${config.providerName}:${config.model}`,
    ...(priorState?.lastCacheWriteAt !== undefined
      ? { lastCacheWriteAt: priorState.lastCacheWriteAt }
      : {}),
    ...(priorState?.lastCacheWriteAt !== undefined &&
    priorState.model !== undefined
      ? { cacheWriteModel: priorState.model }
      : {}),
  };
  let stopHeartbeat: (() => void) | undefined;

  const persist = async (
    status: "running" | "done" | "failed" | "cancelled",
    extra?: { error?: string },
  ): Promise<void> => {
    if (finalized && status === "running") return;
    if (status !== "running") {
      finalized = true;
      stopHeartbeat?.();
      stopHeartbeat = undefined;
    }
    const model = `${config.providerName}:${config.model}`;
    const nextTurnsUsed = runSink?.getTurnCount() ?? turnsUsed;
    const activatedTools = activatedToolsRef.current?.list() ?? [];
    syncRunStateHandle(activeRunHandle, {
      turnsUsed: nextTurnsUsed,
      task,
      startedAt,
      model,
      activatedTools,
    });
    const snapshot = {
      status,
      turnsUsed: nextTurnsUsed,
      task,
      startedAt,
      model,
      mcpServers: connectedMcp,
      ...(activatedTools.length > 0 ? { activatedTools } : {}),
      ...(activeRunHandle.lastCacheWriteAt !== undefined
        ? { lastCacheWriteAt: activeRunHandle.lastCacheWriteAt }
        : {}),
      ...(status !== "running" ? { finishedAt: Date.now() } : {}),
      ...(extra?.error !== undefined ? { error: extra.error } : {}),
    };
    const write =
      status === "running"
        ? saveState(config.cwd, sessionId, snapshot).then(() => {
            if (!finalized) {
              setActiveRun(activeRunHandle);
            }
          })
        : finalizeRunState(config.cwd, sessionId, snapshot);
    await write.catch((err: unknown) => {
      // Persistence failure must not fail the run, but dropping it silently
      // hides disk/permission problems that leave run.json stale.
      logger.warn(
        "saveState failed for session {sessionId} status={status}: {error}",
        {
          sessionId,
          status,
          error: formatCaughtError(err),
        },
      );
    });
  };

  await persist("running");
  stopHeartbeat = startRunHeartbeat({
    shouldTick: () => !finalized,
    tick: () => persist("running"),
  });

  try {
    // Pricing seed is optional for exec; continue without rates rather than fail the run.
    const inferenceDeps = await assembleInferenceBase((err: unknown) => {
      logger.debug("seedPricingMetadataFromCache failed: {error}", {
        error: formatCaughtError(err),
      });
    });

    // One-shot migration only when the path-trust file does not exist yet.
    // Headless exec has no frame to corrupt, so a skipped marketplace member
    // writes straight to stderr here — an explicit choice at this call site,
    // not `expandPluginPath` falling back to it on its own.
    const sessionTrust = await assembleSessionTrust({
      cwd: config.cwd,
      pluginPaths: config.settings?.pluginPaths,
      discoverClaudePlugins: config.settings?.discoverClaudePlugins,
      onExpandSkip: (skip: ExpandPluginPathSkip) => {
        process.stderr.write(`plugins: ${formatExpandSkip(skip)}\n`);
      },
      telemetry: liveTelemetry,
    });
    const { projectTrust, pluginModules } = sessionTrust;
    const pluginLoadDiag = sessionTrust.diagnostics;
    emitPluginWarningSummary(pluginLoadDiag, (line) => logger.warn(line));
    // Metadata-only (untrusted) modules stay out of executable plugins.
    const executablePlugins = () =>
      pluginModules.filter((m) => m.metadataOnly !== true);
    const pluginConfig = config.settings?.plugins ?? {};

    const toolPluginCandidates = collectToolPlugins(executablePlugins());
    const extraToolPlugins = await resolveToolPlugins({
      candidates: toolPluginCandidates,
      pluginConfig,
    });

    const skillDirs = skillDirsFromEnabledPlugins(
      executablePlugins(),
      pluginConfig,
    );

    const profilesDir = join(config.cwd, ".agents", "agents");
    const liveAgentProfiles = await loadAgentProfiles(profilesDir);

    // loadLocalSettings maps ENOENT → null; a throw is real I/O or schema failure.
    const localSettingsForMode = await loadSessionLocalSettings({
      cwd: config.cwd,
      globalSettingsPath: config.globalSettingsPath,
      onError: (err: unknown) => {
        logger.warn("Failed to load local settings: {error}", {
          error: formatCaughtError(err),
        });
      },
    });
    const sessionMode: SessionMode =
      resolveSessionMode(config.settings, localSettingsForMode) ??
      "orchestrator";

    const interactive = input.isTTY === true && output.isTTY === true;

    if (config.skipPermissionsFromSettings) {
      stderr.write(
        `${savedSkipPermissionsWarning(config.globalSettingsPath, "exec")}\n`,
      );
    }

    const { gate: permissionGate } = await assembleSessionGate({
      cwd: config.cwd,
      sessionId,
      providerName: config.providerName,
      model: config.model,
      telemetry: liveTelemetry,
      requestApproval: (request: PermissionRequest): Promise<ApprovalOutcome> =>
        promptPermission(request, interactive),
      getActiveProviderModel: () => `${config.providerName}:${config.model}`,
      onPersistNotice: (text) => {
        stderr.write(`${text}\n`);
      },
      onPendingProjectGrants: (text) => {
        stderr.write(`${text}\n`);
      },
      interactive,
      skipPermissions: config.dangerouslySkipPermissions,
      auto: config.auto,
      // Main session: gating rides the reactor's approval-suspend seam. In
      // exec (headless) an ask resolves to a deny effect, so the hook blocks
      // — the same policy the middleware path applied.
      reactorGated: true,
    });

    const liveSubAgentProvider: { current: SubAgentProvider } = {
      current: buildSubAgentProvider(config),
    };
    const fleetSessions = createSubAgentSessionStore({
      admission: getProcessAdmissionQueue(),
    });
    subAgentSessions = fleetSessions;
    const shellTimeout = shellTimeoutFromSettings(config.settings);
    const toolWatchdog = toolWatchdogFromSettings(config.settings);
    const toolAvailability: ToolAvailability = {
      languageServerAvailable: detectLanguageServerAvailable(config.cwd),
      operatorAvailable: interactive,
      // Exec primary keeps wait_agents mounted: headless runs have no
      // mailbox-mail flush, so wait_agents stays the collection path here.
      waitAgentsMounted: true,
    };

    let currentAgent: Agent | null = null;
    let currentStorage: ContextStore | null = null;
    const evidenceArchiveHolder: { current?: CompactionArchive } = {};

    const overlay = resolveExecDirectorOverlay(config.director);
    const workflowHostHolder: { instance?: WorkflowHost } = {};
    const subAgentSettings = config.settings;

    const agentToolset = await createAgentToolset({
      cwd: config.cwd,
      permissionGate,
      skillDirs,
      telemetry: liveTelemetry,
      isCodex: isCodexProviderName(config.providerName),
      ...(shellTimeout !== undefined ? { shellTimeout } : {}),
      ...(toolWatchdog !== undefined ? { toolWatchdog } : {}),
      ...(localSettingsForMode?.env !== undefined
        ? { shellEnv: localSettingsForMode.env }
        : {}),
      ...(localSettingsForMode?.pinnedTools !== undefined
        ? { pinnedTools: localSettingsForMode.pinnedTools }
        : {}),
      getBlobWriter: () => currentStorage?.writeBlob,
      getEvidenceArchive: () => evidenceArchiveHolder.current,
      getContextDir: () => workdir,
      // Background run_shell completions re-enter the reactor on a later turn.
      onBackgroundShellExit: (exit) => {
        currentAgent?.deliver(buildShellBackgroundMessage(exit));
      },
      getBlobReader: () => {
        if (currentAgent === null) {
          throw new Error("blob reader requested before agent init");
        }
        return currentAgent.blobReader;
      },
      isWorkflowActive: () => workflowHostHolder.instance?.isActive() === true,
      completeWorkflowStep: (stepId) =>
        workflowHostHolder.instance?.complete(stepId) ?? "not-current",
      ...(interactive
        ? {
            onOperatorGate: (question: string, options: string[]) =>
              promptOperator(question, options, true),
          }
        : {}),
      sessionMode,
      toolAvailability,
      // Closed director overlays unmount tool_search (unless allowed) and
      // filter its index, so search cannot surface outside-allow tools.
      ...(overlay.advertisedAllow !== undefined
        ? { toolSearchAllow: overlay.advertisedAllow }
        : {}),
      // Exec-primary keeps wait_agents mounted (with an advertised allow):
      // headless runs have no mailbox-mail flush, so wait_agents stays the
      // collection path here. TUI primary and nested orchestrators omit it.
      mountWaitAgents: true,
      ...(config.mcpServers !== undefined
        ? { mcpServers: config.mcpServers }
        : {}),
      mcpServersSource: config.mcpServersSource ?? "none",
      projectTrust,
      requestMcpTrust: async (server) => {
        if (!interactive) return false;
        const result = await promptOperator(
          formatExecMcpTrustQuestion(server),
          ["Trust and connect", "Deny"],
          true,
        );
        return result.kind === "option" && result.index === 0;
      },
      ...(overlay.mountFleet
        ? {
            subAgent: {
              provider: () => liveSubAgentProvider.current,
              sessions: fleetSessions,
              getWorkdirBase: () => sessionDir(config.cwd, sessionId),
              onProgress: () => undefined,
              ...(subAgentSettings !== undefined
                ? { settings: () => subAgentSettings }
                : {}),
              catalog: () => config.providers,
              profiles: () => liveAgentProfiles,
            },
          }
        : {}),
      ...(extraToolPlugins.length > 0 ? { extraToolPlugins } : {}),
    });
    toolset = agentToolset;
    setActiveDisposeHost(() =>
      disposeExecRuntime({ agent, toolset, subAgentSessions }),
    );

    const systemPrompt =
      overlay.systemPrompt ??
      (
        await loadSessionChatPrompt({
          cwd: config.cwd,
          skillDirs,
          ...(config.systemPromptExtensions !== undefined
            ? { systemPromptExtensions: config.systemPromptExtensions }
            : {}),
          ...(config.promptSectionOmit !== undefined
            ? { promptSectionOmit: config.promptSectionOmit }
            : {}),
          sessionMode,
          toolAvailability,
          skills: agentToolset.skills,
          providerName: config.providerName,
          model: config.model,
        })
      ).systemPrompt;

    const initialCodexProfile = codexProfileFromProviderName(
      config.providerName,
    );
    const initialXaiProfile = xaiProfileFromProviderName(config.providerName);
    const initialBundle = resolveLiveSessionSources(config, sessionId);
    const liveSources = initialBundle.sources;
    const liveDefaultSource = initialBundle.defaultSource;
    const selectedSource = initialBundle.selected;
    let liveSource: InferenceSource = selectedSource;

    // Refresh OAuth tokens before first inference when starting on codex/xai.
    if (initialCodexProfile !== undefined) {
      const { access } = await refreshSelectedProviderCredential(() =>
        getValidCodexToken(initialCodexProfile),
      );
      registerSourceCredential(liveSource.credentialId, access);
      liveSubAgentProvider.current = {
        ...liveSubAgentProvider.current,
        apiKey: access,
      };
    }
    if (initialXaiProfile !== undefined) {
      const { access } = await refreshSelectedProviderCredential(() =>
        getValidXaiToken(initialXaiProfile),
      );
      registerSourceCredential(liveSource.credentialId, access);
      liveSubAgentProvider.current = {
        ...liveSubAgentProvider.current,
        apiKey: access,
      };
    }

    const summarizeForCompaction = createModelSummarizer({
      getSource: () => liveSource,
      deps: inferenceDeps,
      getArchive: () => evidenceArchiveHolder.current,
      timeoutMs: config.summarizerTimeoutMs,
      telemetry: liveTelemetry,
      // A 401 here usually means the shared OAuth file rotated under another
      // process; re-read it so the retry runs on the fresh token.
      refreshAuth: async () => {
        const before = peekSourceCredentialSecret(liveSource.credentialId);
        const fresh = await ensureFreshInferenceSource(
          liveSource,
          config.providers,
        );
        liveSource = fresh;
        if (peekSourceCredentialSecret(fresh.credentialId) === before) return;
        if (currentAgent !== null)
          setAgentSourceUnlessClosed(currentAgent, fresh);
      },
    });

    const {
      activated: activatedToolNames,
      computeAdvertised,
      isAdvertised,
      flushPromotions,
    } = createAdvertisedToolset({
      sessionMode,
      toolAvailability,
      getProvider: () => config,
      builtInPrefix: overlay.advertisedAllow,
      ...(localSettingsForMode?.pinnedTools !== undefined
        ? { pinnedTools: localSettingsForMode.pinnedTools }
        : {}),
    });
    activatedToolsRef.current = activatedToolNames;
    // Same wire contract as the TUI: a registered tool the model was never
    // shown errors toward tool_search instead of dispatching blind.
    agentToolset.dynamicRunner.setCallGate(
      createExecToolCallGate(isAdvertised, {
        isCodex: isCodexProviderName(config.providerName),
      }),
      // Same promoted-but-unmounted contract as the TUI gate: an activated
      // name missing from the registry errors toward retry (see run() in
      // DynamicToolRunner).
      { isActivated: (name) => activatedToolNames.has(name) },
    );

    const { directorHolder, buildAgent } = assembleChatAgent({
      toolsId: `${ID_PREFIX}/exec-tools`,
      agentId: `${ID_PREFIX}/exec-agent`,
      systemPrompt,
      getDynamicRunner: () => agentToolset.dynamicRunner,
      computeAdvertised,
      inactivityTimeoutMs: config.inactivityTimeoutMs ?? 750_000,
      totalTimeoutMs: config.totalTimeoutMs,
      clearDenials: () => permissionGate.clearDenials(),
      getProvider: () => config,
      getWorkdir: () => workdir,
      getSessionId: () => sessionId,
      authorize: createReactorAuthorize(permissionGate),
      inferenceDeps,
      getSources: () => {
        const sources = liveSources.length > 0 ? liveSources : [liveSource];
        // OAuth refreshes land in the shared credential cell (keyed by source
        // id), so every source already resolves the live secret — no per-send
        // credential copy is needed.
        return sources;
      },
      getDefaultSource: () =>
        liveDefaultSource.length > 0 ? liveDefaultSource : liveSource.id,
      anthropicCachePrompt: () => config.anthropicCachePrompt,
      getCompactor: () =>
        createSessionPruningCompactor({
          summarize: summarizeForCompaction,
          summaryContext: () => {
            const tools = activatedToolNames.list();
            return tools.length > 0 ? { activatedTools: tools } : undefined;
          },
          readPriorHandoff: () =>
            tryReadPriorHandoffFile(
              currentStorage?.readBlob.bind(currentStorage),
            ),
          telemetry: liveTelemetry,
          onFolded: () => {
            // Fold restarts the cached prefix, so catch promotions still
            // pending. Search already flushed names onto the next infer.
            if (flushPromotions()) {
              directorHolder.instance?.updateToolDefinitions(
                computeAdvertised(
                  agentToolset.dynamicRunner.currentDefinitions(),
                ),
              );
            }
          },
        }),
      getCacheWriteSeed: () =>
        resumeCacheWriteSeed({
          at: activeRunHandle.lastCacheWriteAt,
          storedModel: activeRunHandle.cacheWriteModel,
          liveProvider: config.providerName,
          liveProtocol: liveSource.provider,
        }),
      onBuilt: (agent, storage) => {
        currentAgent = agent;
        currentStorage = storage;
      },
      evidenceArchiveHolder,
    });

    // tool_search starts as a no-op promoter; without this, the call gate
    // refuses MCP/present/plugin names the result just told the model to
    // invoke. Under a closed overlay the promoter only activates allowed
    // names, so outside-allow tools can never become advertised or callable.
    const commitPromotedWire = (): void => {
      if (flushPromotions()) {
        directorHolder.instance?.updateToolDefinitions(
          computeAdvertised(agentToolset.dynamicRunner.currentDefinitions()),
        );
      }
    };
    const promoteAndCommitWire = createExecToolPromoter({
      activate: (names) => activatedToolNames.activate(names),
      isAllowed: (name) => isExecOverlayToolAllowed(overlay, name),
      persist: () => {
        void persist("running");
      },
      commitWire: commitPromotedWire,
    });
    agentToolset.setToolPromoter(promoteAndCommitWire);

    const workflowHost = new WorkflowHost({
      cwd: config.cwd,
      getSessionId: () => sessionId,
      getToolDefinitions: () => agentToolset.dynamicRunner.currentDefinitions(),
      getDirector: () => directorHolder.instance,
    });
    workflowHostHolder.instance = workflowHost;

    const emitter = new EventEmitter();
    const {
      hookManager,
      runSink: liveSink,
      cycleRecorder,
    } = await assembleSessionLifecycle({
      cwd: config.cwd,
      emitter,
      getTelemetry: () => liveTelemetry,
      getSessionId: () => sessionId,
      getSource: () => liveSource,
      onTurnBoundarySnapshot: (event) => {
        if (event?.type === "inference.done") {
          const at = anthropicCacheWriteAt(event.data.source, Date.now());
          if (at !== undefined) {
            activeRunHandle.lastCacheWriteAt = at;
            activeRunHandle.cacheWriteModel = `${event.data.source.provider}:${event.data.source.model}`;
          }
        }
        void persist("running");
      },
      resolveContextDir: () => workdir,
    });
    runSink = liveSink;

    currentAgent = await buildAgent();
    agent = currentAgent;
    // Local non-null handle after init (currentAgent stays mutable for compaction deliver).
    const activeAgent = currentAgent;
    const activeStorage = currentStorage;
    if (activeStorage === null)
      throw new Error("approval resume: no context store");

    if (agentToolset.connectMCP !== undefined) {
      const handshake = armExecMcpHandshakeAbort(EXEC_MCP_HANDSHAKE_TIMEOUT_MS);
      const connecting = followExecMcpHandshake(
        agentToolset.connectMCP(
          {
            interactiveAuth: false,
            onStatus: (status) => {
              if (status.state === "connected") {
                connectedMcp = [
                  ...connectedMcp.filter((s) => s.name !== status.name),
                  { name: status.name, toolCount: status.tools.length },
                ];
              }
            },
            onToolsChanged: (definitions) =>
              directorHolder.instance?.updateToolDefinitions(
                computeAdvertised(definitions),
              ),
          },
          handshake.signal,
        ),
        handshake,
      ).catch((err: unknown) => {
        logger.warn("MCP connect failed: {error}", {
          error: formatCaughtError(err),
        });
      });
      await awaitExecMcpThenResume(connecting, () => workflowHost.resume(), {
        waitMs: EXEC_MCP_CONNECT_WAIT_MS,
        abort: handshake.signal,
        onWaitTimeout: () => {
          logger.warn(
            "MCP handshake still in progress after {waitMs}ms; waiting for settle or abort before resume",
            { waitMs: EXEC_MCP_CONNECT_WAIT_MS },
          );
        },
      });
    } else {
      await workflowHost.resume();
    }

    const textChunks: string[] = [];
    // Consume-once gate for the compaction continuation emit: a replayed
    // duplicate of an already-answered emission must not re-deliver.
    const continuationGate = createContinuationGate();
    // Approval-delivery parity with the TUI: the resume routes its deliver
    // through the shared acceptance-tracked deliverer, and the stream sink
    // below feeds acceptances so a delivered decision settles instead of
    // timing out.
    const approvalAcceptance = createCorrelationAcceptance();
    const approvalDeliverer = createApprovalDeliverer({
      deliverToAgent: (message) => activeAgent.deliver(message),
      acceptance: approvalAcceptance,
    });
    // Cycles persist to the context store only on inference.done; the recorder
    // keeps the in-flight cycle's text so an errored or aborted turn leaves
    // its partial output in partial.jsonl instead of vanishing.
    const sink = (event: ReactorEmittedEvent): void => {
      approvalAcceptance.observe(event);
      // Chat-director reactor events (replacing the former onTasksChange /
      // onActivateTools closures). Exec mode has no live task panel or task
      // stdout output today (unlike the TUI's chrome zone) — debug logging
      // is the closest match to how this mode already surfaces other
      // in-session state changes.
      handleChatDirectorEvent(
        event,
        {
          onTasksChanged: (tasks) => {
            logger.debug("tasks updated: {tasks}", {
              tasks: tasks.map((t) => `${t.status}:${t.title}`).join(", "),
            });
          },
          onToolsActivate: (names) => promoteAndCommitWire(names),
        },
        (message, fields) => logger.debug(message, fields),
      );
      if (event.type === "inference.start" || event.type === "inference.done") {
        providerFailureObserved = false;
        providerError = undefined;
      } else if (event.type === "inference.error") {
        providerFailureObserved = true;
        const error = event.data.error;
        providerError = {
          category: error.category,
          ...(error.message !== undefined ? { message: error.message } : {}),
          ...(error.statusCode !== undefined
            ? { statusCode: error.statusCode }
            : {}),
          ...("providerId" in error && typeof error.providerId === "string"
            ? { providerId: error.providerId }
            : {}),
        };
      } else if (event.type === COMPACTION_CONTINUATION_EVENT) {
        // Compaction governor self-delivers after compact so the loop re-enters.
        // Each emission is answered once: a replayed duplicate of an
        // already-answered emission is ignored instead of re-delivered.
        if (continuationGate.shouldDeliver(event.seq)) {
          currentAgent?.deliver(buildCompactionContinuationMessage());
        }
      }
      liveSink.sink(event);
      cycleRecorder.handleEvent(event);
      if (event.type === "inference.text.delta") {
        const token = (event.data as { token?: string }).token;
        if (typeof token === "string" && token.length > 0) {
          textChunks.push(token);
          output.write(token);
        }
      }
    };

    const streamPromise = consumeStream(activeAgent.stream(), sink);

    // send() resolves on connector.reply when the reactor finishes the cycle.
    // Chat sessions never emit reactor.done until close, so runSink status after
    // an intentional post-send close is "cancelled" even on success. Treat a
    // completed send as success unless the sink recorded a real run error.
    // Snapshot sink state BEFORE close: close emits reactor.done which clears
    // sticky inference.error and would hide a real failure.
    let sendCompleted = false;
    let runError: string | undefined;
    let sinkStatus: ReturnType<typeof liveSink.getStatus> = "cancelled";
    try {
      // Final OAuth refresh immediately before send (token may have aged during MCP).
      if (initialCodexProfile !== undefined) {
        const { access } = await getValidCodexToken(initialCodexProfile);
        if (access !== peekSourceCredentialSecret(liveSource.credentialId)) {
          registerSourceCredential(liveSource.credentialId, access);
          setAgentSourceUnlessClosed(activeAgent, liveSource);
        }
      }
      if (initialXaiProfile !== undefined) {
        const { access } = await getValidXaiToken(initialXaiProfile);
        if (access !== peekSourceCredentialSecret(liveSource.credentialId)) {
          registerSourceCredential(liveSource.credentialId, access);
          setAgentSourceUnlessClosed(activeAgent, liveSource);
        }
      }

      // Stream stays open for multi-turn chat until close() — close first, then
      // drain, or streamPromise never settles.
      const sendResult = await activeAgent.send(operatorTaskMessage(task));
      // A suspension must not park silently in exec: the approval resume owns
      // the terminal prompt flow and delivers the decision to the reactor.
      await createApprovalResume({
        getAgent: () => activeAgent,
        resolveParkedCallId: (correlationId) =>
          resolveParkedCallIdFromStore(activeStorage, correlationId),
        gate: permissionGate,
        deliver: async (message, stillCurrent) => {
          if (!stillCurrent()) return;
          await approvalDeliverer.deliver(message);
        },
      }).handle(sendResult);
      sendCompleted = true;
      runError = runSink.getRunError();
      sinkStatus = runSink.getStatus();
    } finally {
      if (sendCompleted && runSink.getRunError() === undefined) {
        // Successful send: the final cycle's inference.done is queued on the
        // stream but may not have reached the sink yet (send resolves on the
        // connector reply). Drain BEFORE disposing so the done event resets
        // the buffer — dispose snapshots at entry and would otherwise write
        // the entire successful reply as a spurious partial. After the drain,
        // dispose is a no-op on success and a real record only if a cycle
        // died without a terminal event.
        await activeAgent.close().catch((err: unknown) => {
          logger.debug(
            "agent.close during successful-send teardown failed: {error}",
            {
              error: formatCaughtError(err),
            },
          );
        });
        await streamPromise.catch((err: unknown) => {
          logger.debug(
            "stream drain during successful-send teardown failed: {error}",
            {
              error: formatCaughtError(err),
            },
          );
        });
        await cycleRecorder.dispose("cancelled");
      } else {
        // Failed or aborted send: close() tears down stream consumers before
        // the dead cycle's inference.error is delivered, so dispose (which
        // snapshots the buffer at entry) runs before closing or the text is
        // lost.
        await cycleRecorder.dispose(
          sendCompleted ? "cancelled" : "send-failed",
        );
        await activeAgent.close().catch((err: unknown) => {
          logger.debug(
            "agent.close during failed-send teardown failed: {error}",
            {
              error: formatCaughtError(err),
            },
          );
        });
        await streamPromise.catch((err: unknown) => {
          logger.debug(
            "stream drain during failed-send teardown failed: {error}",
            {
              error: formatCaughtError(err),
            },
          );
        });
      }
    }

    textOut = textChunks.join("");
    if (textOut.length > 0 && !textOut.endsWith("\n")) output.write("\n");

    // Counts always live on runSink; getTurnCollector() is null when no
    // lifecycle hooks are configured (typical exec/eval path) and only
    // exposes retained turn history for post-run hooks.
    const turnCollector = runSink.getTurnCollector();
    turnsUsed = runSink.getTurnCount();
    const finishedAt = Date.now();
    const summaryStatus = resolveExecRunStatus({
      sendCompleted,
      sinkStatus,
      runError,
    });

    const runSummary = createRunSummary({
      task,
      status: summaryStatus,
      startedAt,
      finishedAt,
      turnsUsed: runSink.getTurnCount(),
      tokenUsage: runSink.getTokenUsage(),
      turns: turnCollector?.getTurns() ?? [],
      toolCallCount: runSink.getToolCallCount(),
      ...(runError !== undefined ? { error: runError } : {}),
    });
    await hookManager.dispatchPostRun(runSummary).catch((err: unknown) => {
      // Post-run hooks are best-effort; keep the exec exit path intact but
      // surface the failure so operators can see hook/script problems.
      const message = formatCaughtError(err);
      logger.warn("dispatchPostRun failed: {error}", { error: message });
      stderr.write(`Warning: post-run hook failed: ${message}\n`);
    });

    if (
      !sendCompleted ||
      runError !== undefined ||
      summaryStatus === "failed"
    ) {
      const diagnosticMessage =
        runError ??
        (summaryStatus === "cancelled"
          ? "run cancelled before completion"
          : "run failed");
      const userMessage =
        summaryStatus === "failed"
          ? execTerminalProviderFailureMessage(
              config,
              providerError ?? {
                category: "unknown",
                message: diagnosticMessage,
              },
            )
          : diagnosticMessage;
      stderr.write(`Error: ${userMessage}\n`);
      const persistStatus =
        summaryStatus === "cancelled" ? "cancelled" : "failed";
      await persist(persistStatus, { error: diagnosticMessage });
      result = {
        exitCode: 1,
        sessionId,
        text: textOut,
        error: userMessage,
        status: summaryStatus,
        durationMs: finishedAt - startedAt,
        turnsUsed: runSink.getTurnCount(),
        toolCallCount: runSink.getToolCallCount(),
        tokenUsage: runSink.getTokenUsage(),
        provider: config.providerName,
        model: config.model,
      };
      return result;
    }

    await persist("done");
    result = {
      exitCode: 0,
      sessionId,
      text: textOut,
      status: "done",
      durationMs: finishedAt - startedAt,
      turnsUsed: runSink.getTurnCount(),
      toolCallCount: runSink.getToolCallCount(),
      tokenUsage: runSink.getTokenUsage(),
      provider: config.providerName,
      model: config.model,
    };
    return result;
  } catch (err) {
    const diagnosticMessage = formatCaughtError(err);
    logger.error("exec failed: {error}", { error: diagnosticMessage });
    const userMessage = execUserFailureMessage(
      config,
      err,
      providerFailureObserved,
      providerError,
    );
    stderr.write(`Error: ${userMessage}\n`);
    await persist("failed", { error: diagnosticMessage });
    result = {
      exitCode: 1,
      sessionId,
      text: textOut,
      error: userMessage,
      status: "failed",
      durationMs: Date.now() - startedAt,
      turnsUsed: runSink?.getTurnCount() ?? turnsUsed,
      toolCallCount: 0,
      tokenUsage: {
        input: 0,
        output: 0,
        cacheRead: 0,
        cacheWrite: 0,
        thinking: 0,
      },
      provider: config.providerName,
      model: config.model,
    };
    return result;
  } finally {
    try {
      await disposeExecRuntime({ agent, toolset, subAgentSessions });
    } catch (err: unknown) {
      const message = formatCaughtError(err);
      logger.error("runtime dispose failed: {error}", { error: message });
      stderr.write(`Error: runtime dispose failed: ${message}\n`);
      if (result !== undefined) {
        result.exitCode = 1;
        result.status = "failed";
        result.error = `runtime dispose failed: ${message}`;
      }
    }
    clearActiveDisposeHost();
  }
}

async function promptOperator(
  question: string,
  options: string[],
  interactive: boolean,
): Promise<OperatorResult> {
  if (!interactive) {
    return { kind: "cancel" };
  }
  stderr.write(`\n${question}\n`);
  options.forEach((opt, i) => stderr.write(`  [${i + 1}] ${opt}\n`));
  stderr.write("Enter number, free text, or empty to cancel: ");
  const rl = createInterface({ input, output: stderr });
  try {
    const line = (await rl.question("")).trim();
    if (line.length === 0) return { kind: "cancel" };
    const n = Number(line);
    if (Number.isInteger(n) && n >= 1 && n <= options.length) {
      return { kind: "option", index: n - 1 };
    }
    return { kind: "custom", text: line };
  } finally {
    rl.close();
  }
}

async function promptPermission(
  request: PermissionRequest,
  interactive: boolean,
): Promise<ApprovalOutcome> {
  if (!interactive) return { allow: false };
  // Same freeze-while-deciding contract as the TUI gate: the tool wall-clock
  // budget must not run down while the operator reads the prompt.
  const budget = getToolApprovalBudget();
  if (budget === undefined) {
    // Every exec tool call runs under the watchdog ALS; an absent store means
    // the gate fired outside a tool run or the ALS context was lost.
    logger.warn(
      "permission prompt reached with no tool budget in ALS for {tool}",
      {
        tool: request.tool,
      },
    );
  }
  const pauseToken = budget?.waitForApproval ? budget.pause() : undefined;
  const summary = `${request.tool}: ${request.subject}`;
  stderr.write(`\nPermission required: ${summary}\n`);
  const scopes = request.scopes;
  scopes.forEach((scope, i) => {
    stderr.write(`  [${i + 1}] ${scope.label}\n`);
  });
  stderr.write(`  [${scopes.length + 1}] Deny\n`);
  const rl = createInterface({ input, output: stderr });
  let timer: ReturnType<typeof setTimeout> | undefined;
  try {
    // The reactor bounds an approval suspension (default 1h); the prompt must
    // settle below that so the gate answers instead of the suspension timing
    // out. The deadline reuses the tool-budget approval-pause ceiling.
    const answer = rl.question("Choice: ");
    const expired = new Promise<undefined>((resolve) => {
      timer = setTimeout(() => resolve(undefined), MAX_TOOL_APPROVAL_PAUSE_MS);
    });
    const line = await Promise.race([answer, expired]);
    if (line === undefined) {
      // The deadline won: rl.close() in the finally below does not settle a
      // pending question, so abandon the loser explicitly. It can never
      // resolve after close; the attached settlement keeps it contained (no
      // floating promise) if a future runtime ever settles it.
      void answer.then(
        () => undefined,
        () => undefined,
      );
      return { allow: false, message: APPROVAL_TIMEOUT_RESULT_TEXT };
    }
    const n = Number(line.trim());
    if (!Number.isInteger(n) || n < 1 || n > scopes.length) {
      return { allow: false };
    }
    const chosen = scopes[n - 1];
    if (chosen === undefined) {
      return { allow: false };
    }
    return {
      allow: true,
      ...(chosen.pattern !== null ? { persist: chosen } : {}),
    };
  } finally {
    if (timer !== undefined) clearTimeout(timer);
    rl.close();
    if (pauseToken !== undefined) budget?.resume(pauseToken);
  }
}