openclaw/src/agents/pi-embedded-runner.ts

import fs from "node:fs/promises";
import os from "node:os";
import path from "node:path";
import { fileURLToPath } from "node:url";

import type {
  AgentMessage,
  AgentTool,
  StreamFn,
  ThinkingLevel,
} from "@mariozechner/pi-agent-core";
import type {
  Api,
  AssistantMessage,
  ImageContent,
  Model,
  SimpleStreamOptions,
} from "@mariozechner/pi-ai";
import { streamSimple } from "@mariozechner/pi-ai";
import {
  createAgentSession,
  discoverAuthStorage,
  discoverModels,
  SessionManager,
  SettingsManager,
} from "@mariozechner/pi-coding-agent";
import { resolveHeartbeatPrompt } from "../auto-reply/heartbeat.js";
import { parseReplyDirectives } from "../auto-reply/reply/reply-directives.js";
import type {
  ReasoningLevel,
  ThinkLevel,
  VerboseLevel,
} from "../auto-reply/thinking.js";
import { formatToolAggregate } from "../auto-reply/tool-meta.js";
import { isCacheEnabled, resolveCacheTtlMs } from "../config/cache-utils.js";
import type { ClawdbotConfig } from "../config/config.js";
import { resolveProviderCapabilities } from "../config/provider-capabilities.js";
import { getMachineDisplayName } from "../infra/machine-name.js";
import { registerUnhandledRejectionHandler } from "../infra/unhandled-rejections.js";
import { createSubsystemLogger } from "../logging.js";
import {
  type enqueueCommand,
  enqueueCommandInLane,
} from "../process/command-queue.js";
import { normalizeMessageProvider } from "../utils/message-provider.js";
import { isReasoningTagProvider } from "../utils/provider-utils.js";
import { resolveUserPath } from "../utils.js";
import { resolveClawdbotAgentDir } from "./agent-paths.js";
import { resolveSessionAgentIds } from "./agent-scope.js";
import {
  markAuthProfileFailure,
  markAuthProfileGood,
  markAuthProfileUsed,
} from "./auth-profiles.js";
import type { ExecElevatedDefaults, ExecToolDefaults } from "./bash-tools.js";
import {
  CONTEXT_WINDOW_HARD_MIN_TOKENS,
  CONTEXT_WINDOW_WARN_BELOW_TOKENS,
  evaluateContextWindowGuard,
  resolveContextWindowInfo,
} from "./context-window-guard.js";
import {
  DEFAULT_CONTEXT_TOKENS,
  DEFAULT_MODEL,
  DEFAULT_PROVIDER,
} from "./defaults.js";
import { FailoverError, resolveFailoverStatus } from "./failover-error.js";
import {
  ensureAuthProfileStore,
  getApiKeyForModel,
  resolveAuthProfileOrder,
  resolveModelAuthMode,
} from "./model-auth.js";
import { normalizeModelCompat } from "./model-compat.js";
import { ensureClawdbotModelsJson } from "./models-config.js";
import type { MessagingToolSend } from "./pi-embedded-messaging.js";
import {
  ensurePiCompactionReserveTokens,
  resolveCompactionReserveTokensFloor,
} from "./pi-settings.js";
import { acquireSessionWriteLock } from "./session-write-lock.js";

export type { MessagingToolSend } from "./pi-embedded-messaging.js";

import {
  buildBootstrapContextFiles,
  classifyFailoverReason,
  downgradeGeminiHistory,
  type EmbeddedContextFile,
  ensureSessionHeader,
  formatAssistantErrorText,
  isAuthAssistantError,
  isCloudCodeAssistFormatError,
  isCompactionFailureError,
  isContextOverflowError,
  isFailoverAssistantError,
  isFailoverErrorMessage,
  isGoogleModelApi,
  isRateLimitAssistantError,
  isTimeoutErrorMessage,
  pickFallbackThinkingLevel,
  sanitizeGoogleTurnOrdering,
  sanitizeSessionMessagesImages,
  validateAnthropicTurns,
  validateGeminiTurns,
} from "./pi-embedded-helpers.js";
import {
  type BlockReplyChunking,
  subscribeEmbeddedPiSession,
} from "./pi-embedded-subscribe.js";
import {
  extractAssistantText,
  extractAssistantThinking,
  formatReasoningMessage,
} from "./pi-embedded-utils.js";
import { setContextPruningRuntime } from "./pi-extensions/context-pruning/runtime.js";
import { computeEffectiveSettings } from "./pi-extensions/context-pruning/settings.js";
import { makeToolPrunablePredicate } from "./pi-extensions/context-pruning/tools.js";
import { toToolDefinitions } from "./pi-tool-definition-adapter.js";
import { createClawdbotCodingTools } from "./pi-tools.js";
import { resolveSandboxContext } from "./sandbox.js";
import { guardSessionManager } from "./session-tool-result-guard-wrapper.js";
import { sanitizeToolUseResultPairing } from "./session-transcript-repair.js";
import {
  applySkillEnvOverrides,
  applySkillEnvOverridesFromSnapshot,
  loadWorkspaceSkillEntries,
  resolveSkillsPromptForRun,
  type SkillSnapshot,
} from "./skills.js";
import { buildAgentSystemPrompt } from "./system-prompt.js";
import { buildToolSummaryMap } from "./tool-summaries.js";
import { normalizeUsage, type UsageLike } from "./usage.js";
import {
  filterBootstrapFilesForSession,
  loadWorkspaceBootstrapFiles,
} from "./workspace.js";

// Optional features can be implemented as Pi extensions that run in the same Node process.

/**
 * Resolve provider-specific extraParams from model config.
 * Auto-enables thinking mode for GLM-4.x models unless explicitly disabled.
 *
 * For ZAI GLM-4.x models, we auto-enable thinking via the Z.AI Cloud API format:
 *   thinking: { type: "enabled", clear_thinking: boolean }
 *
 * - GLM-4.7: Preserved thinking (clear_thinking: false) - reasoning kept across turns
 * - GLM-4.5/4.6: Interleaved thinking (clear_thinking: true) - reasoning cleared each turn
 *
 * Users can override via config:
 *   agents.defaults.models["zai/glm-4.7"].params.thinking = { type: "disabled" }
 *
 * Or disable via runtime flag: --thinking off
 *
 * @see https://docs.z.ai/guides/capabilities/thinking-mode
 * @internal Exported for testing only
 */
export function resolveExtraParams(params: {
  cfg: ClawdbotConfig | undefined;
  provider: string;
  modelId: string;
  thinkLevel?: string;
}): Record<string, unknown> | undefined {
  const modelKey = `${params.provider}/${params.modelId}`;
  const modelConfig = params.cfg?.agents?.defaults?.models?.[modelKey];
  let extraParams = modelConfig?.params ? { ...modelConfig.params } : undefined;

  // Auto-enable thinking for ZAI GLM-4.x models when not explicitly configured
  // Skip if user explicitly disabled thinking via --thinking off
  if (params.provider === "zai" && params.thinkLevel !== "off") {
    const modelIdLower = params.modelId.toLowerCase();
    const isGlm4 = modelIdLower.includes("glm-4");

    if (isGlm4) {
      // Check if user has explicitly configured thinking params
      const hasThinkingConfig = extraParams?.thinking !== undefined;

      if (!hasThinkingConfig) {
        // GLM-4.7 supports preserved thinking (reasoning kept across turns)
        // GLM-4.5/4.6 use interleaved thinking (reasoning cleared each turn)
        // Z.AI Cloud API format: thinking: { type: "enabled", clear_thinking: boolean }
        const isGlm47 = modelIdLower.includes("glm-4.7");
        const clearThinking = !isGlm47;

        extraParams = {
          ...extraParams,
          thinking: {
            type: "enabled",
            clear_thinking: clearThinking,
          },
        };

        log.debug(
          `auto-enabled thinking for ${modelKey}: type=enabled, clear_thinking=${clearThinking}`,
        );
      }
    }
  }

  return extraParams;
}

/**
 * Create a wrapped streamFn that injects extra params (like temperature) from config.
 *
 * @internal
 */
function createStreamFnWithExtraParams(
  baseStreamFn: StreamFn | undefined,
  extraParams: Record<string, unknown> | undefined,
): StreamFn | undefined {
  if (!extraParams || Object.keys(extraParams).length === 0) {
    return undefined; // No wrapper needed
  }

  const streamParams: Partial<SimpleStreamOptions> = {};
  if (typeof extraParams.temperature === "number") {
    streamParams.temperature = extraParams.temperature;
  }
  if (typeof extraParams.maxTokens === "number") {
    streamParams.maxTokens = extraParams.maxTokens;
  }

  if (Object.keys(streamParams).length === 0) {
    return undefined;
  }

  log.debug(
    `creating streamFn wrapper with params: ${JSON.stringify(streamParams)}`,
  );

  const underlying = baseStreamFn ?? streamSimple;
  const wrappedStreamFn: StreamFn = (model, context, options) =>
    underlying(model, context, {
      ...streamParams,
      ...options, // Caller options take precedence
    });

  return wrappedStreamFn;
}

/**
 * Apply extra params (like temperature) to an agent's streamFn.
 *
 * @internal Exported for testing
 */
export function applyExtraParamsToAgent(
  agent: { streamFn?: StreamFn },
  cfg: ClawdbotConfig | undefined,
  provider: string,
  modelId: string,
  thinkLevel?: string,
): void {
  const extraParams = resolveExtraParams({
    cfg,
    provider,
    modelId,
    thinkLevel,
  });
  const wrappedStreamFn = createStreamFnWithExtraParams(
    agent.streamFn,
    extraParams,
  );

  if (wrappedStreamFn) {
    log.debug(
      `applying extraParams to agent streamFn for ${provider}/${modelId}`,
    );
    agent.streamFn = wrappedStreamFn;
  }
}

// We configure context pruning per-session via a WeakMap registry keyed by the SessionManager instance.

function resolvePiExtensionPath(id: string): string {
  const self = fileURLToPath(import.meta.url);
  const dir = path.dirname(self);
  // In dev this file is `.ts` (tsx), in production it's `.js`.
  const ext = path.extname(self) === ".ts" ? "ts" : "js";
  return path.join(dir, "pi-extensions", `${id}.${ext}`);
}

function resolveContextWindowTokens(params: {
  cfg: ClawdbotConfig | undefined;
  provider: string;
  modelId: string;
  model: Model<Api> | undefined;
}): number {
  return resolveContextWindowInfo({
    cfg: params.cfg,
    provider: params.provider,
    modelId: params.modelId,
    modelContextWindow: params.model?.contextWindow,
    defaultTokens: DEFAULT_CONTEXT_TOKENS,
  }).tokens;
}

function buildContextPruningExtension(params: {
  cfg: ClawdbotConfig | undefined;
  sessionManager: SessionManager;
  provider: string;
  modelId: string;
  model: Model<Api> | undefined;
}): { additionalExtensionPaths?: string[] } {
  const raw = params.cfg?.agents?.defaults?.contextPruning;
  if (raw?.mode !== "adaptive" && raw?.mode !== "aggressive") return {};

  const settings = computeEffectiveSettings(raw);
  if (!settings) return {};

  setContextPruningRuntime(params.sessionManager, {
    settings,
    contextWindowTokens: resolveContextWindowTokens(params),
    isToolPrunable: makeToolPrunablePredicate(settings.tools),
  });

  return {
    additionalExtensionPaths: [resolvePiExtensionPath("context-pruning")],
  };
}

function buildEmbeddedExtensionPaths(params: {
  cfg: ClawdbotConfig | undefined;
  sessionManager: SessionManager;
  provider: string;
  modelId: string;
  model: Model<Api> | undefined;
}): string[] {
  const paths = [resolvePiExtensionPath("transcript-sanitize")];
  const pruning = buildContextPruningExtension(params);
  if (pruning.additionalExtensionPaths) {
    paths.push(...pruning.additionalExtensionPaths);
  }
  return paths;
}

export type EmbeddedPiAgentMeta = {
  sessionId: string;
  provider: string;
  model: string;
  usage?: {
    input?: number;
    output?: number;
    cacheRead?: number;
    cacheWrite?: number;
    total?: number;
  };
};

export type EmbeddedPiRunMeta = {
  durationMs: number;
  agentMeta?: EmbeddedPiAgentMeta;
  aborted?: boolean;
};

function buildModelAliasLines(cfg?: ClawdbotConfig) {
  const models = cfg?.agents?.defaults?.models ?? {};
  const entries: Array<{ alias: string; model: string }> = [];
  for (const [keyRaw, entryRaw] of Object.entries(models)) {
    const model = String(keyRaw ?? "").trim();
    if (!model) continue;
    const alias = String(
      (entryRaw as { alias?: string } | undefined)?.alias ?? "",
    ).trim();
    if (!alias) continue;
    entries.push({ alias, model });
  }
  return entries
    .sort((a, b) => a.alias.localeCompare(b.alias))
    .map((entry) => `- ${entry.alias}: ${entry.model}`);
}

type ApiKeyInfo = {
  apiKey: string;
  profileId?: string;
  source: string;
};

export type EmbeddedPiRunResult = {
  payloads?: Array<{
    text?: string;
    mediaUrl?: string;
    mediaUrls?: string[];
    replyToId?: string;
    isError?: boolean;
  }>;
  meta: EmbeddedPiRunMeta;
  // True if a messaging tool (telegram, whatsapp, discord, slack, sessions_send)
  // successfully sent a message. Used to suppress agent's confirmation text.
  didSendViaMessagingTool?: boolean;
  // Texts successfully sent via messaging tools during the run.
  messagingToolSentTexts?: string[];
  // Messaging tool targets that successfully sent a message during the run.
  messagingToolSentTargets?: MessagingToolSend[];
};

export type EmbeddedPiCompactResult = {
  ok: boolean;
  compacted: boolean;
  reason?: string;
  result?: {
    summary: string;
    firstKeptEntryId: string;
    tokensBefore: number;
    details?: unknown;
  };
};

type EmbeddedPiQueueHandle = {
  queueMessage: (text: string) => Promise<void>;
  isStreaming: () => boolean;
  isCompacting: () => boolean;
  abort: () => void;
};

const log = createSubsystemLogger("agent/embedded");
const GOOGLE_TURN_ORDERING_CUSTOM_TYPE = "google-turn-ordering-bootstrap";

registerUnhandledRejectionHandler((reason) => {
  const message = describeUnknownError(reason);
  if (!isCompactionFailureError(message)) return false;
  log.error(`Auto-compaction failed (unhandled): ${message}`);
  return true;
});

type CustomEntryLike = { type?: unknown; customType?: unknown };

function hasGoogleTurnOrderingMarker(sessionManager: SessionManager): boolean {
  try {
    return sessionManager
      .getEntries()
      .some(
        (entry) =>
          (entry as CustomEntryLike)?.type === "custom" &&
          (entry as CustomEntryLike)?.customType ===
            GOOGLE_TURN_ORDERING_CUSTOM_TYPE,
      );
  } catch {
    return false;
  }
}

function markGoogleTurnOrderingMarker(sessionManager: SessionManager): void {
  try {
    sessionManager.appendCustomEntry(GOOGLE_TURN_ORDERING_CUSTOM_TYPE, {
      timestamp: Date.now(),
    });
  } catch {
    // ignore marker persistence failures
  }
}

export function applyGoogleTurnOrderingFix(params: {
  messages: AgentMessage[];
  modelApi?: string | null;
  sessionManager: SessionManager;
  sessionId: string;
  warn?: (message: string) => void;
}): { messages: AgentMessage[]; didPrepend: boolean } {
  if (!isGoogleModelApi(params.modelApi)) {
    return { messages: params.messages, didPrepend: false };
  }
  const first = params.messages[0] as
    | { role?: unknown; content?: unknown }
    | undefined;
  if (first?.role !== "assistant") {
    return { messages: params.messages, didPrepend: false };
  }
  const sanitized = sanitizeGoogleTurnOrdering(params.messages);
  const didPrepend = sanitized !== params.messages;
  if (didPrepend && !hasGoogleTurnOrderingMarker(params.sessionManager)) {
    const warn = params.warn ?? ((message: string) => log.warn(message));
    warn(
      `google turn ordering fixup: prepended user bootstrap (sessionId=${params.sessionId})`,
    );
    markGoogleTurnOrderingMarker(params.sessionManager);
  }
  return { messages: sanitized, didPrepend };
}

async function sanitizeSessionHistory(params: {
  messages: AgentMessage[];
  modelApi?: string | null;
  sessionManager: SessionManager;
  sessionId: string;
}): Promise<AgentMessage[]> {
  const sanitizedImages = await sanitizeSessionMessagesImages(
    params.messages,
    "session:history",
    {
      sanitizeToolCallIds: isGoogleModelApi(params.modelApi),
      enforceToolCallLast: params.modelApi === "anthropic-messages",
    },
  );
  const repairedTools = sanitizeToolUseResultPairing(sanitizedImages);

  // Downgrade tool calls missing thought_signature if using Gemini
  const downgraded = isGoogleModelApi(params.modelApi)
    ? downgradeGeminiHistory(repairedTools)
    : repairedTools;

  return applyGoogleTurnOrderingFix({
    messages: downgraded,
    modelApi: params.modelApi,
    sessionManager: params.sessionManager,
    sessionId: params.sessionId,
  }).messages;
}

/**
 * Limits conversation history to the last N user turns (and their associated
 * assistant responses). This reduces token usage for long-running DM sessions.
 *
 * @param messages - The full message history
 * @param limit - Max number of user turns to keep (undefined = no limit)
 * @returns Messages trimmed to the last `limit` user turns
 */
export function limitHistoryTurns(
  messages: AgentMessage[],
  limit: number | undefined,
): AgentMessage[] {
  if (!limit || limit <= 0 || messages.length === 0) return messages;

  // Count user messages from the end, find cutoff point
  let userCount = 0;
  let lastUserIndex = messages.length;

  for (let i = messages.length - 1; i >= 0; i--) {
    if (messages[i].role === "user") {
      userCount++;
      if (userCount > limit) {
        // We exceeded the limit; keep from the last valid user turn onwards
        return messages.slice(lastUserIndex);
      }
      lastUserIndex = i;
    }
  }
  // Fewer than limit user turns, keep all
  return messages;
}

/**
 * Extracts the provider name and user ID from a session key and looks up
 * dmHistoryLimit from the provider config, with per-DM override support.
 *
 * Session key formats:
 * - `telegram:dm:123` → provider = telegram, userId = 123
 * - `agent:main:telegram:dm:123` → provider = telegram, userId = 123
 *
 * Resolution order:
 * 1. Per-DM override: provider.dms[userId].historyLimit
 * 2. Provider default: provider.dmHistoryLimit
 */
export function getDmHistoryLimitFromSessionKey(
  sessionKey: string | undefined,
  config: ClawdbotConfig | undefined,
): number | undefined {
  if (!sessionKey || !config) return undefined;

  const parts = sessionKey.split(":").filter(Boolean);
  // Handle agent-prefixed keys: agent:<agentId>:<provider>:...
  const providerParts =
    parts.length >= 3 && parts[0] === "agent" ? parts.slice(2) : parts;

  const provider = providerParts[0]?.toLowerCase();
  if (!provider) return undefined;

  // Extract userId: format is provider:dm:userId or provider:dm:userId:...
  // The userId may contain colons (e.g., email addresses), so join remaining parts
  const kind = providerParts[1]?.toLowerCase();
  const userId = providerParts.slice(2).join(":");
  if (kind !== "dm") return undefined;

  // Helper to get limit with per-DM override support
  const getLimit = (
    providerConfig:
      | {
          dmHistoryLimit?: number;
          dms?: Record<string, { historyLimit?: number }>;
        }
      | undefined,
  ): number | undefined => {
    if (!providerConfig) return undefined;
    // Check per-DM override first
    if (
      userId &&
      kind === "dm" &&
      providerConfig.dms?.[userId]?.historyLimit !== undefined
    ) {
      return providerConfig.dms[userId].historyLimit;
    }
    // Fall back to provider default
    return providerConfig.dmHistoryLimit;
  };

  // Map provider to config key
  switch (provider) {
    case "telegram":
      return getLimit(config.telegram);
    case "whatsapp":
      return getLimit(config.whatsapp);
    case "discord":
      return getLimit(config.discord);
    case "slack":
      return getLimit(config.slack);
    case "signal":
      return getLimit(config.signal);
    case "imessage":
      return getLimit(config.imessage);
    case "msteams":
      return getLimit(config.msteams);
    default:
      return undefined;
  }
}

const ACTIVE_EMBEDDED_RUNS = new Map<string, EmbeddedPiQueueHandle>();
type EmbeddedRunWaiter = {
  resolve: (ended: boolean) => void;
  timer: NodeJS.Timeout;
};
const EMBEDDED_RUN_WAITERS = new Map<string, Set<EmbeddedRunWaiter>>();

// ============================================================================
// SessionManager Pre-warming Cache
// ============================================================================

type SessionManagerCacheEntry = {
  sessionFile: string;
  loadedAt: number;
};

const SESSION_MANAGER_CACHE = new Map<string, SessionManagerCacheEntry>();
const DEFAULT_SESSION_MANAGER_TTL_MS = 45_000; // 45 seconds

function getSessionManagerTtl(): number {
  return resolveCacheTtlMs({
    envValue: process.env.CLAWDBOT_SESSION_MANAGER_CACHE_TTL_MS,
    defaultTtlMs: DEFAULT_SESSION_MANAGER_TTL_MS,
  });
}

function isSessionManagerCacheEnabled(): boolean {
  return isCacheEnabled(getSessionManagerTtl());
}

function trackSessionManagerAccess(sessionFile: string): void {
  if (!isSessionManagerCacheEnabled()) return;
  const now = Date.now();
  SESSION_MANAGER_CACHE.set(sessionFile, {
    sessionFile,
    loadedAt: now,
  });
}

function isSessionManagerCached(sessionFile: string): boolean {
  if (!isSessionManagerCacheEnabled()) return false;
  const entry = SESSION_MANAGER_CACHE.get(sessionFile);
  if (!entry) return false;
  const now = Date.now();
  const ttl = getSessionManagerTtl();
  return now - entry.loadedAt <= ttl;
}

async function prewarmSessionFile(sessionFile: string): Promise<void> {
  if (!isSessionManagerCacheEnabled()) return;
  if (isSessionManagerCached(sessionFile)) return;

  try {
    // Read a small chunk to encourage OS page cache warmup.
    const handle = await fs.open(sessionFile, "r");
    try {
      const buffer = Buffer.alloc(4096);
      await handle.read(buffer, 0, buffer.length, 0);
    } finally {
      await handle.close();
    }
    trackSessionManagerAccess(sessionFile);
  } catch {
    // File doesn't exist yet, SessionManager will create it
  }
}

const isAbortError = (err: unknown): boolean => {
  if (!err || typeof err !== "object") return false;
  const name = "name" in err ? String(err.name) : "";
  if (name === "AbortError") return true;
  const message =
    "message" in err && typeof err.message === "string"
      ? err.message.toLowerCase()
      : "";
  return message.includes("aborted");
};

type EmbeddedSandboxInfo = {
  enabled: boolean;
  workspaceDir?: string;
  workspaceAccess?: "none" | "ro" | "rw";
  agentWorkspaceMount?: string;
  browserControlUrl?: string;
  browserNoVncUrl?: string;
  hostBrowserAllowed?: boolean;
  allowedControlUrls?: string[];
  allowedControlHosts?: string[];
  allowedControlPorts?: number[];
  elevated?: {
    allowed: boolean;
    defaultLevel: "on" | "off";
  };
};

function resolveSessionLane(key: string) {
  const cleaned = key.trim() || "main";
  return cleaned.startsWith("session:") ? cleaned : `session:${cleaned}`;
}

function resolveGlobalLane(lane?: string) {
  const cleaned = lane?.trim();
  return cleaned ? cleaned : "main";
}

function resolveUserTimezone(configured?: string): string {
  const trimmed = configured?.trim();
  if (trimmed) {
    try {
      new Intl.DateTimeFormat("en-US", { timeZone: trimmed }).format(
        new Date(),
      );
      return trimmed;
    } catch {
      // ignore invalid timezone
    }
  }
  const host = Intl.DateTimeFormat().resolvedOptions().timeZone;
  return host?.trim() || "UTC";
}

function formatUserTime(date: Date, timeZone: string): string | undefined {
  try {
    const parts = new Intl.DateTimeFormat("en-CA", {
      timeZone,
      weekday: "long",
      year: "numeric",
      month: "2-digit",
      day: "2-digit",
      hour: "2-digit",
      minute: "2-digit",
      hourCycle: "h23",
    }).formatToParts(date);
    const map: Record<string, string> = {};
    for (const part of parts) {
      if (part.type !== "literal") map[part.type] = part.value;
    }
    if (
      !map.weekday ||
      !map.year ||
      !map.month ||
      !map.day ||
      !map.hour ||
      !map.minute
    ) {
      return undefined;
    }
    return `${map.weekday} ${map.year}-${map.month}-${map.day} ${map.hour}:${map.minute}`;
  } catch {
    return undefined;
  }
}

function describeUnknownError(error: unknown): string {
  if (error instanceof Error) return error.message;
  if (typeof error === "string") return error;
  try {
    const serialized = JSON.stringify(error);
    return serialized ?? "Unknown error";
  } catch {
    return "Unknown error";
  }
}

export function buildEmbeddedSandboxInfo(
  sandbox?: Awaited<ReturnType<typeof resolveSandboxContext>>,
  execElevated?: ExecElevatedDefaults,
): EmbeddedSandboxInfo | undefined {
  if (!sandbox?.enabled) return undefined;
  const elevatedAllowed = Boolean(
    execElevated?.enabled && execElevated.allowed,
  );
  return {
    enabled: true,
    workspaceDir: sandbox.workspaceDir,
    workspaceAccess: sandbox.workspaceAccess,
    agentWorkspaceMount:
      sandbox.workspaceAccess === "ro" ? "/agent" : undefined,
    browserControlUrl: sandbox.browser?.controlUrl,
    browserNoVncUrl: sandbox.browser?.noVncUrl,
    hostBrowserAllowed: sandbox.browserAllowHostControl,
    allowedControlUrls: sandbox.browserAllowedControlUrls,
    allowedControlHosts: sandbox.browserAllowedControlHosts,
    allowedControlPorts: sandbox.browserAllowedControlPorts,
    ...(elevatedAllowed
      ? {
          elevated: {
            allowed: true,
            defaultLevel: execElevated?.defaultLevel ?? "off",
          },
        }
      : {}),
  };
}

function buildEmbeddedSystemPrompt(params: {
  workspaceDir: string;
  defaultThinkLevel?: ThinkLevel;
  reasoningLevel?: ReasoningLevel;
  extraSystemPrompt?: string;
  ownerNumbers?: string[];
  reasoningTagHint: boolean;
  heartbeatPrompt?: string;
  skillsPrompt?: string;
  runtimeInfo: {
    host: string;
    os: string;
    arch: string;
    node: string;
    model: string;
    provider?: string;
    capabilities?: string[];
  };
  sandboxInfo?: EmbeddedSandboxInfo;
  tools: AgentTool[];
  modelAliasLines: string[];
  userTimezone: string;
  userTime?: string;
  contextFiles?: EmbeddedContextFile[];
}): string {
  return buildAgentSystemPrompt({
    workspaceDir: params.workspaceDir,
    defaultThinkLevel: params.defaultThinkLevel,
    reasoningLevel: params.reasoningLevel,
    extraSystemPrompt: params.extraSystemPrompt,
    ownerNumbers: params.ownerNumbers,
    reasoningTagHint: params.reasoningTagHint,
    heartbeatPrompt: params.heartbeatPrompt,
    skillsPrompt: params.skillsPrompt,
    runtimeInfo: params.runtimeInfo,
    sandboxInfo: params.sandboxInfo,
    toolNames: params.tools.map((tool) => tool.name),
    toolSummaries: buildToolSummaryMap(params.tools),
    modelAliasLines: params.modelAliasLines,
    userTimezone: params.userTimezone,
    userTime: params.userTime,
    contextFiles: params.contextFiles,
  });
}

export function createSystemPromptOverride(
  systemPrompt: string,
): (defaultPrompt: string) => string {
  const trimmed = systemPrompt.trim();
  return () => trimmed;
}

// We always pass tools via `customTools` so our policy filtering, sandbox integration,
// and extended toolset remain consistent across providers.

type AnyAgentTool = AgentTool;

export function splitSdkTools(options: {
  tools: AnyAgentTool[];
  sandboxEnabled: boolean;
}): {
  builtInTools: AnyAgentTool[];
  customTools: ReturnType<typeof toToolDefinitions>;
} {
  // Always pass all tools as customTools so the SDK doesn't "helpfully" swap in
  // its own built-in implementations (we need our tool wrappers + policy).
  const { tools } = options;
  return {
    builtInTools: [],
    customTools: toToolDefinitions(tools),
  };
}

export function queueEmbeddedPiMessage(
  sessionId: string,
  text: string,
): boolean {
  const handle = ACTIVE_EMBEDDED_RUNS.get(sessionId);
  if (!handle) return false;
  if (!handle.isStreaming()) return false;
  if (handle.isCompacting()) return false;
  void handle.queueMessage(text);
  return true;
}

export function abortEmbeddedPiRun(sessionId: string): boolean {
  const handle = ACTIVE_EMBEDDED_RUNS.get(sessionId);
  if (!handle) return false;
  handle.abort();
  return true;
}

export function isEmbeddedPiRunActive(sessionId: string): boolean {
  return ACTIVE_EMBEDDED_RUNS.has(sessionId);
}

export function isEmbeddedPiRunStreaming(sessionId: string): boolean {
  const handle = ACTIVE_EMBEDDED_RUNS.get(sessionId);
  if (!handle) return false;
  return handle.isStreaming();
}

export function waitForEmbeddedPiRunEnd(
  sessionId: string,
  timeoutMs = 15_000,
): Promise<boolean> {
  if (!sessionId || !ACTIVE_EMBEDDED_RUNS.has(sessionId))
    return Promise.resolve(true);
  return new Promise((resolve) => {
    const waiters = EMBEDDED_RUN_WAITERS.get(sessionId) ?? new Set();
    const waiter: EmbeddedRunWaiter = {
      resolve,
      timer: setTimeout(
        () => {
          waiters.delete(waiter);
          if (waiters.size === 0) EMBEDDED_RUN_WAITERS.delete(sessionId);
          resolve(false);
        },
        Math.max(100, timeoutMs),
      ),
    };
    waiters.add(waiter);
    EMBEDDED_RUN_WAITERS.set(sessionId, waiters);
    if (!ACTIVE_EMBEDDED_RUNS.has(sessionId)) {
      waiters.delete(waiter);
      if (waiters.size === 0) EMBEDDED_RUN_WAITERS.delete(sessionId);
      clearTimeout(waiter.timer);
      resolve(true);
    }
  });
}

function notifyEmbeddedRunEnded(sessionId: string) {
  const waiters = EMBEDDED_RUN_WAITERS.get(sessionId);
  if (!waiters || waiters.size === 0) return;
  EMBEDDED_RUN_WAITERS.delete(sessionId);
  for (const waiter of waiters) {
    clearTimeout(waiter.timer);
    waiter.resolve(true);
  }
}

export function resolveEmbeddedSessionLane(key: string) {
  return resolveSessionLane(key);
}

function mapThinkingLevel(level?: ThinkLevel): ThinkingLevel {
  // pi-agent-core supports "xhigh" too; Clawdbot doesn't surface it for now.
  if (!level) return "off";
  return level;
}

function resolveExecToolDefaults(
  config?: ClawdbotConfig,
): ExecToolDefaults | undefined {
  const tools = config?.tools;
  if (!tools) return undefined;
  if (!tools.exec) return tools.bash;
  if (!tools.bash) return tools.exec;
  return { ...tools.bash, ...tools.exec };
}

function resolveModel(
  provider: string,
  modelId: string,
  agentDir?: string,
  cfg?: ClawdbotConfig,
): {
  model?: Model<Api>;
  error?: string;
  authStorage: ReturnType<typeof discoverAuthStorage>;
  modelRegistry: ReturnType<typeof discoverModels>;
} {
  const resolvedAgentDir = agentDir ?? resolveClawdbotAgentDir();
  const authStorage = discoverAuthStorage(resolvedAgentDir);
  const modelRegistry = discoverModels(authStorage, resolvedAgentDir);
  const model = modelRegistry.find(provider, modelId) as Model<Api> | null;
  if (!model) {
    const providers = cfg?.models?.providers ?? {};
    const inlineModels =
      providers[provider]?.models ??
      Object.values(providers)
        .flatMap((entry) => entry?.models ?? [])
        .map((entry) => ({ ...entry, provider }));
    const inlineMatch = inlineModels.find((entry) => entry.id === modelId);
    if (inlineMatch) {
      const normalized = normalizeModelCompat(inlineMatch as Model<Api>);
      return {
        model: normalized,
        authStorage,
        modelRegistry,
      };
    }
    const providerCfg = providers[provider];
    if (providerCfg || modelId.startsWith("mock-")) {
      const fallbackModel: Model<Api> = normalizeModelCompat({
        id: modelId,
        name: modelId,
        api: providerCfg?.api ?? "openai-responses",
        provider,
        reasoning: false,
        input: ["text"],
        cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 },
        contextWindow:
          providerCfg?.models?.[0]?.contextWindow ?? DEFAULT_CONTEXT_TOKENS,
        maxTokens:
          providerCfg?.models?.[0]?.maxTokens ?? DEFAULT_CONTEXT_TOKENS,
      } as Model<Api>);
      return { model: fallbackModel, authStorage, modelRegistry };
    }
    return {
      error: `Unknown model: ${provider}/${modelId}`,
      authStorage,
      modelRegistry,
    };
  }
  return { model: normalizeModelCompat(model), authStorage, modelRegistry };
}

export async function compactEmbeddedPiSession(params: {
  sessionId: string;
  sessionKey?: string;
  messageProvider?: string;
  agentAccountId?: string;
  sessionFile: string;
  workspaceDir: string;
  agentDir?: string;
  config?: ClawdbotConfig;
  skillsSnapshot?: SkillSnapshot;
  provider?: string;
  model?: string;
  thinkLevel?: ThinkLevel;
  reasoningLevel?: ReasoningLevel;
  bashElevated?: ExecElevatedDefaults;
  customInstructions?: string;
  lane?: string;
  enqueue?: typeof enqueueCommand;
  extraSystemPrompt?: string;
  ownerNumbers?: string[];
}): Promise<EmbeddedPiCompactResult> {
  const sessionLane = resolveSessionLane(
    params.sessionKey?.trim() || params.sessionId,
  );
  const globalLane = resolveGlobalLane(params.lane);
  const enqueueGlobal =
    params.enqueue ??
    ((task, opts) => enqueueCommandInLane(globalLane, task, opts));
  return enqueueCommandInLane(sessionLane, () =>
    enqueueGlobal(async () => {
      const resolvedWorkspace = resolveUserPath(params.workspaceDir);
      const prevCwd = process.cwd();

      const provider =
        (params.provider ?? DEFAULT_PROVIDER).trim() || DEFAULT_PROVIDER;
      const modelId = (params.model ?? DEFAULT_MODEL).trim() || DEFAULT_MODEL;
      const agentDir = params.agentDir ?? resolveClawdbotAgentDir();
      await ensureClawdbotModelsJson(params.config, agentDir);
      const { model, error, authStorage, modelRegistry } = resolveModel(
        provider,
        modelId,
        agentDir,
        params.config,
      );
      if (!model) {
        return {
          ok: false,
          compacted: false,
          reason: error ?? `Unknown model: ${provider}/${modelId}`,
        };
      }
      try {
        const apiKeyInfo = await getApiKeyForModel({
          model,
          cfg: params.config,
        });

        if (model.provider === "github-copilot") {
          const { resolveCopilotApiToken } = await import(
            "../providers/github-copilot-token.js"
          );
          const copilotToken = await resolveCopilotApiToken({
            githubToken: apiKeyInfo.apiKey,
          });
          authStorage.setRuntimeApiKey(model.provider, copilotToken.token);
        } else {
          authStorage.setRuntimeApiKey(model.provider, apiKeyInfo.apiKey);
        }
      } catch (err) {
        return {
          ok: false,
          compacted: false,
          reason: describeUnknownError(err),
        };
      }

      await fs.mkdir(resolvedWorkspace, { recursive: true });
      const sandboxSessionKey = params.sessionKey?.trim() || params.sessionId;
      const sandbox = await resolveSandboxContext({
        config: params.config,
        sessionKey: sandboxSessionKey,
        workspaceDir: resolvedWorkspace,
      });
      const effectiveWorkspace = sandbox?.enabled
        ? sandbox.workspaceAccess === "rw"
          ? resolvedWorkspace
          : sandbox.workspaceDir
        : resolvedWorkspace;
      await fs.mkdir(effectiveWorkspace, { recursive: true });
      await ensureSessionHeader({
        sessionFile: params.sessionFile,
        sessionId: params.sessionId,
        cwd: effectiveWorkspace,
      });

      let restoreSkillEnv: (() => void) | undefined;
      process.chdir(effectiveWorkspace);
      try {
        const shouldLoadSkillEntries =
          !params.skillsSnapshot || !params.skillsSnapshot.resolvedSkills;
        const skillEntries = shouldLoadSkillEntries
          ? loadWorkspaceSkillEntries(effectiveWorkspace)
          : [];
        restoreSkillEnv = params.skillsSnapshot
          ? applySkillEnvOverridesFromSnapshot({
              snapshot: params.skillsSnapshot,
              config: params.config,
            })
          : applySkillEnvOverrides({
              skills: skillEntries ?? [],
              config: params.config,
            });
        const skillsPrompt = resolveSkillsPromptForRun({
          skillsSnapshot: params.skillsSnapshot,
          entries: shouldLoadSkillEntries ? skillEntries : undefined,
          config: params.config,
          workspaceDir: effectiveWorkspace,
        });

        const bootstrapFiles = filterBootstrapFilesForSession(
          await loadWorkspaceBootstrapFiles(effectiveWorkspace),
          params.sessionKey ?? params.sessionId,
        );
        const contextFiles = buildBootstrapContextFiles(bootstrapFiles);
        const runAbortController = new AbortController();
        const tools = createClawdbotCodingTools({
          exec: {
            ...resolveExecToolDefaults(params.config),
            elevated: params.bashElevated,
          },
          sandbox,
          messageProvider: params.messageProvider,
          agentAccountId: params.agentAccountId,
          sessionKey: params.sessionKey ?? params.sessionId,
          agentDir,
          workspaceDir: effectiveWorkspace,
          config: params.config,
          abortSignal: runAbortController.signal,
          modelProvider: model.provider,
          modelId,
          modelAuthMode: resolveModelAuthMode(model.provider, params.config),
          // No currentChannelId/currentThreadTs for compaction - not in message context
        });
        const machineName = await getMachineDisplayName();
        const runtimeProvider = normalizeMessageProvider(
          params.messageProvider,
        );
        const runtimeCapabilities = runtimeProvider
          ? (resolveProviderCapabilities({
              cfg: params.config,
              provider: runtimeProvider,
              accountId: params.agentAccountId,
            }) ?? [])
          : undefined;
        const runtimeInfo = {
          host: machineName,
          os: `${os.type()} ${os.release()}`,
          arch: os.arch(),
          node: process.version,
          model: `${provider}/${modelId}`,
          provider: runtimeProvider,
          capabilities: runtimeCapabilities,
        };
        const sandboxInfo = buildEmbeddedSandboxInfo(
          sandbox,
          params.bashElevated,
        );
        const reasoningTagHint = isReasoningTagProvider(provider);
        const userTimezone = resolveUserTimezone(
          params.config?.agents?.defaults?.userTimezone,
        );
        const userTime = formatUserTime(new Date(), userTimezone);
        // Only include heartbeat prompt for the default agent
        const { defaultAgentId, sessionAgentId } = resolveSessionAgentIds({
          sessionKey: params.sessionKey,
          config: params.config,
        });
        const isDefaultAgent = sessionAgentId === defaultAgentId;
        const appendPrompt = buildEmbeddedSystemPrompt({
          workspaceDir: effectiveWorkspace,
          defaultThinkLevel: params.thinkLevel,
          reasoningLevel: params.reasoningLevel ?? "off",
          extraSystemPrompt: params.extraSystemPrompt,
          ownerNumbers: params.ownerNumbers,
          reasoningTagHint,
          heartbeatPrompt: isDefaultAgent
            ? resolveHeartbeatPrompt(
                params.config?.agents?.defaults?.heartbeat?.prompt,
              )
            : undefined,
          skillsPrompt,
          runtimeInfo,
          sandboxInfo,
          tools,
          modelAliasLines: buildModelAliasLines(params.config),
          userTimezone,
          userTime,
          contextFiles,
        });
        const systemPrompt = createSystemPromptOverride(appendPrompt);

        const sessionLock = await acquireSessionWriteLock({
          sessionFile: params.sessionFile,
        });
        try {
          // Pre-warm session file to bring it into OS page cache
          await prewarmSessionFile(params.sessionFile);
          const sessionManager = guardSessionManager(
            SessionManager.open(params.sessionFile),
          );
          trackSessionManagerAccess(params.sessionFile);
          const settingsManager = SettingsManager.create(
            effectiveWorkspace,
            agentDir,
          );
          ensurePiCompactionReserveTokens({
            settingsManager,
            minReserveTokens: resolveCompactionReserveTokensFloor(
              params.config,
            ),
          });
          const additionalExtensionPaths = buildEmbeddedExtensionPaths({
            cfg: params.config,
            sessionManager,
            provider,
            modelId,
            model,
          });

          const { builtInTools, customTools } = splitSdkTools({
            tools,
            sandboxEnabled: !!sandbox?.enabled,
          });

          let session: Awaited<
            ReturnType<typeof createAgentSession>
          >["session"];
          ({ session } = await createAgentSession({
            cwd: resolvedWorkspace,
            agentDir,
            authStorage,
            modelRegistry,
            model,
            thinkingLevel: mapThinkingLevel(params.thinkLevel),
            systemPrompt,
            tools: builtInTools,
            customTools,
            sessionManager,
            settingsManager,
            skills: [],
            contextFiles: [],
            additionalExtensionPaths,
          }));

          // Wire up config-driven model params (e.g., temperature/maxTokens)
          applyExtraParamsToAgent(
            session.agent,
            params.config,
            provider,
            modelId,
            params.thinkLevel,
          );

          try {
            const prior = await sanitizeSessionHistory({
              messages: session.messages,
              modelApi: model.api,
              sessionManager,
              sessionId: params.sessionId,
            });
            // Validate turn ordering for both Gemini (consecutive assistant) and Anthropic (consecutive user)
            const validatedGemini = validateGeminiTurns(prior);
            const validated = validateAnthropicTurns(validatedGemini);
            const limited = limitHistoryTurns(
              validated,
              getDmHistoryLimitFromSessionKey(params.sessionKey, params.config),
            );
            if (limited.length > 0) {
              session.agent.replaceMessages(limited);
            }
            const result = await session.compact(params.customInstructions);
            return {
              ok: true,
              compacted: true,
              result: {
                summary: result.summary,
                firstKeptEntryId: result.firstKeptEntryId,
                tokensBefore: result.tokensBefore,
                details: result.details,
              },
            };
          } finally {
            sessionManager.flushPendingToolResults?.();
            session.dispose();
          }
        } finally {
          await sessionLock.release();
        }
      } catch (err) {
        return {
          ok: false,
          compacted: false,
          reason: describeUnknownError(err),
        };
      } finally {
        restoreSkillEnv?.();
        process.chdir(prevCwd);
      }
    }),
  );
}

export async function runEmbeddedPiAgent(params: {
  sessionId: string;
  sessionKey?: string;
  messageProvider?: string;
  agentAccountId?: string;
  /** Current channel ID for auto-threading (Slack). */
  currentChannelId?: string;
  /** Current thread timestamp for auto-threading (Slack). */
  currentThreadTs?: string;
  /** Reply-to mode for Slack auto-threading. */
  replyToMode?: "off" | "first" | "all";
  /** Mutable ref to track if a reply was sent (for "first" mode). */
  hasRepliedRef?: { value: boolean };
  sessionFile: string;
  workspaceDir: string;
  agentDir?: string;
  config?: ClawdbotConfig;
  skillsSnapshot?: SkillSnapshot;
  prompt: string;
  /** Optional image attachments for multimodal messages. */
  images?: ImageContent[];
  provider?: string;
  model?: string;
  authProfileId?: string;
  thinkLevel?: ThinkLevel;
  verboseLevel?: VerboseLevel;
  reasoningLevel?: ReasoningLevel;
  bashElevated?: ExecElevatedDefaults;
  timeoutMs: number;
  runId: string;
  abortSignal?: AbortSignal;
  shouldEmitToolResult?: () => boolean;
  onPartialReply?: (payload: {
    text?: string;
    mediaUrls?: string[];
  }) => void | Promise<void>;
  onBlockReply?: (payload: {
    text?: string;
    mediaUrls?: string[];
    audioAsVoice?: boolean;
  }) => void | Promise<void>;
  /** Flush pending block replies (e.g., before tool execution to preserve message boundaries). */
  onBlockReplyFlush?: () => void | Promise<void>;
  blockReplyBreak?: "text_end" | "message_end";
  blockReplyChunking?: BlockReplyChunking;
  onReasoningStream?: (payload: {
    text?: string;
    mediaUrls?: string[];
  }) => void | Promise<void>;
  onToolResult?: (payload: {
    text?: string;
    mediaUrls?: string[];
  }) => void | Promise<void>;
  onAgentEvent?: (evt: {
    stream: string;
    data: Record<string, unknown>;
  }) => void;
  lane?: string;
  enqueue?: typeof enqueueCommand;
  extraSystemPrompt?: string;
  ownerNumbers?: string[];
  enforceFinalTag?: boolean;
}): Promise<EmbeddedPiRunResult> {
  const sessionLane = resolveSessionLane(
    params.sessionKey?.trim() || params.sessionId,
  );
  const globalLane = resolveGlobalLane(params.lane);
  const enqueueGlobal =
    params.enqueue ??
    ((task, opts) => enqueueCommandInLane(globalLane, task, opts));
  const runAbortController = new AbortController();
  return enqueueCommandInLane(sessionLane, () =>
    enqueueGlobal(async () => {
      const started = Date.now();
      const resolvedWorkspace = resolveUserPath(params.workspaceDir);
      const prevCwd = process.cwd();

      const provider =
        (params.provider ?? DEFAULT_PROVIDER).trim() || DEFAULT_PROVIDER;
      const modelId = (params.model ?? DEFAULT_MODEL).trim() || DEFAULT_MODEL;
      const agentDir = params.agentDir ?? resolveClawdbotAgentDir();
      await ensureClawdbotModelsJson(params.config, agentDir);
      const { model, error, authStorage, modelRegistry } = resolveModel(
        provider,
        modelId,
        agentDir,
        params.config,
      );
      if (!model) {
        throw new Error(error ?? `Unknown model: ${provider}/${modelId}`);
      }

      const ctxInfo = resolveContextWindowInfo({
        cfg: params.config,
        provider,
        modelId,
        modelContextWindow: model.contextWindow,
        defaultTokens: DEFAULT_CONTEXT_TOKENS,
      });
      const ctxGuard = evaluateContextWindowGuard({
        info: ctxInfo,
        warnBelowTokens: CONTEXT_WINDOW_WARN_BELOW_TOKENS,
        hardMinTokens: CONTEXT_WINDOW_HARD_MIN_TOKENS,
      });
      if (ctxGuard.shouldWarn) {
        log.warn(
          `low context window: ${provider}/${modelId} ctx=${ctxGuard.tokens} (warn<${CONTEXT_WINDOW_WARN_BELOW_TOKENS}) source=${ctxGuard.source}`,
        );
      }
      if (ctxGuard.shouldBlock) {
        log.error(
          `blocked model (context window too small): ${provider}/${modelId} ctx=${ctxGuard.tokens} (min=${CONTEXT_WINDOW_HARD_MIN_TOKENS}) source=${ctxGuard.source}`,
        );
        throw new FailoverError(
          `Model context window too small (${ctxGuard.tokens} tokens). Minimum is ${CONTEXT_WINDOW_HARD_MIN_TOKENS}.`,
          { reason: "unknown", provider, model: modelId },
        );
      }
      const authStore = ensureAuthProfileStore(agentDir);
      const explicitProfileId = params.authProfileId?.trim();
      const profileOrder = resolveAuthProfileOrder({
        cfg: params.config,
        store: authStore,
        provider,
        preferredProfile: explicitProfileId,
      });
      if (explicitProfileId && !profileOrder.includes(explicitProfileId)) {
        throw new Error(
          `Auth profile "${explicitProfileId}" is not configured for ${provider}.`,
        );
      }
      const profileCandidates =
        profileOrder.length > 0 ? profileOrder : [undefined];
      let profileIndex = 0;
      const initialThinkLevel = params.thinkLevel ?? "off";
      let thinkLevel = initialThinkLevel;
      const attemptedThinking = new Set<ThinkLevel>();
      let apiKeyInfo: ApiKeyInfo | null = null;
      let lastProfileId: string | undefined;

      const resolveApiKeyForCandidate = async (candidate?: string) => {
        return getApiKeyForModel({
          model,
          cfg: params.config,
          profileId: candidate,
          store: authStore,
        });
      };

      const applyApiKeyInfo = async (candidate?: string): Promise<void> => {
        apiKeyInfo = await resolveApiKeyForCandidate(candidate);

        if (model.provider === "github-copilot") {
          const { resolveCopilotApiToken } = await import(
            "../providers/github-copilot-token.js"
          );
          const copilotToken = await resolveCopilotApiToken({
            githubToken: apiKeyInfo.apiKey,
          });
          authStorage.setRuntimeApiKey(model.provider, copilotToken.token);
        } else {
          authStorage.setRuntimeApiKey(model.provider, apiKeyInfo.apiKey);
        }

        lastProfileId = apiKeyInfo.profileId;
      };

      const advanceAuthProfile = async (): Promise<boolean> => {
        let nextIndex = profileIndex + 1;
        while (nextIndex < profileCandidates.length) {
          const candidate = profileCandidates[nextIndex];
          try {
            await applyApiKeyInfo(candidate);
            profileIndex = nextIndex;
            thinkLevel = initialThinkLevel;
            attemptedThinking.clear();
            return true;
          } catch (err) {
            if (candidate && candidate === explicitProfileId) throw err;
            nextIndex += 1;
          }
        }
        return false;
      };

      try {
        await applyApiKeyInfo(profileCandidates[profileIndex]);
      } catch (err) {
        if (profileCandidates[profileIndex] === explicitProfileId) throw err;
        const advanced = await advanceAuthProfile();
        if (!advanced) throw err;
      }

      while (true) {
        const thinkingLevel = mapThinkingLevel(thinkLevel);
        attemptedThinking.add(thinkLevel);

        log.debug(
          `embedded run start: runId=${params.runId} sessionId=${params.sessionId} provider=${provider} model=${modelId} thinking=${thinkLevel} messageProvider=${params.messageProvider ?? "unknown"}`,
        );

        await fs.mkdir(resolvedWorkspace, { recursive: true });
        const sandboxSessionKey = params.sessionKey?.trim() || params.sessionId;
        const sandbox = await resolveSandboxContext({
          config: params.config,
          sessionKey: sandboxSessionKey,
          workspaceDir: resolvedWorkspace,
        });
        const effectiveWorkspace = sandbox?.enabled
          ? sandbox.workspaceAccess === "rw"
            ? resolvedWorkspace
            : sandbox.workspaceDir
          : resolvedWorkspace;
        await fs.mkdir(effectiveWorkspace, { recursive: true });

        let restoreSkillEnv: (() => void) | undefined;
        process.chdir(effectiveWorkspace);
        try {
          const shouldLoadSkillEntries =
            !params.skillsSnapshot || !params.skillsSnapshot.resolvedSkills;
          const skillEntries = shouldLoadSkillEntries
            ? loadWorkspaceSkillEntries(effectiveWorkspace)
            : [];
          restoreSkillEnv = params.skillsSnapshot
            ? applySkillEnvOverridesFromSnapshot({
                snapshot: params.skillsSnapshot,
                config: params.config,
              })
            : applySkillEnvOverrides({
                skills: skillEntries ?? [],
                config: params.config,
              });
          const skillsPrompt = resolveSkillsPromptForRun({
            skillsSnapshot: params.skillsSnapshot,
            entries: shouldLoadSkillEntries ? skillEntries : undefined,
            config: params.config,
            workspaceDir: effectiveWorkspace,
          });

          const bootstrapFiles = filterBootstrapFilesForSession(
            await loadWorkspaceBootstrapFiles(effectiveWorkspace),
            params.sessionKey ?? params.sessionId,
          );
          const contextFiles = buildBootstrapContextFiles(bootstrapFiles);
          // Tool schemas must be provider-compatible (OpenAI requires top-level `type: "object"`).
          // `createClawdbotCodingTools()` normalizes schemas so the session can pass them through unchanged.
          const tools = createClawdbotCodingTools({
            exec: {
              ...resolveExecToolDefaults(params.config),
              elevated: params.bashElevated,
            },
            sandbox,
            messageProvider: params.messageProvider,
            agentAccountId: params.agentAccountId,
            sessionKey: params.sessionKey ?? params.sessionId,
            agentDir,
            workspaceDir: effectiveWorkspace,
            config: params.config,
            abortSignal: runAbortController.signal,
            modelProvider: model.provider,
            modelId,
            modelAuthMode: resolveModelAuthMode(model.provider, params.config),
            currentChannelId: params.currentChannelId,
            currentThreadTs: params.currentThreadTs,
            replyToMode: params.replyToMode,
            hasRepliedRef: params.hasRepliedRef,
          });
          const machineName = await getMachineDisplayName();
          const runtimeInfo = {
            host: machineName,
            os: `${os.type()} ${os.release()}`,
            arch: os.arch(),
            node: process.version,
            model: `${provider}/${modelId}`,
          };
          const sandboxInfo = buildEmbeddedSandboxInfo(
            sandbox,
            params.bashElevated,
          );
          const reasoningTagHint = isReasoningTagProvider(provider);
          const userTimezone = resolveUserTimezone(
            params.config?.agents?.defaults?.userTimezone,
          );
          const userTime = formatUserTime(new Date(), userTimezone);
          // Only include heartbeat prompt for the default agent
          const { defaultAgentId, sessionAgentId } = resolveSessionAgentIds({
            sessionKey: params.sessionKey,
            config: params.config,
          });
          const isDefaultAgent = sessionAgentId === defaultAgentId;
          const appendPrompt = buildEmbeddedSystemPrompt({
            workspaceDir: effectiveWorkspace,
            defaultThinkLevel: thinkLevel,
            reasoningLevel: params.reasoningLevel ?? "off",
            extraSystemPrompt: params.extraSystemPrompt,
            ownerNumbers: params.ownerNumbers,
            reasoningTagHint,
            heartbeatPrompt: isDefaultAgent
              ? resolveHeartbeatPrompt(
                  params.config?.agents?.defaults?.heartbeat?.prompt,
                )
              : undefined,
            skillsPrompt,
            runtimeInfo,
            sandboxInfo,
            tools,
            modelAliasLines: buildModelAliasLines(params.config),
            userTimezone,
            userTime,
            contextFiles,
          });
          const systemPrompt = createSystemPromptOverride(appendPrompt);

          const sessionLock = await acquireSessionWriteLock({
            sessionFile: params.sessionFile,
          });
          // Pre-warm session file to bring it into OS page cache
          await prewarmSessionFile(params.sessionFile);
          const sessionManager = guardSessionManager(
            SessionManager.open(params.sessionFile),
          );
          trackSessionManagerAccess(params.sessionFile);
          const settingsManager = SettingsManager.create(
            effectiveWorkspace,
            agentDir,
          );
          ensurePiCompactionReserveTokens({
            settingsManager,
            minReserveTokens: resolveCompactionReserveTokensFloor(
              params.config,
            ),
          });
          const additionalExtensionPaths = buildEmbeddedExtensionPaths({
            cfg: params.config,
            sessionManager,
            provider,
            modelId,
            model,
          });

          const { builtInTools, customTools } = splitSdkTools({
            tools,
            sandboxEnabled: !!sandbox?.enabled,
          });

          let session: Awaited<
            ReturnType<typeof createAgentSession>
          >["session"];
          ({ session } = await createAgentSession({
            cwd: resolvedWorkspace,
            agentDir,
            authStorage,
            modelRegistry,
            model,
            thinkingLevel,
            systemPrompt,
            // Built-in tools recognized by pi-coding-agent SDK
            tools: builtInTools,
            // Custom clawdbot tools (browser, canvas, nodes, cron, etc.)
            customTools,
            sessionManager,
            settingsManager,
            skills: [],
            contextFiles: [],
            additionalExtensionPaths,
          }));

          // Wire up config-driven model params (e.g., temperature/maxTokens)
          applyExtraParamsToAgent(
            session.agent,
            params.config,
            provider,
            modelId,
            params.thinkLevel,
          );

          try {
            const prior = await sanitizeSessionHistory({
              messages: session.messages,
              modelApi: model.api,
              sessionManager,
              sessionId: params.sessionId,
            });
            // Validate turn ordering for both Gemini (consecutive assistant) and Anthropic (consecutive user)
            const validatedGemini = validateGeminiTurns(prior);
            const validated = validateAnthropicTurns(validatedGemini);
            const limited = limitHistoryTurns(
              validated,
              getDmHistoryLimitFromSessionKey(params.sessionKey, params.config),
            );
            if (limited.length > 0) {
              session.agent.replaceMessages(limited);
            }
          } catch (err) {
            sessionManager.flushPendingToolResults?.();
            session.dispose();
            await sessionLock.release();
            throw err;
          }
          let aborted = Boolean(params.abortSignal?.aborted);
          let timedOut = false;
          const abortRun = (isTimeout = false) => {
            aborted = true;
            if (isTimeout) timedOut = true;
            runAbortController.abort();
            void session.abort();
          };
          let subscription: ReturnType<typeof subscribeEmbeddedPiSession>;
          try {
            subscription = subscribeEmbeddedPiSession({
              session,
              runId: params.runId,
              verboseLevel: params.verboseLevel,
              reasoningMode: params.reasoningLevel ?? "off",
              shouldEmitToolResult: params.shouldEmitToolResult,
              onToolResult: params.onToolResult,
              onReasoningStream: params.onReasoningStream,
              onBlockReply: params.onBlockReply,
              onBlockReplyFlush: params.onBlockReplyFlush,
              blockReplyBreak: params.blockReplyBreak,
              blockReplyChunking: params.blockReplyChunking,
              onPartialReply: params.onPartialReply,
              onAgentEvent: params.onAgentEvent,
              enforceFinalTag: params.enforceFinalTag,
            });
          } catch (err) {
            sessionManager.flushPendingToolResults?.();
            session.dispose();
            await sessionLock.release();
            throw err;
          }
          const {
            assistantTexts,
            toolMetas,
            unsubscribe,
            waitForCompactionRetry,
            getMessagingToolSentTexts,
            getMessagingToolSentTargets,
            didSendViaMessagingTool,
          } = subscription;

          const queueHandle: EmbeddedPiQueueHandle = {
            queueMessage: async (text: string) => {
              await session.steer(text);
            },
            isStreaming: () => session.isStreaming,
            isCompacting: () => subscription.isCompacting(),
            abort: abortRun,
          };
          ACTIVE_EMBEDDED_RUNS.set(params.sessionId, queueHandle);

          let abortWarnTimer: NodeJS.Timeout | undefined;
          const abortTimer = setTimeout(
            () => {
              log.warn(
                `embedded run timeout: runId=${params.runId} sessionId=${params.sessionId} timeoutMs=${params.timeoutMs}`,
              );
              abortRun(true);
              if (!abortWarnTimer) {
                abortWarnTimer = setTimeout(() => {
                  if (!session.isStreaming) return;
                  log.warn(
                    `embedded run abort still streaming: runId=${params.runId} sessionId=${params.sessionId}`,
                  );
                }, 10_000);
              }
            },
            Math.max(1, params.timeoutMs),
          );

          let messagesSnapshot: AgentMessage[] = [];
          let sessionIdUsed = session.sessionId;
          const onAbort = () => {
            abortRun();
          };
          if (params.abortSignal) {
            if (params.abortSignal.aborted) {
              onAbort();
            } else {
              params.abortSignal.addEventListener("abort", onAbort, {
                once: true,
              });
            }
          }
          let promptError: unknown = null;
          try {
            const promptStartedAt = Date.now();
            log.debug(
              `embedded run prompt start: runId=${params.runId} sessionId=${params.sessionId}`,
            );
            try {
              await session.prompt(params.prompt, {
                images: params.images,
              });
            } catch (err) {
              promptError = err;
            } finally {
              log.debug(
                `embedded run prompt end: runId=${params.runId} sessionId=${params.sessionId} durationMs=${Date.now() - promptStartedAt}`,
              );
            }
            try {
              await waitForCompactionRetry();
            } catch (err) {
              // Capture AbortError from waitForCompactionRetry to enable fallback/rotation.
              if (isAbortError(err)) {
                if (!promptError) promptError = err;
              } else {
                throw err;
              }
            }
            messagesSnapshot = session.messages.slice();
            sessionIdUsed = session.sessionId;
          } finally {
            clearTimeout(abortTimer);
            if (abortWarnTimer) {
              clearTimeout(abortWarnTimer);
              abortWarnTimer = undefined;
            }
            unsubscribe();
            if (ACTIVE_EMBEDDED_RUNS.get(params.sessionId) === queueHandle) {
              ACTIVE_EMBEDDED_RUNS.delete(params.sessionId);
              notifyEmbeddedRunEnded(params.sessionId);
            }
            sessionManager.flushPendingToolResults?.();
            session.dispose();
            await sessionLock.release();
            params.abortSignal?.removeEventListener?.("abort", onAbort);
          }
          if (promptError && !aborted) {
            const errorText = describeUnknownError(promptError);
            if (isContextOverflowError(errorText)) {
              return {
                payloads: [
                  {
                    text:
                      "Context overflow: the conversation history is too large for the model. " +
                      "Use /new or /reset to start a fresh session, or try a model with a larger context window.",
                    isError: true,
                  },
                ],
                meta: {
                  durationMs: Date.now() - started,
                  agentMeta: {
                    sessionId: sessionIdUsed,
                    provider,
                    model: model.id,
                  },
                },
              };
            }
            const promptFailoverReason = classifyFailoverReason(errorText);
            if (
              promptFailoverReason &&
              promptFailoverReason !== "timeout" &&
              lastProfileId
            ) {
              await markAuthProfileFailure({
                store: authStore,
                profileId: lastProfileId,
                reason: promptFailoverReason,
                cfg: params.config,
                agentDir: params.agentDir,
              });
            }
            if (
              isFailoverErrorMessage(errorText) &&
              promptFailoverReason !== "timeout" &&
              (await advanceAuthProfile())
            ) {
              continue;
            }
            const fallbackThinking = pickFallbackThinkingLevel({
              message: errorText,
              attempted: attemptedThinking,
            });
            if (fallbackThinking) {
              log.warn(
                `unsupported thinking level for ${provider}/${modelId}; retrying with ${fallbackThinking}`,
              );
              thinkLevel = fallbackThinking;
              continue;
            }
            throw promptError;
          }

          const lastAssistant = messagesSnapshot
            .slice()
            .reverse()
            .find((m) => (m as AgentMessage)?.role === "assistant") as
            | AssistantMessage
            | undefined;

          const fallbackThinking = pickFallbackThinkingLevel({
            message: lastAssistant?.errorMessage,
            attempted: attemptedThinking,
          });
          if (fallbackThinking && !aborted) {
            log.warn(
              `unsupported thinking level for ${provider}/${modelId}; retrying with ${fallbackThinking}`,
            );
            thinkLevel = fallbackThinking;
            continue;
          }

          const fallbackConfigured =
            (params.config?.agents?.defaults?.model?.fallbacks?.length ?? 0) >
            0;
          const authFailure = isAuthAssistantError(lastAssistant);
          const rateLimitFailure = isRateLimitAssistantError(lastAssistant);
          const failoverFailure = isFailoverAssistantError(lastAssistant);
          const assistantFailoverReason = classifyFailoverReason(
            lastAssistant?.errorMessage ?? "",
          );
          const cloudCodeAssistFormatError = lastAssistant?.errorMessage
            ? isCloudCodeAssistFormatError(lastAssistant.errorMessage)
            : false;

          // Treat timeout as potential rate limit (Antigravity hangs on rate limit)
          const shouldRotate = (!aborted && failoverFailure) || timedOut;

          if (shouldRotate) {
            // Mark current profile for cooldown before rotating
            if (lastProfileId) {
              const reason =
                timedOut || assistantFailoverReason === "timeout"
                  ? "timeout"
                  : (assistantFailoverReason ?? "unknown");
              await markAuthProfileFailure({
                store: authStore,
                profileId: lastProfileId,
                reason,
                cfg: params.config,
                agentDir: params.agentDir,
              });
              if (timedOut) {
                log.warn(
                  `Profile ${lastProfileId} timed out (possible rate limit). Trying next account...`,
                );
              }
              if (cloudCodeAssistFormatError) {
                log.warn(
                  `Profile ${lastProfileId} hit Cloud Code Assist format error. Tool calls will be sanitized on retry.`,
                );
              }
            }
            const rotated = await advanceAuthProfile();
            if (rotated) {
              continue;
            }
            if (fallbackConfigured) {
              const message =
                lastAssistant?.errorMessage?.trim() ||
                (lastAssistant
                  ? formatAssistantErrorText(lastAssistant, {
                      cfg: params.config,
                      sessionKey: params.sessionKey ?? params.sessionId,
                    })
                  : "") ||
                (timedOut
                  ? "LLM request timed out."
                  : rateLimitFailure
                    ? "LLM request rate limited."
                    : authFailure
                      ? "LLM request unauthorized."
                      : "LLM request failed.");
              const status =
                resolveFailoverStatus(assistantFailoverReason ?? "unknown") ??
                (isTimeoutErrorMessage(message) ? 408 : undefined);
              throw new FailoverError(message, {
                reason: assistantFailoverReason ?? "unknown",
                provider,
                model: modelId,
                profileId: lastProfileId,
                status,
              });
            }
          }

          const usage = normalizeUsage(lastAssistant?.usage as UsageLike);
          const agentMeta: EmbeddedPiAgentMeta = {
            sessionId: sessionIdUsed,
            provider: lastAssistant?.provider ?? provider,
            model: lastAssistant?.model ?? model.id,
            usage,
          };

          const replyItems: Array<{
            text: string;
            media?: string[];
            isError?: boolean;
            audioAsVoice?: boolean;
            replyToId?: string;
            replyToTag?: boolean;
            replyToCurrent?: boolean;
          }> = [];

          const errorText = lastAssistant
            ? formatAssistantErrorText(lastAssistant, {
                cfg: params.config,
                sessionKey: params.sessionKey ?? params.sessionId,
              })
            : undefined;

          if (errorText) replyItems.push({ text: errorText, isError: true });

          const inlineToolResults =
            params.verboseLevel === "on" &&
            !params.onPartialReply &&
            !params.onToolResult &&
            toolMetas.length > 0;
          if (inlineToolResults) {
            for (const { toolName, meta } of toolMetas) {
              const agg = formatToolAggregate(toolName, meta ? [meta] : []);
              const {
                text: cleanedText,
                mediaUrls,
                audioAsVoice,
                replyToId,
                replyToTag,
                replyToCurrent,
              } = parseReplyDirectives(agg);
              if (cleanedText)
                replyItems.push({
                  text: cleanedText,
                  media: mediaUrls,
                  audioAsVoice,
                  replyToId,
                  replyToTag,
                  replyToCurrent,
                });
            }
          }

          const reasoningText =
            lastAssistant && params.reasoningLevel === "on"
              ? formatReasoningMessage(extractAssistantThinking(lastAssistant))
              : "";
          if (reasoningText) replyItems.push({ text: reasoningText });

          const fallbackAnswerText = lastAssistant
            ? extractAssistantText(lastAssistant)
            : "";
          const answerTexts = assistantTexts.length
            ? assistantTexts
            : fallbackAnswerText
              ? [fallbackAnswerText]
              : [];
          for (const text of answerTexts) {
            const {
              text: cleanedText,
              mediaUrls,
              audioAsVoice,
              replyToId,
              replyToTag,
              replyToCurrent,
            } = parseReplyDirectives(text);
            if (
              !cleanedText &&
              (!mediaUrls || mediaUrls.length === 0) &&
              !audioAsVoice
            )
              continue;
            replyItems.push({
              text: cleanedText,
              media: mediaUrls,
              audioAsVoice,
              replyToId,
              replyToTag,
              replyToCurrent,
            });
          }

          // Check if any replyItem has audioAsVoice tag - if so, apply to all media payloads
          const hasAudioAsVoiceTag = replyItems.some(
            (item) => item.audioAsVoice,
          );
          const payloads = replyItems
            .map((item) => ({
              text: item.text?.trim() ? item.text.trim() : undefined,
              mediaUrls: item.media?.length ? item.media : undefined,
              mediaUrl: item.media?.[0],
              isError: item.isError,
              replyToId: item.replyToId,
              replyToTag: item.replyToTag,
              replyToCurrent: item.replyToCurrent,
              // Apply audioAsVoice to media payloads if tag was found anywhere in response
              audioAsVoice:
                item.audioAsVoice || (hasAudioAsVoiceTag && item.media?.length),
            }))
            .filter(
              (p) =>
                p.text || p.mediaUrl || (p.mediaUrls && p.mediaUrls.length > 0),
            );

          log.debug(
            `embedded run done: runId=${params.runId} sessionId=${params.sessionId} durationMs=${Date.now() - started} aborted=${aborted}`,
          );
          if (lastProfileId) {
            await markAuthProfileGood({
              store: authStore,
              provider,
              profileId: lastProfileId,
            });
            // Track usage for round-robin rotation
            await markAuthProfileUsed({
              store: authStore,
              profileId: lastProfileId,
            });
          }
          return {
            payloads: payloads.length ? payloads : undefined,
            meta: {
              durationMs: Date.now() - started,
              agentMeta,
              aborted,
            },
            didSendViaMessagingTool: didSendViaMessagingTool(),
            messagingToolSentTexts: getMessagingToolSentTexts(),
            messagingToolSentTargets: getMessagingToolSentTargets(),
          };
        } finally {
          restoreSkillEnv?.();
          process.chdir(prevCwd);
        }
      }
    }),
  );
}