import { createLogger } from '@sim/logger'
import { getErrorMessage, toError } from '@sim/utils/errors'
import { sleep } from '@sim/utils/helpers'
import { isPlainRecord } from '@sim/utils/object'
import { backoffWithJitter, parseRetryAfter } from '@sim/utils/retry'
import { getBYOKKey } from '@/lib/api-key/byok'
import {
  generateInternalToken,
  type InternalSandboxProfile,
  type InternalTokenClaims,
} from '@/lib/auth/internal'
import {
  BILLING_ATTRIBUTION_HEADER,
  type BillingAttributionSnapshot,
  serializeBillingAttributionHeader,
} from '@/lib/billing/core/billing-attribution'
import { isHosted } from '@/lib/core/config/env-flags'
import { DEFAULT_EXECUTION_TIMEOUT_MS, getMaxExecutionTimeout } from '@/lib/core/execution-limits'
import { getHostedKeyRateLimiter } from '@/lib/core/rate-limiter'
import {
  secureFetchWithPinnedIP,
  validateAndPinProxyUrl,
  validateUrlWithDNS,
} from '@/lib/core/security/input-validation.server'
import { PlatformEvents } from '@/lib/core/telemetry'
import { HttpError } from '@/lib/core/utils/http-error'
import { generateRequestId } from '@/lib/core/utils/request'
import {
  isPayloadSizeLimitError,
  readResponseToBufferWithLimit,
} from '@/lib/core/utils/stream-limits'
import { getBaseUrl, getInternalApiBaseUrl } from '@/lib/core/utils/urls'
import { isUserFile } from '@/lib/core/utils/user-file'
import { isSameOrigin } from '@/lib/core/utils/validation'
import { SIM_VIA_HEADER, serializeCallChain } from '@/lib/execution/call-chain'
import {
  INTERNAL_EXECUTION_DEADLINE_HEADER,
  serializeExecutionDeadlineHeader,
} from '@/lib/execution/execution-deadline-header'
import {
  inspectPrivateToolMetadataEnvelope,
  inspectPrivateToolMetadataResponseCapability,
  MAX_PRIVATE_TOOL_METADATA_OVERHEAD_BYTES,
  PRIVATE_TOOL_METADATA_REQUEST_HEADER,
  PRIVATE_TOOL_METADATA_RESPONSE_HEADER,
  type PrivateToolMetadataType,
  RESOLVED_SECRET_NAMES_DURABLE_FILES_METADATA_V2,
  RESOLVED_SECRET_NAMES_FIELD,
  RESOLVED_SECRET_NAMES_METADATA_V1,
  RESOLVED_SECRET_PROVENANCE_FIELD,
  RESOLVED_SECRET_PROVENANCE_METADATA_V1,
} from '@/lib/execution/private-tool-metadata'
import { parseMcpToolId } from '@/lib/mcp/utils'
import { hostedKeyMetrics } from '@/lib/monitoring/metrics'
import { resolveWorkspaceFileReference } from '@/lib/uploads/contexts/workspace/workspace-file-manager'
import { markWorkspaceFileSecretProvenanceUnknown } from '@/lib/uploads/contexts/workspace/workspace-file-secret-provenance'
import { assertPermissionsAllowed } from '@/ee/access-control/utils/permission-check'
import { isCustomTool, isMcpTool } from '@/executor/constants'
import { resolveSkillContent } from '@/executor/handlers/agent/skills-resolver'
import type { ExecutionContext, UserFile } from '@/executor/types'
import { resolveEnvVarReferences } from '@/executor/utils/reference-validation'
import { projectResolvedSecretDiagnosticContent } from '@/executor/utils/resolved-secret-content-projection'
import {
  isResolvedSecretTraceProvenanceV1,
  type ResolvedSecretTraceRegistry,
} from '@/executor/utils/resolved-secret-trace-registry'
import type { ErrorInfo } from '@/tools/error-extractors'
import { extractErrorMessage } from '@/tools/error-extractors'
import { HostedKeyRateLimitedError, HostedKeyUnavailableError } from '@/tools/errors'
import {
  getOwnEnumerableDataEntries,
  prepareToolRequest,
  projectToolModelInputParams,
} from '@/tools/request-transport'
import type {
  BYOKProviderId,
  OAuthTokenPayload,
  ToolConfig,
  ToolHostingPricing,
  ToolResponse,
  ToolRetryConfig,
} from '@/tools/types'
import { getTool, validateRequiredParametersAfterMerge } from '@/tools/utils'
import * as toolsUtilsServer from '@/tools/utils.server'

const logger = createLogger('Tools')
const PRIVATE_TOOL_METADATA_ERROR_MESSAGE = 'Internal tool response metadata could not be verified'
const PRIVATE_MODEL_INPUT_DIRECT_EXECUTION_ERROR_MESSAGE =
  'Private model input provenance is not supported by direct execution'
const PRIVATE_SECRET_PROVENANCE_DIRECT_EXECUTION_ERROR_MESSAGE =
  'Private secret provenance is not supported by direct execution'

function projectToolLogMetadata(
  metadata: Record<string, unknown>,
  registry: ResolvedSecretTraceRegistry | undefined,
  structuralFallback: Record<string, unknown>,
  structuralOnly = false
): Record<string, unknown> {
  if (structuralOnly) return { ...structuralFallback, redacted: true }
  if (!registry) return metadata

  const projection = projectResolvedSecretDiagnosticContent(metadata, registry)
  return projection.safe && isPlainRecord(projection.value)
    ? projection.value
    : { ...structuralFallback, redacted: true }
}

interface ToolExecutionScope {
  workspaceId?: string
  workflowId?: string
  userId?: string
  executionId?: string
  callChain?: string[]
  isDeployedContext?: boolean
  enforceCredentialAccess?: boolean
  copilotToolExecution?: boolean
  billingAttribution?: BillingAttributionSnapshot
}

function resolveToolScope(
  params: Record<string, unknown>,
  executionContext?: ExecutionContext
): ToolExecutionScope {
  const ctx = params._context as Record<string, unknown> | undefined
  return {
    workspaceId: (executionContext?.workspaceId ?? ctx?.workspaceId) as string | undefined,
    workflowId: (executionContext?.workflowId ?? ctx?.workflowId) as string | undefined,
    userId: (executionContext?.userId ?? ctx?.userId) as string | undefined,
    executionId: (executionContext?.executionId ?? ctx?.executionId) as string | undefined,
    callChain: (executionContext?.callChain ?? ctx?.callChain) as string[] | undefined,
    isDeployedContext: (executionContext?.isDeployedContext ?? ctx?.isDeployedContext) as
      | boolean
      | undefined,
    enforceCredentialAccess: (executionContext?.enforceCredentialAccess ??
      ctx?.enforceCredentialAccess) as boolean | undefined,
    copilotToolExecution: (executionContext?.copilotToolExecution ?? ctx?.copilotToolExecution) as
      | boolean
      | undefined,
    billingAttribution: (executionContext?.metadata.billingAttribution ??
      ctx?.billingAttribution) as BillingAttributionSnapshot | undefined,
  }
}

function toUserFileFromWorkspaceRecord(record: {
  id: string
  name: string
  path: string
  url?: string
  size: number
  type: string
  key: string
}): UserFile {
  return {
    id: record.id,
    name: record.name,
    url: record.url ?? record.path,
    size: record.size,
    type: record.type,
    key: record.key,
    context: 'workspace',
  }
}

async function resolveCopilotFileReference(
  value: unknown,
  workspaceId: string,
  paramId: string
): Promise<UserFile | unknown> {
  if (isUserFile(value)) {
    return value
  }

  const referenceId =
    typeof value === 'string'
      ? value
      : value &&
          typeof value === 'object' &&
          typeof (value as Record<string, unknown>).id === 'string'
        ? ((value as Record<string, unknown>).id as string)
        : null

  if (!referenceId) {
    return value
  }

  const fileRecord = await resolveWorkspaceFileReference(workspaceId, referenceId)
  if (!fileRecord) {
    throw new Error(
      `Could not resolve workspace file reference "${referenceId}" for parameter "${paramId}"`
    )
  }

  const resolvedFile = toUserFileFromWorkspaceRecord(fileRecord)
  if (!value || typeof value !== 'object') {
    return resolvedFile
  }

  const candidate = value as Record<string, unknown>
  return {
    ...resolvedFile,
    context: typeof candidate.context === 'string' ? candidate.context : resolvedFile.context,
    base64: typeof candidate.base64 === 'string' ? candidate.base64 : undefined,
  }
}

async function normalizeCopilotFileParams(
  tool: ToolConfig,
  params: Record<string, unknown>,
  scope: ToolExecutionScope
): Promise<void> {
  if (!scope.copilotToolExecution) {
    return
  }

  for (const [paramId, paramDef] of Object.entries(tool.params || {})) {
    const paramType = paramDef?.type
    const currentValue = params[paramId]
    if (currentValue === undefined || currentValue === null) {
      continue
    }

    if (paramType === 'file') {
      if (!scope.workspaceId) {
        throw new Error(`Missing workspaceId while resolving file parameter "${paramId}"`)
      }
      params[paramId] = await resolveCopilotFileReference(currentValue, scope.workspaceId, paramId)
      continue
    }

    if (paramType === 'file[]') {
      if (!scope.workspaceId) {
        throw new Error(`Missing workspaceId while resolving file parameter "${paramId}"`)
      }

      const values = Array.isArray(currentValue) ? currentValue : [currentValue]
      params[paramId] = await Promise.all(
        values.map((item) => resolveCopilotFileReference(item, scope.workspaceId!, paramId))
      )
    }
  }
}

/**
 * Resolves whole-value {{ENV_VAR}} references in user-only params for copilot
 * tool executions. Chat agents never see secret values (the workspace VFS
 * exposes env var names only), so they pass references; workflow runs resolve
 * these in the executor, and this is the equivalent step for direct tool
 * calls, delegating to the executor's resolver so both paths share one set of
 * reference semantics. Resolution is deliberately restricted to params
 * declared `visibility: 'user-only'` (API keys and other operator-supplied
 * secrets) and to values that are exactly one reference, so LLM-writable
 * params (URLs, headers, bodies) can never be used to extract secret values.
 *
 * Mutates only the given params object — callers pass the per-execution copy,
 * never the copilot-side tool-call state, so decrypted values cannot leak
 * into failure logs or persisted chat state.
 */
async function resolveCopilotEnvReferences(
  tool: ToolConfig,
  params: Record<string, unknown>,
  scope: ToolExecutionScope,
  resolvedSecretTraceRegistry?: ResolvedSecretTraceRegistry
): Promise<void> {
  if (!scope.copilotToolExecution) {
    return
  }

  const pending: Array<{ paramId: string; value: string }> = []
  for (const [paramId, paramDef] of Object.entries(tool.params || {})) {
    if (paramDef?.visibility !== 'user-only') continue
    const value = params[paramId]
    if (typeof value === 'string' && value.startsWith('{{') && value.endsWith('}}')) {
      pending.push({ paramId, value })
    }
  }

  if (pending.length === 0) {
    return
  }

  if (!scope.userId) {
    throw new Error(
      `Cannot resolve environment variable reference in parameter "${pending[0].paramId}" without an authenticated user context.`
    )
  }

  const completePendingActivation = resolvedSecretTraceRegistry?.beginPendingActivation()
  try {
    const { getEffectiveDecryptedEnv } = await import('@/lib/environment/utils')
    const envVars = await getEffectiveDecryptedEnv(scope.userId, scope.workspaceId)

    for (const { paramId, value } of pending) {
      const missingKeys: string[] = []
      const resolved = resolveEnvVarReferences(value, envVars, {
        allowEmbedded: false,
        missingKeys,
        onResolved: (name, resolvedValue) => {
          resolvedSecretTraceRegistry?.recordResolvedAtInputPath(name, resolvedValue, [paramId], {
            propagated: true,
          })
        },
      })
      if (missingKeys.length > 0) {
        const scopeHint = scope.workspaceId
          ? ''
          : ' (no workspace context — only personal variables are available here)'
        throw new Error(
          `Environment variable "${missingKeys[0]}" referenced by parameter "${paramId}" was not found${scopeHint}. ` +
            `Check environment/variables.json for available variable names.`
        )
      }
      params[paramId] = resolved as string
      resolvedSecretTraceRegistry?.recordResolvedInputProjection(
        [paramId],
        resolved as string,
        value
      )
    }
  } finally {
    completePendingActivation?.()
  }
}

function readExplicitCredentialSelector(params: Record<string, unknown>): string | undefined {
  for (const key of ['credentialId', 'oauthCredential', 'credential'] as const) {
    const value = params[key]
    if (typeof value === 'string' && value.trim().length > 0) {
      return value.trim()
    }
  }
  return undefined
}

function normalizeCopilotCredentialParams(params: Record<string, unknown>): void {
  const credentialId = typeof params.credentialId === 'string' ? params.credentialId.trim() : ''
  if (credentialId && !params.credential && !params.oauthCredential) {
    params.credential = credentialId
  }
}

function enforceCopilotCredentialSelection(
  toolId: string,
  tool: ToolConfig,
  params: Record<string, unknown>,
  scope: ToolExecutionScope
): void {
  if (!scope.copilotToolExecution || !tool.oauth?.required) {
    return
  }

  if (readExplicitCredentialSelector(params)) {
    return
  }

  const toolLabel = tool.name || toolId
  throw new Error(
    `Copilot must pass credentialId for ${toolLabel}. Read environment/credentials.json and pass the exact credentialId for provider "${tool.oauth.provider}".`
  )
}

/** Result from hosted key injection */
interface HostedKeyInjectionResult {
  isUsingHostedKey: boolean
  envVarName?: string
}

/**
 * Inject hosted API key if tool supports it and user didn't provide one.
 * Checks BYOK workspace keys first, then uses the HostedKeyRateLimiter for round-robin key selection.
 * Returns whether a hosted (billable) key was injected and which env var it came from.
 */
async function injectHostedKeyIfNeeded(
  tool: ToolConfig,
  params: Record<string, unknown>,
  executionContext: ExecutionContext | undefined,
  requestId: string
): Promise<HostedKeyInjectionResult> {
  if (!tool.hosting) return { isUsingHostedKey: false }
  if (!isHosted) return { isUsingHostedKey: false }
  if (tool.hosting.enabled && !tool.hosting.enabled(params)) {
    return { isUsingHostedKey: false }
  }

  const { envKeyPrefix, apiKeyParam, byokProviderId, rateLimit } = tool.hosting
  const userProvidedKey = params[apiKeyParam]
  if (typeof userProvidedKey === 'string' && userProvidedKey.trim().length > 0) {
    return { isUsingHostedKey: false }
  }

  const { workspaceId, userId, workflowId } = resolveToolScope(params, executionContext)

  // Check BYOK workspace key first
  if (byokProviderId && workspaceId) {
    try {
      const byokResult = await getBYOKKey(workspaceId, byokProviderId as BYOKProviderId)
      if (byokResult) {
        params[apiKeyParam] = byokResult.apiKey
        logger.info(`[${requestId}] Using BYOK key for ${tool.id}`)
        return { isUsingHostedKey: false } // Don't bill - user's own key
      }
    } catch (error) {
      logger.error(`[${requestId}] Failed to get BYOK key for ${tool.id}:`, error)
      // Fall through to hosted key
    }
  }

  const rateLimiter = getHostedKeyRateLimiter()
  const provider = byokProviderId || tool.id
  const billingActorId = workspaceId

  if (!billingActorId) {
    logger.error(`[${requestId}] No workspace ID available for hosted key rate limiting`)
    return { isUsingHostedKey: false }
  }

  const acquireResult = await rateLimiter.acquireKey(
    provider,
    envKeyPrefix,
    rateLimit,
    billingActorId,
    executionContext?.abortSignal
  )

  if (!acquireResult.success && acquireResult.billingActorRateLimited) {
    logger.warn(`[${requestId}] Billing actor ${billingActorId} rate limited for ${tool.id}`, {
      provider,
      retryAfterMs: acquireResult.retryAfterMs,
    })

    PlatformEvents.hostedKeyUserThrottled({
      toolId: tool.id,
      reason: 'billing_actor_limit',
      provider,
      retryAfterMs: acquireResult.retryAfterMs ?? 0,
      userId,
      workspaceId,
      workflowId,
    })

    throw new HostedKeyRateLimitedError(
      acquireResult.error || `Rate limit exceeded for ${tool.id}`,
      acquireResult.retryAfterMs
    )
  }

  // Handle no keys configured (503)
  if (!acquireResult.success) {
    logger.error(`[${requestId}] No hosted keys configured for ${tool.id}: ${acquireResult.error}`)
    throw new HostedKeyUnavailableError(
      acquireResult.error || `No hosted keys configured for ${tool.id}`
    )
  }

  params[apiKeyParam] = acquireResult.key
  params.__usingHostedKey = true
  logger.info(`[${requestId}] Using hosted key for ${tool.id} (${acquireResult.envVarName})`, {
    keyIndex: acquireResult.keyIndex,
    provider,
  })

  return {
    isUsingHostedKey: true,
    envVarName: acquireResult.envVarName,
  }
}

/**
 * Re-acquire a hosted key after upstream-429 retries have been exhausted. Calls
 * `acquireKey` (which now blocks on the per-workspace bucket) and re-injects the
 * fresh key into `params`. Returns false if no key could be obtained — caller
 * should re-throw the original upstream 429.
 *
 * Does not consult BYOK. We only enter this path from inside the hosted-key
 * branch of `executeTool`, so BYOK has already been ruled out for this call.
 */
async function reacquireHostedKey(
  tool: ToolConfig,
  params: Record<string, unknown>,
  executionContext: ExecutionContext | undefined,
  requestId: string
): Promise<string | null> {
  if (!tool.hosting) return null
  const { envKeyPrefix, apiKeyParam, byokProviderId, rateLimit } = tool.hosting
  const { workspaceId } = resolveToolScope(params, executionContext)
  if (!workspaceId) return null

  const provider = byokProviderId || tool.id
  const acquireResult = await getHostedKeyRateLimiter().acquireKey(
    provider,
    envKeyPrefix,
    rateLimit,
    workspaceId,
    executionContext?.abortSignal
  )

  if (!acquireResult.success || !acquireResult.key) {
    logger.warn(
      `[${requestId}] Re-acquire of hosted key for ${tool.id} failed: ${acquireResult.error ?? 'unknown'}`
    )
    return null
  }

  params[apiKeyParam] = acquireResult.key
  logger.info(
    `[${requestId}] Re-acquired hosted key for ${tool.id} (${acquireResult.envVarName}) after upstream throttling`
  )
  return acquireResult.envVarName ?? 'unknown'
}

/**
 * Check if an error is a rate limit (throttling) or quota exhaustion error.
 * Some providers (e.g. Perplexity) return 401/403 with "insufficient_quota"
 * instead of the standard 429, so we also inspect the error message.
 */
function isRateLimitError(error: unknown): boolean {
  if (error && typeof error === 'object') {
    const status = (error as { status?: number }).status
    if (status === 429 || status === 503) return true

    if (status === 401 || status === 403) {
      const message = ((error as { message?: string }).message || '').toLowerCase()
      if (message.includes('quota') || message.includes('rate limit')) {
        return true
      }
    }
  }
  return false
}

/**
 * Map a thrown tool error to a hosted-key failure reason for metrics. Mirrors
 * `isRateLimitError`: some providers signal quota/rate-limit via 401/403 with a
 * descriptive message, so those count as `rate_limited`, not `auth`.
 */
function classifyHostedKeyFailure(error: unknown): 'rate_limited' | 'auth' | 'other' {
  const status = (error as { status?: number } | null)?.status
  if (status === 429 || status === 503) return 'rate_limited'
  if (status === 401 || status === 403) {
    const message = ((error as { message?: string } | null)?.message ?? '').toLowerCase()
    if (message.includes('quota') || message.includes('rate limit')) return 'rate_limited'
    return 'auth'
  }
  return 'other'
}

/** Context for retry with rate limit tracking */
interface RetryContext {
  requestId: string
  toolId: string
  provider: string
  envVarName: string
  executionContext?: ExecutionContext
  /**
   * Optional callback invoked after the local exponential backoff has been exhausted by
   * upstream 429s. Should re-enter the per-workspace hosted-key queue (which now blocks
   * on the bucket) and return a fresh execution thunk bound to the newly acquired key.
   * If the callback returns null, we give up and re-throw the last error.
   */
  reacquireAfterRetriesExhausted?: () => Promise<(() => Promise<unknown>) | null>
}

/**
 * Execute a function with exponential backoff retry for rate limiting errors.
 * Only used for hosted key requests. Tracks rate limit events via telemetry.
 *
 * On terminal upstream 429, optionally re-enters the hosted-key queue (which waits for
 * the per-workspace bucket to refill) and retries once with a freshly acquired key.
 * This handles the case where the upstream provider's limit is tighter than ours — we
 * re-queue the call instead of surfacing the error.
 */
async function executeWithRetry<T>(
  fn: () => Promise<T>,
  context: RetryContext,
  maxRetries = 3,
  baseDelayMs = 1000
): Promise<T> {
  const {
    requestId,
    toolId,
    provider,
    envVarName,
    executionContext,
    reacquireAfterRetriesExhausted,
  } = context
  let lastError: unknown

  for (let attempt = 0; attempt <= maxRetries; attempt++) {
    try {
      return await fn()
    } catch (error) {
      lastError = error

      if (!isRateLimitError(error) || attempt === maxRetries) {
        if (isRateLimitError(error) && attempt === maxRetries) {
          if (reacquireAfterRetriesExhausted) {
            try {
              const requeued = await reacquireAfterRetriesExhausted()
              if (requeued) {
                logger.warn(
                  `[${requestId}] Upstream retries exhausted for ${toolId} (${envVarName}); re-queued and retrying once with fresh key`
                )
                return (await requeued()) as T
              }
            } catch (requeueError) {
              logger.error(
                `[${requestId}] Re-queue after exhausted upstream retries failed for ${toolId}`,
                { error: toError(requeueError).message }
              )
            }
          }

          PlatformEvents.hostedKeyUserThrottled({
            toolId,
            reason: 'upstream_retries_exhausted',
            provider,
            userId: executionContext?.userId,
            workspaceId: executionContext?.workspaceId,
            workflowId: executionContext?.workflowId,
          })
        }
        throw error
      }

      const delayMs = backoffWithJitter(attempt + 1, null, { baseMs: baseDelayMs })

      // Track throttling event via telemetry
      PlatformEvents.hostedKeyRateLimited({
        toolId,
        envVarName,
        attempt: attempt + 1,
        maxRetries,
        delayMs,
        userId: executionContext?.userId,
        workspaceId: executionContext?.workspaceId,
        workflowId: executionContext?.workflowId,
      })

      logger.warn(
        `[${requestId}] Rate limited for ${toolId} (${envVarName}), retrying in ${delayMs}ms (attempt ${attempt + 1}/${maxRetries})`
      )
      await sleep(delayMs)
    }
  }

  throw lastError
}

/** Result from cost calculation */
interface ToolCostResult {
  cost: number
  metadata?: Record<string, unknown>
}

/**
 * Rejects a cost that cannot be billed. `NaN` would silently vanish from every
 * downstream sum and `Infinity` would poison the ledger, so a pricing bug must
 * surface as a metering failure instead of a corrupt charge.
 */
function assertBillableCost(cost: unknown, toolId: string): number {
  if (typeof cost !== 'number' || !Number.isFinite(cost) || cost < 0) {
    throw new Error(`Hosted-key pricing for ${toolId} produced an unusable cost: ${String(cost)}`)
  }
  return cost
}

/**
 * Calculate cost based on pricing model
 */
function calculateToolCost(
  pricing: ToolHostingPricing,
  params: Record<string, unknown>,
  response: Record<string, unknown>,
  toolId: string
): ToolCostResult {
  switch (pricing.type) {
    case 'per_request':
      return { cost: assertBillableCost(pricing.cost, toolId) }

    case 'custom': {
      const result = pricing.getCost(params, response)
      if (typeof result === 'number') {
        return { cost: assertBillableCost(result, toolId) }
      }
      return { ...result, cost: assertBillableCost(result.cost, toolId) }
    }

    default: {
      const exhaustiveCheck: never = pricing
      throw new Error(`Unknown pricing type: ${(exhaustiveCheck as ToolHostingPricing).type}`)
    }
  }
}

interface HostedKeyCostResult {
  cost: number
  metadata?: Record<string, unknown>
}

/**
 * Calculate and log hosted key cost for a tool execution.
 * Logs to usageLog for audit trail and returns cost + metadata for output.
 */
async function processHostedKeyCost(
  tool: ToolConfig,
  params: Record<string, unknown>,
  response: Record<string, unknown>,
  executionContext: ExecutionContext | undefined,
  requestId: string
): Promise<HostedKeyCostResult> {
  if (!tool.hosting?.pricing) {
    return { cost: 0 }
  }

  const { cost, metadata } = calculateToolCost(tool.hosting.pricing, params, response, tool.id)

  if (cost <= 0) return { cost: 0 }

  const { userId } = resolveToolScope(params, executionContext)

  if (!userId) return { cost, metadata }

  logger.debug(
    `[${requestId}] Hosted key cost for ${tool.id}: $${cost}`,
    metadata ? { metadata } : {}
  )

  return { cost, metadata }
}

/**
 * Report custom dimension usage after successful hosted-key tool execution.
 * Only applies to tools with `custom` rate limit mode. Fires and logs;
 * failures here do not block the response since execution already succeeded.
 */
async function reportCustomDimensionUsage(
  tool: ToolConfig,
  params: Record<string, unknown>,
  response: Record<string, unknown>,
  executionContext: ExecutionContext | undefined,
  requestId: string
): Promise<void> {
  if (tool.hosting?.rateLimit.mode !== 'custom') return
  const { workspaceId: billingActorId } = resolveToolScope(params, executionContext)
  if (!billingActorId) return

  const rateLimiter = getHostedKeyRateLimiter()
  const provider = tool.hosting.byokProviderId || tool.id

  try {
    const result = await rateLimiter.reportUsage(
      provider,
      billingActorId,
      tool.hosting.rateLimit,
      params,
      response
    )

    for (const dim of result.dimensions) {
      if (!dim.allowed) {
        logger.warn(`[${requestId}] Dimension ${dim.name} overdrawn after ${tool.id} execution`, {
          consumed: dim.consumed,
          tokensRemaining: dim.tokensRemaining,
        })
      }
    }
  } catch (error) {
    logger.error(`[${requestId}] Failed to report custom dimension usage for ${tool.id}:`, error)
  }
}

/**
 * Strips internal fields (keys starting with `__`) from tool output before
 * returning to users. The double-underscore prefix is reserved for transient
 * data (e.g. `__costDollars`) and will never collide with legitimate API
 * fields like `_id`.
 */
function stripInternalFields(output: Record<string, unknown>): Record<string, unknown> {
  if (typeof output !== 'object' || output === null || Array.isArray(output)) {
    return output
  }
  const result: Record<string, unknown> = {}
  for (const [key, value] of Object.entries(output)) {
    if (!key.startsWith('__')) {
      result[key] = value
    }
  }
  return result
}

export function postProcessToolOutput(toolId: string, output: Record<string, unknown>) {
  return isCustomTool(toolId) ? output : stripInternalFields(output)
}

/**
 * Apply post-execution hosted-key cost tracking to a successful tool result.
 * Reports custom dimension usage, calculates cost, and merges it into the output.
 *
 * Billing capture differs by caller:
 * - Workflow executions bill `output.cost.total` through trace spans and the
 *   execution ledger (`recordUsage`), so the `cost` field alone suffices.
 * - Copilot tool executions have no execution ledger. Their only billing hook
 *   is Go's `extractServiceCost`, which reads a top-level `_serviceCost` field
 *   from the tool result and charges it through the per-round update-cost
 *   callback (the same path the media tools use). Without it, hosted-key spend
 *   from copilot-dispatched integration tools is never charged.
 */
async function applyHostedKeyCostToResult(
  finalResult: ToolResponse,
  tool: ToolConfig,
  params: Record<string, unknown>,
  executionContext: ExecutionContext | undefined,
  requestId: string,
  envVarName: string | undefined
): Promise<void> {
  await reportCustomDimensionUsage(tool, params, finalResult.output, executionContext, requestId)

  const provider = tool.hosting?.byokProviderId || tool.id
  const key = envVarName ?? 'unknown'

  let hostedKeyCost = 0
  let metadata: Record<string, unknown> | undefined

  try {
    ;({ cost: hostedKeyCost, metadata } = await processHostedKeyCost(
      tool,
      params,
      finalResult.output,
      executionContext,
      requestId
    ))
  } catch (error) {
    // The provider already ran and already charged Sim's key. Failing the
    // execution here would destroy the caller's result without recovering the
    // spend, so the run stands and the gap is raised for reconciliation.
    logger.error(
      `[${requestId}] Hosted-key metering failed for ${tool.id}; execution succeeded unbilled`,
      { provider, error: getErrorMessage(error) }
    )
    hostedKeyMetrics.recordFailed({ provider, tool: tool.id, key, reason: 'metering' })
  }

  hostedKeyMetrics.recordUsed({ provider, tool: tool.id, key })
  hostedKeyMetrics.recordCostCharged(hostedKeyCost, { provider, tool: tool.id })

  if (hostedKeyCost > 0) {
    const { copilotToolExecution } = resolveToolScope(params, executionContext)
    finalResult.output = {
      ...finalResult.output,
      cost: {
        ...metadata,
        total: hostedKeyCost,
      },
      // Copilot-only: workflow runs must not emit _serviceCost or the cost
      // would be billed twice (execution ledger + Go service charge).
      ...(copilotToolExecution ? { _serviceCost: { service: provider, cost: hostedKeyCost } } : {}),
    }
  }
}

import { normalizeToolId } from '@/tools/normalize'

/**
 * Maximum request body size in bytes before we warn/error about size limits.
 * Next.js 16 has a default middleware/proxy body limit of 10MB.
 */
const MAX_REQUEST_BODY_SIZE_BYTES = 10 * 1024 * 1024 // 10MB
const MAX_TOOL_RESPONSE_BODY_BYTES = 10 * 1024 * 1024 // 10MB

/**
 * User-friendly error message for body size limit exceeded
 */
const BODY_SIZE_LIMIT_ERROR_MESSAGE =
  'Request body size limit exceeded (10MB). The workflow data is too large to process. Try reducing the size of variables, inputs, or data being passed between blocks.'

const RESPONSE_SIZE_LIMIT_ERROR_MESSAGE =
  'Tool response size limit exceeded (10MB). The response is too large to keep in workflow data. Reduce the response size or return a file reference instead.'

/**
 * Validates request body size and throws a user-friendly error if exceeded
 * @param body - The request body string to check
 * @param requestId - Request ID for logging
 * @param context - Context string for logging (e.g., toolId)
 * @throws Error if body size exceeds the limit
 */
function validateRequestBodySize(
  body: string | undefined,
  requestId: string,
  context: string
): void {
  if (!body) return

  const bodySize = Buffer.byteLength(body, 'utf8')
  if (bodySize > MAX_REQUEST_BODY_SIZE_BYTES) {
    const bodySizeMB = (bodySize / (1024 * 1024)).toFixed(2)
    const maxSizeMB = (MAX_REQUEST_BODY_SIZE_BYTES / (1024 * 1024)).toFixed(0)
    logger.error(`[${requestId}] Request body size exceeds limit for ${context}:`, {
      bodySize,
      bodySizeMB: `${bodySizeMB}MB`,
      maxSize: MAX_REQUEST_BODY_SIZE_BYTES,
      maxSizeMB: `${maxSizeMB}MB`,
    })
    throw new Error(BODY_SIZE_LIMIT_ERROR_MESSAGE)
  }
}

/**
 * Checks if an error message indicates a body size limit issue
 * @param errorMessage - The error message to check
 * @returns true if the error is related to body size limits
 */
function isBodySizeLimitError(errorMessage: string): boolean {
  const lowerMessage = errorMessage.toLowerCase()
  return (
    lowerMessage.includes('body size') ||
    lowerMessage.includes('payload too large') ||
    lowerMessage.includes('entity too large') ||
    lowerMessage.includes('request entity too large') ||
    lowerMessage.includes('body_not_allowed') ||
    lowerMessage.includes('request body larger than')
  )
}

/**
 * Handles body size limit errors by logging and throwing a user-friendly error
 * @param error - The original error
 * @param requestId - Request ID for logging
 * @param context - Context string for logging (e.g., toolId)
 * @throws Error with user-friendly message if it's a size limit error
 * @returns false if not a size limit error (caller should continue handling)
 */
function handleBodySizeLimitError(
  error: unknown,
  requestId: string,
  context: string,
  resolvedSecretTraceRegistry?: ResolvedSecretTraceRegistry,
  structuralOnlyWithoutRegistry = false
): boolean {
  const errorMessage = toError(error).message

  if (isBodySizeLimitError(errorMessage)) {
    logger.error(
      `[${requestId}] Request body size limit exceeded for ${context}:`,
      projectToolLogMetadata(
        { originalError: errorMessage },
        resolvedSecretTraceRegistry,
        {
          hasOriginalError: errorMessage.length > 0,
        },
        structuralOnlyWithoutRegistry
      )
    )
    throw new Error(BODY_SIZE_LIMIT_ERROR_MESSAGE)
  }

  return false
}

function handleResponseSizeLimitError(error: unknown, requestId: string, context: string): boolean {
  if (!isPayloadSizeLimitError(error)) return false

  logger.error(`[${requestId}] Response body size limit exceeded for ${context}:`, {
    label: error.label,
    maxBytes: error.maxBytes,
    observedBytes: error.observedBytes,
  })
  throw new Error(RESPONSE_SIZE_LIMIT_ERROR_MESSAGE)
}

function cloneResponseHeaders(headers: Headers | HeadersInit | undefined): Headers {
  const clonedHeaders = new Headers()
  if (!headers) return clonedHeaders

  if (typeof (headers as Headers).forEach === 'function') {
    ;(headers as Headers).forEach((value, key) => {
      clonedHeaders.set(key, value)
    })
    return clonedHeaders
  }

  return new Headers(headers)
}

async function readToolResponseBody(
  response: {
    ok?: boolean
    headers?: { get(name: string): string | null }
    body?: ReadableStream<Uint8Array> | null
    arrayBuffer?: () => Promise<ArrayBuffer>
    text?: () => Promise<string>
  },
  options: {
    requestId: string
    toolId: string
    signal?: AbortSignal
    maxBytes?: number
  }
): Promise<Buffer> {
  try {
    return await readResponseToBufferWithLimit(response, {
      maxBytes: options.maxBytes ?? MAX_TOOL_RESPONSE_BODY_BYTES,
      label: `${options.toolId} response body`,
      signal: options.signal,
      allowNoBodyFallback: true,
    })
  } catch (error) {
    if (isPayloadSizeLimitError(error) || response.ok !== false) {
      throw error
    }

    logger.warn(
      `[${options.requestId}] Failed to read non-OK response body for ${options.toolId}`,
      {
        errorName: toError(error).name,
      }
    )
    return Buffer.alloc(0)
  }
}

/**
 * System parameters that should be filtered out when extracting tool arguments
 * These are internal parameters used by the execution framework, not tool inputs
 */
const MCP_SYSTEM_PARAMETERS = new Set([
  'serverId',
  'serverUrl',
  'toolName',
  'serverName',
  '_context',
  'envVars',
  'workflowVariables',
  'blockData',
  'blockNameMapping',
  '_toolSchema',
])

/**
 * Create an Error instance from errorInfo and attach useful context
 * Uses the error extractor registry to find the best error message
 */
function createTransformedErrorFromErrorInfo(errorInfo?: ErrorInfo, extractorId?: string): Error {
  const message = extractErrorMessage(errorInfo, extractorId)
  const transformed = new Error(message)
  Object.assign(transformed, {
    status: errorInfo?.status,
    statusText: errorInfo?.statusText,
    data: errorInfo?.data,
  })
  return transformed
}

/**
 * Process file outputs for a tool result if execution context is available
 * Uses dynamic imports to avoid client-side bundling issues
 */
async function processFileOutputs(
  result: ToolResponse,
  tool: ToolConfig,
  executionContext?: ExecutionContext
): Promise<ToolResponse> {
  // Skip file processing if no execution context or not successful
  if (!executionContext || !result.success) {
    return result
  }

  // Skip file processing on client-side (no Node.js modules available)
  if (typeof window !== 'undefined') {
    return result
  }

  try {
    // Dynamic import to avoid client-side bundling issues
    const { FileToolProcessor } = await import('@/executor/utils/file-tool-processor')

    // Check if tool has file outputs
    if (!FileToolProcessor.hasFileOutputs(tool)) {
      return result
    }

    const processedOutput = await FileToolProcessor.processToolOutputs(
      result.output,
      tool,
      executionContext
    )

    return {
      ...result,
      output: processedOutput,
    }
  } catch (error) {
    const normalizedError = toError(error)
    logger.error(
      `Error processing file outputs for tool ${tool.id}:`,
      projectToolLogMetadata(
        {
          error: normalizedError.message,
          stack: error instanceof Error ? error.stack : undefined,
        },
        executionContext.resolvedSecretTraceRegistry,
        {
          errorName: normalizedError.name,
          hasStack: Boolean(error instanceof Error && error.stack),
        },
        tool.id === 'function_execute' || isCustomTool(tool.id)
      )
    )
    // Return original result if file processing fails
    return result
  }
}

export interface ExecuteToolOptions {
  skipPostProcess?: boolean
  executionContext?: ExecutionContext
  signal?: AbortSignal
  resolvedSecretTraceRegistry?: ResolvedSecretTraceRegistry
  /** Trusted base image claim for an internal Function execution. */
  internalSandboxProfile?: InternalSandboxProfile
}

interface PrivateToolResponseMetadataResult {
  response: Response
}

interface PrivateToolMetadataPolicy {
  type: PrivateToolMetadataType
  incomplete: 'reject' | 'propagate'
}

type PrivateToolMetadataConsumption = 'verified' | 'incomplete' | 'invalid'

function getFunctionExportedWorkspaceFileIds(payload: Record<string, unknown>): string[] {
  const ids = new Set<string>()
  const addId = (value: unknown): void => {
    if (typeof value === 'string' && value.length > 0) ids.add(value)
  }
  const output = isPlainRecord(payload.output) ? payload.output : undefined
  const result = output && isPlainRecord(output.result) ? output.result : undefined
  if (result) {
    addId(result.fileId)
    if (Array.isArray(result.files)) {
      for (const file of result.files) {
        if (isPlainRecord(file)) addId(file.fileId)
      }
    }
  }
  if (Array.isArray(payload.resources)) {
    for (const resource of payload.resources) {
      if (isPlainRecord(resource) && resource.type === 'file') addId(resource.id)
    }
  }
  return [...ids]
}

function consumeResolvedSecretNames(
  payload: unknown,
  params: Record<string, any>,
  registry?: ResolvedSecretTraceRegistry
): boolean {
  if (!payload || typeof payload !== 'object' || Array.isArray(payload)) return false

  const response = payload as Record<string, unknown>
  if (!Object.hasOwn(response, RESOLVED_SECRET_NAMES_FIELD)) return false

  const names = response[RESOLVED_SECRET_NAMES_FIELD]
  response[RESOLVED_SECRET_NAMES_FIELD] = undefined
  if (!Array.isArray(names) || !names.every((name) => typeof name === 'string')) {
    return false
  }

  const envVars = params.envVars
  if (!envVars || typeof envVars !== 'object' || Array.isArray(envVars)) {
    return false
  }

  const targetRegistry = registry?.forkForToolCall()
  for (const name of names) {
    const value = (envVars as Record<string, unknown>)[name]
    if (typeof value !== 'string') return false
    if (targetRegistry && !targetRegistry.recordResolved(name, value, { propagated: true })) {
      return false
    }
  }
  if (registry && targetRegistry) registry.mergeToolCallRegistry(targetRegistry)
  return true
}

async function consumeResolvedSecretProvenance(
  payload: unknown,
  registry?: ResolvedSecretTraceRegistry
): Promise<boolean> {
  if (!payload || typeof payload !== 'object' || Array.isArray(payload)) return false

  const response = payload as Record<string, unknown>
  if (!Object.hasOwn(response, RESOLVED_SECRET_PROVENANCE_FIELD)) return false

  const provenance = response[RESOLVED_SECRET_PROVENANCE_FIELD]
  response[RESOLVED_SECRET_PROVENANCE_FIELD] = undefined
  if (!isResolvedSecretTraceProvenanceV1(provenance)) return false
  if (!registry) return true

  const targetRegistry = registry.forkForToolCall()
  const imported = await targetRegistry.importCrossingProvenance(provenance, response, {
    trusted: true,
  })
  if (!imported) return false
  registry.mergeToolCallRegistry(targetRegistry)
  return true
}

function rebuildResponseWithoutPrivateToolMetadata(
  response: Response,
  payload: Record<string, unknown>
): Response {
  payload[RESOLVED_SECRET_NAMES_FIELD] = undefined
  payload[RESOLVED_SECRET_PROVENANCE_FIELD] = undefined
  const headers = new Headers(response.headers)
  headers.delete('content-length')
  headers.delete(PRIVATE_TOOL_METADATA_RESPONSE_HEADER)
  return new Response(JSON.stringify(payload), {
    status: response.status,
    statusText: response.statusText,
    headers,
  })
}

function rebuildSafePrivateToolResponse(response: Response): Response {
  const hasHttpErrorStatus = response.status >= 400 && response.status <= 599
  const status = hasHttpErrorStatus ? response.status : 502
  const error = hasHttpErrorStatus
    ? `Internal tool request failed (HTTP ${response.status})`
    : PRIVATE_TOOL_METADATA_ERROR_MESSAGE
  const headers = new Headers({ 'content-type': 'application/json' })
  return new Response(
    JSON.stringify({
      success: false,
      error,
    }),
    {
      status,
      ...(!hasHttpErrorStatus ? { statusText: 'Bad Gateway' } : {}),
      headers,
    }
  )
}

async function consumePrivateToolPayloadMetadata(
  payload: unknown,
  headers: Headers,
  requestedType: PrivateToolMetadataType | undefined,
  params: Record<string, any>,
  registry?: ResolvedSecretTraceRegistry
): Promise<PrivateToolMetadataConsumption> {
  if (!requestedType) return 'verified'

  const inspection = inspectPrivateToolMetadataEnvelope(headers, payload, requestedType)
  const record =
    payload !== null && typeof payload === 'object' && !Array.isArray(payload)
      ? (payload as Record<string, unknown>)
      : undefined

  if (requestedType === RESOLVED_SECRET_NAMES_DURABLE_FILES_METADATA_V2 && record) {
    const capability = inspectPrivateToolMetadataResponseCapability(headers, requestedType)
    const isLegacyNamesProducer =
      capability.status === 'mismatched' &&
      capability.receivedType === RESOLVED_SECRET_NAMES_METADATA_V1
    if (capability.status === 'unsupported' || isLegacyNamesProducer) {
      if (isLegacyNamesProducer) {
        const legacyInspection = inspectPrivateToolMetadataEnvelope(
          headers,
          record,
          RESOLVED_SECRET_NAMES_METADATA_V1
        )
        if (legacyInspection.status !== 'verified') return 'invalid'
        if (!consumeResolvedSecretNames(record, params, registry)) return 'invalid'
      } else {
        if (inspection.status !== 'unsupported') return 'invalid'
      }

      const fileIds = getFunctionExportedWorkspaceFileIds(record)
      if (fileIds.length > 0) {
        const workspaceId =
          typeof params.workspaceId === 'string'
            ? params.workspaceId
            : typeof params._context?.workspaceId === 'string'
              ? params._context.workspaceId
              : undefined
        if (!workspaceId) return 'invalid'
        await markWorkspaceFileSecretProvenanceUnknown(workspaceId, fileIds)
      }
      record[RESOLVED_SECRET_NAMES_FIELD] = undefined
      return registry?.isPermanentlyIncomplete() ? 'incomplete' : 'verified'
    }
  }

  if (inspection.status === 'unsupported') {
    return 'verified'
  }

  if (inspection.status === 'invalid' || !record) {
    return 'invalid'
  }

  try {
    if (
      requestedType === RESOLVED_SECRET_NAMES_METADATA_V1 ||
      requestedType === RESOLVED_SECRET_NAMES_DURABLE_FILES_METADATA_V2
    ) {
      if (!consumeResolvedSecretNames(record, params, registry)) return 'invalid'
    } else {
      if (!(await consumeResolvedSecretProvenance(record, registry))) return 'invalid'
    }
  } catch {
    return 'invalid'
  }

  record[RESOLVED_SECRET_NAMES_FIELD] = undefined
  record[RESOLVED_SECRET_PROVENANCE_FIELD] = undefined
  return registry?.isPermanentlyIncomplete() ? 'incomplete' : 'verified'
}

async function consumePrivateToolResponseMetadata(
  response: Response,
  requestedType: PrivateToolMetadataType | undefined,
  params: Record<string, any>,
  registry?: ResolvedSecretTraceRegistry
): Promise<PrivateToolResponseMetadataResult> {
  if (!requestedType) return { response }

  let payload: unknown
  try {
    payload = await response.clone().json()
  } catch {
    const inspection = inspectPrivateToolMetadataEnvelope(
      response.headers,
      undefined,
      requestedType
    )
    if (inspection.status === 'invalid') {
      return { response: rebuildSafePrivateToolResponse(response) }
    }
    return { response }
  }

  const consumption = await consumePrivateToolPayloadMetadata(
    payload,
    response.headers,
    requestedType,
    params,
    registry
  )
  if (consumption === 'invalid') {
    return { response: rebuildSafePrivateToolResponse(response) }
  }

  if (payload === null || typeof payload !== 'object' || Array.isArray(payload)) {
    return { response }
  }
  return {
    response: rebuildResponseWithoutPrivateToolMetadata(
      response,
      payload as Record<string, unknown>
    ),
  }
}

function getPrivateToolMetadataPolicy(toolId: string): PrivateToolMetadataPolicy | undefined {
  const normalizedToolId = normalizeToolId(toolId)
  if (normalizedToolId === 'file_get_content' || isMcpTool(normalizedToolId)) {
    return { type: RESOLVED_SECRET_PROVENANCE_METADATA_V1, incomplete: 'reject' }
  }
  if (normalizedToolId === 'function_execute' || isCustomTool(normalizedToolId)) {
    return { type: RESOLVED_SECRET_NAMES_DURABLE_FILES_METADATA_V2, incomplete: 'reject' }
  }
  const configured = getTool(normalizedToolId)?.request.secretProvenance?.response
  if (configured) {
    return { type: RESOLVED_SECRET_PROVENANCE_METADATA_V1, incomplete: configured.incomplete }
  }
  return undefined
}

/**
 * Runs private-provenance tools against an isolated registry. Unavailable authenticated lineage
 * marks the parent unknown without replacing the tool's functional result; malformed metadata is
 * rejected inside the transport consumer and never committed to the parent.
 */
export async function executeTool(
  toolId: string,
  params: Record<string, any>,
  options: ExecuteToolOptions = {}
): Promise<ToolResponse> {
  const parentRegistry =
    options.resolvedSecretTraceRegistry ?? options.executionContext?.resolvedSecretTraceRegistry
  const privateMetadataPolicy = getPrivateToolMetadataPolicy(toolId)
  if (!parentRegistry || !privateMetadataPolicy) {
    return executeToolImplementation(toolId, params, options)
  }
  if (privateMetadataPolicy.incomplete === 'propagate') {
    return executeToolImplementation(toolId, params, options)
  }

  const paramEntries = getOwnEnumerableDataEntries(params)
  const toolRegistry = paramEntries
    ? parentRegistry.forkForInputPaths(paramEntries.map(([key]) => [key] as const))
    : parentRegistry.forkForToolCall()
  if (!paramEntries) toolRegistry.markIncomplete()
  const executionContext = options.executionContext
    ? { ...options.executionContext, resolvedSecretTraceRegistry: toolRegistry }
    : undefined
  let result: ToolResponse
  try {
    result = await executeToolImplementation(toolId, params, {
      ...options,
      ...(executionContext ? { executionContext } : {}),
      resolvedSecretTraceRegistry: toolRegistry,
    })
  } catch (error) {
    parentRegistry.mergeToolCallRegistry(toolRegistry)
    throw error
  }

  parentRegistry.mergeToolCallRegistry(toolRegistry)
  return result
}

/**
 * Execute a tool by making the appropriate HTTP request
 * All requests go directly - internal routes use regular fetch, external use SSRF-protected fetch
 */
async function executeToolImplementation(
  toolId: string,
  params: Record<string, any>,
  options: ExecuteToolOptions = {}
): Promise<ToolResponse> {
  const {
    skipPostProcess = false,
    executionContext,
    signal,
    resolvedSecretTraceRegistry: explicitResolvedSecretTraceRegistry,
    internalSandboxProfile,
  } = options
  const resolvedSecretTraceRegistry =
    explicitResolvedSecretTraceRegistry ?? executionContext?.resolvedSecretTraceRegistry
  const executeNestedTool: typeof executeTool = (nestedToolId, nestedParams, nestedOptions = {}) =>
    executeTool(nestedToolId, nestedParams, {
      ...nestedOptions,
      resolvedSecretTraceRegistry:
        nestedOptions.resolvedSecretTraceRegistry ?? resolvedSecretTraceRegistry,
    })
  // Fall back to the workflow execution's abort signal so plan-based execution timeouts
  // and cancellation propagate to tool fetches when the caller passes no explicit signal.
  const effectiveSignal = signal ?? executionContext?.abortSignal
  // Capture start time for precise timing
  const startTime = new Date()
  const startTimeISO = startTime.toISOString()
  const requestId = generateRequestId()
  const privateToolMetadataPolicy = resolvedSecretTraceRegistry
    ? getPrivateToolMetadataPolicy(toolId)
    : undefined
  const structuralOnlyToolLogs =
    normalizeToolId(toolId) === 'function_execute' ||
    isCustomTool(toolId) ||
    privateToolMetadataPolicy !== undefined

  // Hoisted so the outer catch can attribute a thrown failure to the chosen key.
  let hostedKeyForMetrics: { provider: string; tool: string; key: string } | undefined
  let completePendingSecretActivation: (() => void) | undefined

  try {
    let tool: ToolConfig | undefined

    // Normalize tool ID to strip resource suffixes (e.g., workflow_executor_<uuid> -> workflow_executor)
    const normalizedToolId = normalizeToolId(toolId)
    if (internalSandboxProfile && normalizedToolId !== 'function_execute') {
      throw new Error('An internal sandbox profile may only be used with function_execute')
    }

    const scope = resolveToolScope(params, executionContext)

    const toolKind: 'skill' | 'custom' | 'mcp' | undefined =
      normalizedToolId === 'load_skill'
        ? 'skill'
        : isCustomTool(normalizedToolId)
          ? 'custom'
          : isMcpTool(normalizedToolId)
            ? 'mcp'
            : undefined
    const privateToolMetadataType = privateToolMetadataPolicy?.type

    if (resolvedSecretTraceRegistry && privateToolMetadataType) {
      completePendingSecretActivation = resolvedSecretTraceRegistry.beginPendingActivation()
    }

    // Runs for ALL tools (not just kinded ones) so the per-tool `deniedTools`
    // denylist is enforced alongside the existing mcp/custom/skill gates.
    if (scope.userId && scope.workspaceId) {
      await assertPermissionsAllowed({
        userId: scope.userId,
        workspaceId: scope.workspaceId,
        toolId: normalizedToolId,
        toolKind,
        ctx: executionContext,
      })
    }

    if (normalizedToolId === 'load_skill') {
      const skillName = params.skill_name
      if (!skillName || !scope.workspaceId) {
        return {
          success: false,
          output: { error: 'Missing skill_name or workspace context' },
          error: 'Missing skill_name or workspace context',
        }
      }
      const content = await resolveSkillContent(skillName, scope.workspaceId)
      if (!content) {
        return {
          success: false,
          output: { error: `Skill "${skillName}" not found` },
          error: `Skill "${skillName}" not found`,
        }
      }
      return {
        success: true,
        output: { content },
      }
    }

    if (isCustomTool(normalizedToolId)) {
      tool = await toolsUtilsServer.getToolAsync(normalizedToolId, {
        workflowId: scope.workflowId,
        userId: scope.userId,
        workspaceId: scope.workspaceId,
      })
      if (!tool) {
        logger.error(`[${requestId}] Custom tool not found: ${normalizedToolId}`)
      }
    } else if (isMcpTool(normalizedToolId)) {
      return await executeMcpTool(
        normalizedToolId,
        params,
        executionContext,
        requestId,
        startTimeISO,
        effectiveSignal,
        resolvedSecretTraceRegistry,
        privateToolMetadataType
      )
    } else {
      // For built-in tools, use the synchronous version
      tool = getTool(normalizedToolId)
      if (!tool) {
        logger.error(`[${requestId}] Built-in tool not found: ${normalizedToolId}`)
      }
    }

    // Ensure context is preserved if it exists
    const contextParams = { ...params }
    if (scope.billingAttribution) {
      contextParams._context = {
        ...(contextParams._context as Record<string, unknown> | undefined),
        billingAttribution: scope.billingAttribution,
      }
    }

    // Validate the tool and its parameters
    validateRequiredParametersAfterMerge(toolId, tool, contextParams)

    // After validation, we know tool exists
    if (!tool) {
      throw new Error(`Tool not found: ${toolId}`)
    }

    await normalizeCopilotFileParams(tool, contextParams, scope)
    normalizeCopilotCredentialParams(contextParams)
    enforceCopilotCredentialSelection(toolId, tool, contextParams, scope)
    await resolveCopilotEnvReferences(tool, contextParams, scope, resolvedSecretTraceRegistry)

    // Inject hosted API key if tool supports it and user didn't provide one
    const hostedKeyInfo = await injectHostedKeyIfNeeded(
      tool,
      contextParams,
      executionContext,
      requestId
    )

    if (hostedKeyInfo.isUsingHostedKey) {
      hostedKeyForMetrics = {
        provider: tool.hosting?.byokProviderId || tool.id,
        tool: tool.id,
        key: hostedKeyInfo.envVarName ?? 'unknown',
      }
    }

    // If we have a credential parameter, fetch the access token
    if (contextParams.oauthCredential) {
      contextParams.credential = contextParams.oauthCredential
    }
    if (contextParams.credential) {
      logger.info(
        `[${requestId}] Tool ${toolId} needs access token for credential: ${contextParams.credential}`
      )
      try {
        const baseUrl = getInternalApiBaseUrl()

        const workflowId = contextParams._context?.workflowId
        const userId = contextParams._context?.userId

        const tokenPayload: OAuthTokenPayload = {
          credentialId: contextParams.credential as string,
        }
        if (workflowId) {
          tokenPayload.workflowId = workflowId
        }
        if (contextParams.impersonateUserEmail) {
          tokenPayload.impersonateEmail = contextParams.impersonateUserEmail as string
        }
        if (tool?.oauth?.provider) {
          const { getCanonicalScopesForProvider } = await import('@/lib/oauth/utils')
          const providerScopes = getCanonicalScopesForProvider(tool.oauth.provider)
          if (providerScopes.length > 0) {
            tokenPayload.scopes = providerScopes
          }
        }

        logger.info(`[${requestId}] Fetching access token from ${baseUrl}/api/auth/oauth/token`)

        const tokenUrlObj = new URL('/api/auth/oauth/token', baseUrl)
        if (workflowId) {
          tokenUrlObj.searchParams.set('workflowId', workflowId)
        }
        if (userId && contextParams._context?.enforceCredentialAccess) {
          tokenUrlObj.searchParams.set('userId', userId)
        }

        // Always send Content-Type; add internal auth on server-side runs
        const tokenHeaders: Record<string, string> = { 'Content-Type': 'application/json' }
        if (typeof window === 'undefined') {
          try {
            const internalToken = await generateInternalToken(userId)
            tokenHeaders.Authorization = `Bearer ${internalToken}`
          } catch (_e) {
            // Swallow token generation errors; the request will fail and be reported upstream
          }
        }

        const response = await fetch(tokenUrlObj.toString(), {
          method: 'POST',
          headers: tokenHeaders,
          body: JSON.stringify(tokenPayload),
        })

        if (!response.ok) {
          const errorText = await response.text()
          logger.error(`[${requestId}] Token fetch failed for ${toolId}:`, {
            status: response.status,
            error: errorText,
          })
          let parsedError = errorText
          try {
            const parsed = JSON.parse(errorText)
            if (parsed.error) parsedError = parsed.error
          } catch {
            // Use raw text
          }
          const toolLabel = tool?.name || toolId
          throw new Error(`Failed to obtain credential for ${toolLabel}: ${parsedError}`)
        }

        const data = await response.json()
        contextParams.accessToken = data.accessToken
        if (data.idToken) {
          contextParams.idToken = data.idToken
        }
        if (data.instanceUrl) {
          contextParams.instanceUrl = data.instanceUrl
        }
        if (data.apiDomain && !contextParams.apiDomain) {
          contextParams.apiDomain = data.apiDomain
        }
        if (data.cloudId && !contextParams.cloudId) {
          contextParams.cloudId = data.cloudId
        }
        if (data.domain && !contextParams.domain) {
          contextParams.domain = data.domain
        }
        if (data.authStyle && !contextParams.authStyle) {
          contextParams.authStyle = data.authStyle
        }

        logger.info(`[${requestId}] Successfully got access token for ${toolId}`)

        // Preserve credential for downstream transforms while removing it from request payload
        // so we don't leak it to external services.
        if (contextParams.credential) {
          ;(contextParams as any)._credentialId = contextParams.credential
        }
        if (workflowId) {
          ;(contextParams as any)._workflowId = workflowId
        }
        // Clean up params we don't need to pass to the actual tool
        contextParams.credential = undefined
        contextParams.impersonateUserEmail = undefined
        if (contextParams.workflowId) contextParams.workflowId = undefined
      } catch (error: any) {
        logger.error(`[${requestId}] Error fetching access token for ${toolId}:`, {
          error: toError(error).message,
        })
        throw error
      }
    }

    // Custom blocks (deploy-as-block) run in-process through WorkflowBlockHandler.
    // The runner is dynamic-imported from a server-only module so the client-bundled
    // tool registry never pulls in the executor/db dependency graph (a static or
    // dynamic executor import in the tool descriptor itself would break the client
    // build — and with it `getTool('workflow_executor')`).
    if (normalizedToolId === 'deployed_block_executor') {
      logger.info(`[${requestId}] Running custom block tool ${toolId}`)
      const { runCustomBlockTool } = await import(
        '@/executor/handlers/workflow/custom-block-tool-runner'
      )
      // Forward the INVOKING run's identifiers so the child's log correlation
      // names a real execution instead of a freshly-minted phantom id. Taken
      // from the server-resolved scope, never from model-supplied params.
      const result = await runCustomBlockTool(
        {
          ...contextParams,
          _context: {
            ...(contextParams._context as Record<string, unknown> | undefined),
            ...(scope.executionId ? { executionId: scope.executionId } : {}),
            requestId,
          },
        },
        {
          abortSignal: effectiveSignal,
          resolvedSecretTraceRegistry,
        }
      )
      const endTime = new Date()
      return {
        ...result,
        // Strip internal `__`-prefixed fields the same way every other tool path does,
        // so child-workflow internals never reach the agent's tool result.
        output: postProcessToolOutput(normalizedToolId, result.output ?? {}),
        timing: {
          startTime: startTimeISO,
          endTime: endTime.toISOString(),
          duration: endTime.getTime() - startTime.getTime(),
        },
      }
    }

    // Check for direct execution (no HTTP request needed)
    if (tool.directExecution) {
      logger.info(`[${requestId}] Using directExecution for ${toolId}`)
      if (
        tool.request.modelInput?.mode === 'private-provenance' ||
        (tool.request.modelInput?.mode === 'project' &&
          tool.request.modelInput.privateInputPaths !== undefined)
      ) {
        throw new Error(PRIVATE_MODEL_INPUT_DIRECT_EXECUTION_ERROR_MESSAGE)
      }
      if (tool.request.secretProvenance) {
        throw new Error(PRIVATE_SECRET_PROVENANCE_DIRECT_EXECUTION_ERROR_MESSAGE)
      }
      const directExecutionInput = projectToolModelInputParams(
        tool,
        contextParams,
        resolvedSecretTraceRegistry
      )
      const result = await tool.directExecution(directExecutionInput, effectiveSignal)

      // Apply post-processing if available and not skipped
      let finalResult = result
      if (tool.postProcess && result.success && !skipPostProcess) {
        try {
          finalResult = await tool.postProcess(result, contextParams, executeNestedTool)
        } catch (error) {
          const normalizedError = toError(error)
          logger.error(
            `[${requestId}] Post-processing error for ${toolId}:`,
            projectToolLogMetadata(
              { error: normalizedError.message },
              resolvedSecretTraceRegistry,
              { errorName: normalizedError.name },
              structuralOnlyToolLogs
            )
          )
          finalResult = result
        }
      }

      // Process file outputs if execution context is available
      finalResult = await processFileOutputs(finalResult, tool, executionContext)

      // Add timing data to the result
      const endTime = new Date()
      const endTimeISO = endTime.toISOString()
      const duration = endTime.getTime() - startTime.getTime()

      if (hostedKeyInfo.isUsingHostedKey && finalResult.success) {
        await applyHostedKeyCostToResult(
          finalResult,
          tool,
          contextParams,
          executionContext,
          requestId,
          hostedKeyInfo.envVarName
        )
      } else if (hostedKeyForMetrics) {
        hostedKeyMetrics.recordFailed({ ...hostedKeyForMetrics, reason: 'other' })
      }

      const strippedOutput = postProcessToolOutput(normalizedToolId, finalResult.output ?? {})

      return {
        ...finalResult,
        output: strippedOutput,
        timing: {
          startTime: startTimeISO,
          endTime: endTimeISO,
          duration,
        },
      }
    }

    // Execute the tool request directly (internal routes use regular fetch, external use SSRF-protected fetch)
    // Wrap with retry logic for hosted keys to handle rate limiting due to higher usage
    const result = hostedKeyInfo.isUsingHostedKey
      ? await executeWithRetry(
          () =>
            executeToolRequest(
              toolId,
              tool,
              contextParams,
              effectiveSignal,
              privateToolMetadataType,
              resolvedSecretTraceRegistry,
              internalSandboxProfile
            ),
          {
            requestId,
            toolId,
            provider: tool.hosting?.byokProviderId || tool.id,
            envVarName: hostedKeyInfo.envVarName!,
            executionContext,
            reacquireAfterRetriesExhausted: async () => {
              const reacquiredEnvVar = await reacquireHostedKey(
                tool,
                contextParams,
                executionContext,
                requestId
              )
              if (!reacquiredEnvVar) return null
              // Re-point metric labels at the freshly acquired key.
              hostedKeyInfo.envVarName = reacquiredEnvVar
              if (hostedKeyForMetrics) hostedKeyForMetrics.key = reacquiredEnvVar
              return () =>
                executeToolRequest(
                  toolId,
                  tool,
                  contextParams,
                  effectiveSignal,
                  privateToolMetadataType,
                  resolvedSecretTraceRegistry,
                  internalSandboxProfile
                )
            },
          }
        )
      : await executeToolRequest(
          toolId,
          tool,
          contextParams,
          effectiveSignal,
          privateToolMetadataType,
          resolvedSecretTraceRegistry,
          internalSandboxProfile
        )

    // Apply post-processing if available and not skipped
    let finalResult = result
    if (tool.postProcess && result.success && !skipPostProcess) {
      try {
        finalResult = await tool.postProcess(result, contextParams, executeNestedTool)
      } catch (error) {
        const normalizedError = toError(error)
        logger.error(
          `[${requestId}] Post-processing error for ${toolId}:`,
          projectToolLogMetadata(
            { error: normalizedError.message },
            resolvedSecretTraceRegistry,
            {
              errorName: normalizedError.name,
            },
            structuralOnlyToolLogs
          )
        )
        finalResult = result
      }
    }

    // Process file outputs if execution context is available
    finalResult = await processFileOutputs(finalResult, tool, executionContext)

    // Add timing data to the result
    const endTime = new Date()
    const endTimeISO = endTime.toISOString()
    const duration = endTime.getTime() - startTime.getTime()

    if (hostedKeyInfo.isUsingHostedKey && finalResult.success) {
      await applyHostedKeyCostToResult(
        finalResult,
        tool,
        contextParams,
        executionContext,
        requestId,
        hostedKeyInfo.envVarName
      )
    } else if (hostedKeyForMetrics) {
      hostedKeyMetrics.recordFailed({ ...hostedKeyForMetrics, reason: 'other' })
    }

    const strippedOutput = postProcessToolOutput(normalizedToolId, finalResult.output ?? {})

    return {
      ...finalResult,
      output: strippedOutput,
      timing: {
        startTime: startTimeISO,
        endTime: endTimeISO,
        duration,
      },
    }
  } catch (error: any) {
    const normalizedError = toError(error)
    logger.error(
      `[${requestId}] Error executing tool ${toolId}:`,
      projectToolLogMetadata(
        {
          error: normalizedError.message,
          stack: error instanceof Error ? error.stack : undefined,
        },
        resolvedSecretTraceRegistry,
        {
          errorName: normalizedError.name,
          hasStack: Boolean(error instanceof Error && error.stack),
        },
        structuralOnlyToolLogs
      )
    )

    if (hostedKeyForMetrics) {
      hostedKeyMetrics.recordFailed({
        ...hostedKeyForMetrics,
        reason: classifyHostedKeyFailure(error),
      })
    }

    // Default error handling
    let errorMessage = 'Unknown error occurred'
    let errorDetails = {}

    if (error instanceof Error) {
      errorMessage = error.message || `Error executing tool ${toolId}`
      // HTTP errors are thrown as Error instances carrying `status`/`statusText`/
      // `data` (see createTransformedErrorFromErrorInfo). Surface them on the
      // output so callers can branch on the status (e.g. treat 404 as a clean
      // no-match) — the object branch below only ran for non-Error throws.
      const httpStatus = (error as { status?: unknown }).status
      if (typeof httpStatus === 'number') {
        errorDetails = {
          status: httpStatus,
          statusText: (error as { statusText?: string }).statusText,
          data: (error as { data?: unknown }).data,
        }
      }
    } else if (typeof error === 'string') {
      errorMessage = error
    } else if (error && typeof error === 'object') {
      // Handle HTTP response errors
      if (error.status) {
        errorMessage = `HTTP ${error.status}: ${error.statusText || 'Request failed'}`

        if (error.data) {
          if (typeof error.data === 'string') {
            errorMessage = `${errorMessage} - ${error.data}`
          } else if (error.data.message) {
            errorMessage = `${errorMessage} - ${error.data.message}`
          } else if (error.data.error) {
            errorMessage = `${errorMessage} - ${
              typeof error.data.error === 'string'
                ? error.data.error
                : JSON.stringify(error.data.error)
            }`
          }
        }

        errorDetails = {
          status: error.status,
          statusText: error.statusText,
          data: error.data,
        }
      }
      // Handle other errors with messages
      else if (error.message) {
        // Don't pass along "undefined (undefined)" messages
        if (error.message === 'undefined (undefined)') {
          errorMessage = `Error executing tool ${toolId}`
          // Add status if available
          if (error.status) {
            errorMessage += ` (Status: ${error.status})`
          }
        } else {
          errorMessage = error.message
        }

        if ((error as any).cause) {
          errorMessage = `${errorMessage} (${(error as any).cause})`
        }
      }
    }

    // Add timing data even for errors
    const endTime = new Date()
    const endTimeISO = endTime.toISOString()
    const duration = endTime.getTime() - startTime.getTime()
    return {
      success: false,
      output: errorDetails,
      error: errorMessage,
      // Sim's own status (hosted-key 429/503) survives the flattening from a
      // thrown error into a result object; an upstream provider's status stays
      // on `output` where it cannot be mistaken for ours.
      ...(error instanceof HttpError ? { statusCode: error.statusCode } : {}),
      timing: {
        startTime: startTimeISO,
        endTime: endTimeISO,
        duration,
      },
    }
  } finally {
    completePendingSecretActivation?.()
  }
}

/**
 * Determines if a response or result represents an error condition
 */
function isErrorResponse(
  response: Response | any,
  data?: any
): { isError: boolean; errorInfo?: { status?: number; statusText?: string; data?: any } } {
  // HTTP Response object
  if (response && typeof response === 'object' && 'ok' in response) {
    if (!response.ok) {
      return {
        isError: true,
        errorInfo: {
          status: response.status,
          statusText: response.statusText,
          data: data,
        },
      }
    }
    return { isError: false }
  }

  // ToolResponse object
  if (response && typeof response === 'object' && 'success' in response) {
    return {
      isError: !response.success,
      errorInfo: response.success ? undefined : { data: response },
    }
  }

  // Check for error indicators in data
  if (data && typeof data === 'object') {
    if (data.error || data.success === false) {
      return {
        isError: true,
        errorInfo: { data: data },
      }
    }
  }

  return { isError: false }
}

/**
 * Checks whether a fully resolved URL points back to this Sim instance.
 * Used to propagate cycle-detection headers on API blocks that target
 * the platform's own workflow execution endpoints via absolute URL.
 */
function isSelfOriginUrl(url: string): boolean {
  return isSameOrigin(url, getBaseUrl()) || isSameOrigin(url, getInternalApiBaseUrl())
}

/**
 * Add internal authentication token to headers if running on server
 * @param headers - Headers object to modify
 * @param isInternalRoute - Whether the target URL is an internal route
 * @param requestId - Request ID for logging
 * @param context - Context string for logging (e.g., toolId or 'proxy')
 */
async function addInternalAuthIfNeeded(
  headers: Headers | Record<string, string>,
  isInternalRoute: boolean,
  requestId: string,
  context: string,
  userId?: string,
  claims?: InternalTokenClaims
): Promise<void> {
  if (typeof window === 'undefined') {
    if (isInternalRoute) {
      try {
        const internalToken = claims
          ? await generateInternalToken(userId, claims)
          : await generateInternalToken(userId)
        if (headers instanceof Headers) {
          headers.set('Authorization', `Bearer ${internalToken}`)
        } else {
          headers.Authorization = `Bearer ${internalToken}`
        }
        logger.info(`[${requestId}] Added internal auth token for ${context}`)
      } catch (error) {
        logger.error(`[${requestId}] Failed to generate internal token for ${context}:`, error)
      }
    } else {
      logger.info(`[${requestId}] Skipping internal auth token for external URL: ${context}`)
    }
  }
}

interface ResolvedRetryConfig {
  maxRetries: number
  initialDelayMs: number
  maxDelayMs: number
}

function getRetryConfig(
  retry: ToolRetryConfig | undefined,
  params: Record<string, any>,
  method: string
): ResolvedRetryConfig | null {
  if (!retry?.enabled) return null

  const isIdempotent = ['GET', 'HEAD', 'PUT', 'DELETE'].includes(method.toUpperCase())
  if (retry.retryIdempotentOnly && !isIdempotent && !params.retryNonIdempotent) {
    return null
  }

  const maxRetries = Math.min(10, Math.max(0, Number(params.retries) || retry.maxRetries || 0))
  if (maxRetries === 0) return null

  return {
    maxRetries,
    initialDelayMs: Number(params.retryDelayMs) || retry.initialDelayMs || 500,
    maxDelayMs: Number(params.retryMaxDelayMs) || retry.maxDelayMs || 30000,
  }
}

function isRetryableFailure(error: unknown, status?: number): boolean {
  if (status === 429 || (status && status >= 500 && status <= 599)) return true
  if (error instanceof Error) {
    const code = (error as NodeJS.ErrnoException).code
    if (code === 'ETIMEDOUT' || code === 'ECONNRESET' || code === 'ECONNABORTED') {
      return true
    }
    const msg = error.message.toLowerCase()
    if (isBodySizeLimitError(msg)) return false
    return msg.includes('timeout') || msg.includes('timed out')
  }
  return false
}

function shouldRetryWithoutReadingBody(
  status: number,
  headers: { get(name: string): string | null },
  retryConfig: ResolvedRetryConfig | null | undefined,
  isLastAttempt: boolean
): boolean {
  if (!retryConfig || isLastAttempt || !isRetryableFailure(null, status)) {
    return false
  }
  return (
    (parseRetryAfter(headers.get('retry-after'), Number.POSITIVE_INFINITY) ?? 0) <=
    retryConfig.maxDelayMs
  )
}

/**
 * Execute a tool request directly
 * Internal routes (/api/...) use regular fetch
 * External URLs use SSRF-protected fetch with DNS validation and IP pinning
 */
async function executeToolRequest(
  toolId: string,
  tool: ToolConfig,
  params: Record<string, any>,
  signal?: AbortSignal,
  privateToolMetadataType?: PrivateToolMetadataType,
  resolvedSecretTraceRegistry?: ResolvedSecretTraceRegistry,
  internalSandboxProfile?: InternalSandboxProfile
): Promise<ToolResponse> {
  const requestId = generateRequestId()
  const structuralOnlyToolLogs =
    normalizeToolId(toolId) === 'function_execute' ||
    isCustomTool(toolId) ||
    privateToolMetadataType !== undefined
  try {
    const requestParams = prepareToolRequest(tool, params, resolvedSecretTraceRegistry)
    const endpointUrl = requestParams.url
    const { headers, isInternalRoute } = requestParams
    const baseUrl = isInternalRoute ? getInternalApiBaseUrl() : getBaseUrl()

    const fullUrlObj = new URL(endpointUrl, baseUrl)

    if (isInternalRoute) {
      const workflowId = params._context?.workflowId
      if (workflowId) {
        fullUrlObj.searchParams.set('workflowId', workflowId)
      }
      const userId = params._context?.userId
      if (userId) {
        fullUrlObj.searchParams.set('userId', userId)
      }
    }

    const fullUrl = fullUrlObj.toString()

    if (isCustomTool(toolId) && requestParams.body) {
      let requestBody: unknown
      try {
        requestBody = JSON.parse(requestParams.body)
      } catch {
        requestBody = undefined
      }
      if (
        typeof requestBody === 'object' &&
        requestBody !== null &&
        'schema' in requestBody &&
        'params' in requestBody
      ) {
        try {
          validateClientSideParams(
            requestBody.params as Record<string, any>,
            requestBody.schema as {
              type: string
              properties: Record<string, any>
              required?: string[]
            }
          )
        } catch (validationError) {
          logger.error(`[${requestId}] Custom tool validation failed for ${toolId}:`, {
            error: toError(validationError).message,
          })
          throw validationError
        }
      }
    }
    await addInternalAuthIfNeeded(
      headers,
      isInternalRoute,
      requestId,
      toolId,
      params._context?.userId,
      internalSandboxProfile ? { sandboxProfile: internalSandboxProfile } : undefined
    )
    if (isInternalRoute && params._context?.billingAttribution) {
      headers.set(
        BILLING_ATTRIBUTION_HEADER,
        serializeBillingAttributionHeader(params._context.billingAttribution)
      )
    }
    if (isInternalRoute && privateToolMetadataType) {
      headers.set(PRIVATE_TOOL_METADATA_REQUEST_HEADER, privateToolMetadataType)
    }
    if (isInternalRoute) {
      const executionDeadline = serializeExecutionDeadlineHeader(signal)
      if (executionDeadline) {
        headers.set(INTERNAL_EXECUTION_DEADLINE_HEADER, executionDeadline)
      }
    }

    const shouldPropagateCallChain = isInternalRoute || isSelfOriginUrl(fullUrl)
    if (shouldPropagateCallChain) {
      const callChain = params._context?.callChain as string[] | undefined
      if (callChain && callChain.length > 0) {
        headers.set(SIM_VIA_HEADER, serializeCallChain(callChain))
      }
    }

    // Check request body size before sending to detect potential size limit issues
    validateRequestBodySize(requestParams.body, requestId, toolId)

    // Convert Headers to plain object for secureFetchWithPinnedIP
    const headersRecord: Record<string, string> = {}
    headers.forEach((value, key) => {
      headersRecord[key] = value
    })

    const retryConfig = getRetryConfig(tool.request.retry, params, requestParams.method)
    const maxAttempts = retryConfig ? 1 + retryConfig.maxRetries : 1

    let response: Response | undefined
    let lastError: unknown
    const nullBodyStatuses = new Set([101, 204, 205, 304])

    for (let attempt = 0; attempt < maxAttempts; attempt++) {
      const isLastAttempt = attempt === maxAttempts - 1

      try {
        if (isInternalRoute) {
          const controller = new AbortController()
          let didTimeout = false
          // With a caller/execution abort signal present, the plan-based timeout bounds the call and
          // this only acts as a ceiling; without one, keep the tighter default as the hang safety net.
          const timeout =
            requestParams.timeout ||
            (signal ? getMaxExecutionTimeout() : DEFAULT_EXECUTION_TIMEOUT_MS)
          const timeoutId = setTimeout(() => {
            didTimeout = true
            controller.abort(new DOMException('timeout', 'AbortError'))
          }, timeout)

          let abortListener: (() => void) | null = null
          if (signal) {
            if (signal.aborted) {
              controller.abort(signal.reason)
            } else {
              abortListener = () => controller.abort(signal.reason)
              signal.addEventListener('abort', abortListener, { once: true })
            }
          }

          try {
            const internalResponse = await fetch(fullUrl, {
              method: requestParams.method,
              headers: headers,
              body: requestParams.body,
              signal: controller.signal,
            })
            if (
              nullBodyStatuses.has(internalResponse.status) ||
              shouldRetryWithoutReadingBody(
                internalResponse.status,
                internalResponse.headers,
                retryConfig,
                isLastAttempt
              )
            ) {
              internalResponse.body?.cancel().catch(() => {})
              response = new Response(null, {
                status: internalResponse.status,
                statusText: internalResponse.statusText,
                headers: cloneResponseHeaders(internalResponse.headers),
              })
            } else {
              const bodyBuffer = await readToolResponseBody(internalResponse, {
                requestId,
                toolId,
                signal: controller.signal,
                ...(privateToolMetadataType
                  ? {
                      maxBytes:
                        MAX_TOOL_RESPONSE_BODY_BYTES + MAX_PRIVATE_TOOL_METADATA_OVERHEAD_BYTES,
                    }
                  : {}),
              })
              response = new Response(new Uint8Array(bodyBuffer), {
                status: internalResponse.status,
                statusText: internalResponse.statusText,
                headers: cloneResponseHeaders(internalResponse.headers),
              })
            }
          } catch (error) {
            if (
              controller.signal.aborted ||
              (error instanceof Error && error.name === 'AbortError')
            ) {
              // Distinguish caller cancellation from local timeout: rethrow the AbortError
              // when the caller's signal triggered the abort so cancellation propagates as-is.
              if (signal?.aborted && !didTimeout) {
                throw signal.reason ?? error
              }
              throw new Error(`Request timed out after ${timeout}ms`)
            }
            throw error
          } finally {
            clearTimeout(timeoutId)
            if (abortListener) {
              signal?.removeEventListener('abort', abortListener)
            }
          }
        } else {
          const urlValidation = await validateUrlWithDNS(fullUrl, 'toolUrl')
          if (!urlValidation.isValid) {
            throw new Error(`Invalid tool URL: ${urlValidation.error}`)
          }

          let proxyOption: string | undefined
          if (requestParams.proxyUrl) {
            const proxyValidation = await validateAndPinProxyUrl(requestParams.proxyUrl)
            if (!proxyValidation.isValid) {
              throw new Error(`Invalid proxy URL: ${proxyValidation.error}`)
            }
            proxyOption = proxyValidation.pinnedProxyUrl
          }

          const secureResponse = await secureFetchWithPinnedIP(fullUrl, urlValidation.resolvedIP!, {
            method: requestParams.method,
            headers: headersRecord,
            body: requestParams.body ?? undefined,
            timeout: requestParams.timeout,
            maxResponseBytes: MAX_TOOL_RESPONSE_BODY_BYTES,
            signal,
            proxyUrl: proxyOption,
            stripAuthOnRedirect: requestParams.stripAuthOnRedirect,
          })

          const responseHeaders = new Headers(secureResponse.headers.toRecord())

          if (
            nullBodyStatuses.has(secureResponse.status) ||
            shouldRetryWithoutReadingBody(
              secureResponse.status,
              responseHeaders,
              retryConfig,
              isLastAttempt
            )
          ) {
            secureResponse.body?.cancel().catch(() => {})
            response = new Response(null, {
              status: secureResponse.status,
              statusText: secureResponse.statusText,
              headers: responseHeaders,
            })
          } else {
            const bodyBuffer = await readToolResponseBody(secureResponse, {
              requestId,
              toolId,
              signal,
            })
            response = new Response(new Uint8Array(bodyBuffer), {
              status: secureResponse.status,
              statusText: secureResponse.statusText,
              headers: responseHeaders,
            })
          }
        }
      } catch (error) {
        lastError = error
        if (signal?.aborted || !retryConfig || isLastAttempt || !isRetryableFailure(error)) {
          throw error
        }
        const delayMs = backoffWithJitter(attempt + 1, null, {
          baseMs: retryConfig.initialDelayMs,
          maxMs: retryConfig.maxDelayMs,
        })
        logger.warn(
          `[${requestId}] Retrying ${toolId} after error (attempt ${attempt + 1}/${maxAttempts})`,
          { delayMs }
        )
        await sleep(delayMs)
        continue
      }

      if (
        retryConfig &&
        !isLastAttempt &&
        response &&
        !response.ok &&
        isRetryableFailure(null, response.status)
      ) {
        const retryAfterMs = parseRetryAfter(
          response.headers.get('retry-after'),
          Number.POSITIVE_INFINITY
        )
        if (retryAfterMs !== null && retryAfterMs > retryConfig.maxDelayMs) {
          logger.warn(
            `[${requestId}] Retry-After (${retryAfterMs}ms) exceeds maxDelayMs (${retryConfig.maxDelayMs}ms), skipping retry`
          )
          break
        }
        try {
          await response.arrayBuffer()
        } catch {
          // Ignore errors when consuming body
        }
        const delayMs = backoffWithJitter(attempt + 1, retryAfterMs, {
          baseMs: retryConfig.initialDelayMs,
          maxMs: retryConfig.maxDelayMs,
        })
        logger.warn(
          `[${requestId}] Retrying ${toolId} after HTTP ${response.status} (attempt ${attempt + 1}/${maxAttempts})`,
          { delayMs }
        )
        await sleep(delayMs)
        continue
      }

      break
    }

    if (!response) {
      throw lastError ?? new Error(`Request failed for ${toolId}`)
    }

    const privateMetadata = await consumePrivateToolResponseMetadata(
      response,
      privateToolMetadataType,
      params,
      resolvedSecretTraceRegistry
    )
    response = privateMetadata.response

    if (privateToolMetadataType) {
      const functionalBody = await readToolResponseBody(response, {
        requestId,
        toolId,
        signal,
      })
      const body =
        response.status === 204 || response.status === 205 || response.status === 304
          ? null
          : new Uint8Array(functionalBody)
      response = new Response(body, {
        status: response.status,
        statusText: response.statusText,
        headers: cloneResponseHeaders(response.headers),
      })
    }

    if (!response.ok) {
      let errorData: any
      try {
        const errorText = await response.text()
        try {
          errorData = JSON.parse(errorText)
        } catch {
          errorData = errorText
        }
      } catch {
        logger.error(`[${requestId}] Failed to read response body for ${toolId}`)
        errorData = null
      }

      const errorInfo: ErrorInfo = {
        status: response.status,
        statusText: response.statusText,
        data: errorData,
      }

      const errorToTransform = createTransformedErrorFromErrorInfo(errorInfo, tool.errorExtractor)
      const hasStructuredErrorPayload =
        errorData !== null &&
        typeof errorData === 'object' &&
        !Array.isArray(errorData) &&
        ('error' in errorData || 'message' in errorData)

      if (response.status === 413 && !hasStructuredErrorPayload) {
        logger.error(
          `[${requestId}] Request body too large for ${toolId} (HTTP 413):`,
          projectToolLogMetadata(
            {
              status: response.status,
              statusText: response.statusText,
              errorData,
            },
            resolvedSecretTraceRegistry,
            {
              status: response.status,
              statusText: response.statusText,
              hasErrorData: errorData !== null,
            },
            structuralOnlyToolLogs
          )
        )
        throw new Error(BODY_SIZE_LIMIT_ERROR_MESSAGE)
      }

      logger.error(
        `[${requestId}] Internal API error for ${toolId}:`,
        projectToolLogMetadata(
          {
            status: errorInfo.status,
            errorData: errorInfo.data,
          },
          resolvedSecretTraceRegistry,
          {
            status: errorInfo.status,
            hasErrorData: errorInfo.data !== null,
          },
          structuralOnlyToolLogs
        )
      )

      throw errorToTransform
    }

    let responseData
    const status = response.status
    if (status === 202 || status === 204 || status === 205) {
      responseData = { status }
    } else {
      if (tool.transformResponse) {
        responseData = null
      } else {
        try {
          responseData = await response.json()
        } catch (jsonError) {
          const normalizedError = toError(jsonError)
          logger.error(
            `[${requestId}] JSON parse error for ${toolId}:`,
            projectToolLogMetadata(
              { error: normalizedError.message },
              resolvedSecretTraceRegistry,
              { errorName: normalizedError.name },
              structuralOnlyToolLogs
            )
          )
          throw new Error(`Failed to parse response from ${toolId}: ${jsonError}`)
        }
      }
    }

    // Check for error conditions
    const { isError, errorInfo } = isErrorResponse(response, responseData)

    if (isError) {
      // Handle error case
      const errorToTransform = createTransformedErrorFromErrorInfo(errorInfo, tool.errorExtractor)

      logger.error(
        `[${requestId}] Internal API error for ${toolId}:`,
        projectToolLogMetadata(
          {
            status: errorInfo?.status,
            errorData: errorInfo?.data,
          },
          resolvedSecretTraceRegistry,
          {
            status: errorInfo?.status,
            hasErrorData: errorInfo?.data !== null && errorInfo?.data !== undefined,
          },
          structuralOnlyToolLogs
        )
      )

      throw errorToTransform
    }

    // Success case: use transformResponse if available
    if (tool.transformResponse) {
      try {
        // Forward the real body stream. Some transformResponse helpers (e.g. TikTok)
        // read via readResponseTextWithLimit, which requires `.body` (or Content-Length)
        // and otherwise mis-reports a false "response exceeded maximum size" error.
        const mockResponse = {
          ok: response.ok,
          status: response.status,
          statusText: response.statusText,
          headers: response.headers,
          url: fullUrl,
          body: response.body,
          json: () => response.json(),
          text: () => response.text(),
          arrayBuffer: () => response.arrayBuffer(),
          blob: () => response.blob(),
        } as Response

        const data = await tool.transformResponse(mockResponse, params)
        return data
      } catch (transformError) {
        const normalizedError = toError(transformError)
        logger.error(
          `[${requestId}] Transform response error for ${toolId}:`,
          projectToolLogMetadata(
            { error: normalizedError.message },
            resolvedSecretTraceRegistry,
            {
              errorName: normalizedError.name,
            },
            structuralOnlyToolLogs
          )
        )
        throw transformError
      }
    }

    // Default success response handling
    return {
      success: true,
      output: responseData.output || responseData,
      error: undefined,
    }
  } catch (error: any) {
    handleResponseSizeLimitError(error, requestId, toolId)

    // Check if this is a body size limit error and throw user-friendly message
    handleBodySizeLimitError(
      error,
      requestId,
      toolId,
      resolvedSecretTraceRegistry,
      structuralOnlyToolLogs
    )

    const normalizedError = toError(error)
    logger.error(
      `[${requestId}] Internal request error for ${toolId}:`,
      projectToolLogMetadata(
        { error: normalizedError.message },
        resolvedSecretTraceRegistry,
        {
          errorName: normalizedError.name,
        },
        structuralOnlyToolLogs
      )
    )

    // Let the error bubble up to be handled in the main executeTool function
    throw error
  }
}

/**
 * Validates parameters on the client side before sending to the execute endpoint
 */
function validateClientSideParams(
  params: Record<string, any>,
  schema: {
    type: string
    properties: Record<string, any>
    required?: string[]
  }
) {
  if (!schema || schema.type !== 'object') {
    throw new Error('Invalid schema format')
  }

  // Internal parameters that should be excluded from validation
  const internalParamSet = new Set([
    '_context',
    '_toolSchema',
    'workflowId',
    'envVars',
    'workflowVariables',
    'blockData',
    'blockNameMapping',
  ])

  // Check required parameters
  if (schema.required) {
    for (const requiredParam of schema.required) {
      if (!(requiredParam in params)) {
        throw new Error(`Required parameter missing: ${requiredParam}`)
      }
    }
  }

  // Check parameter types (basic validation)
  for (const [paramName, paramValue] of Object.entries(params)) {
    // Skip validation for internal parameters
    if (internalParamSet.has(paramName)) {
      continue
    }

    const paramSchema = schema.properties[paramName]
    if (!paramSchema) {
      throw new Error(`Unknown parameter: ${paramName}`)
    }

    // Basic type checking
    const type = paramSchema.type
    if (type === 'string' && typeof paramValue !== 'string') {
      throw new Error(`Parameter ${paramName} should be a string`)
    }
    if (type === 'number' && typeof paramValue !== 'number') {
      throw new Error(`Parameter ${paramName} should be a number`)
    }
    if (type === 'boolean' && typeof paramValue !== 'boolean') {
      throw new Error(`Parameter ${paramName} should be a boolean`)
    }
    if (type === 'array' && !Array.isArray(paramValue)) {
      throw new Error(`Parameter ${paramName} should be an array`)
    }
    if (type === 'object' && (typeof paramValue !== 'object' || paramValue === null)) {
      throw new Error(`Parameter ${paramName} should be an object`)
    }
  }
}

/**
 * Execute an MCP tool via the server-side MCP endpoint
 *
 * @param toolId - MCP tool ID in format "mcp-serverId-toolName"
 * @param params - Tool parameters
 * @param executionContext - Execution context
 * @param requestId - Request ID for logging
 * @param startTimeISO - Start time for timing
 */
async function executeMcpTool(
  toolId: string,
  params: Record<string, any>,
  executionContext?: ExecutionContext,
  requestId?: string,
  startTimeISO?: string,
  signal?: AbortSignal,
  resolvedSecretTraceRegistry?: ResolvedSecretTraceRegistry,
  privateToolMetadataType?: PrivateToolMetadataType
): Promise<ToolResponse> {
  const actualRequestId = requestId || generateRequestId()
  const actualStartTime = startTimeISO || new Date().toISOString()

  try {
    logger.info(`[${actualRequestId}] Executing MCP tool: ${toolId}`)

    const { serverId, toolName } = parseMcpToolId(toolId)

    const baseUrl = getInternalApiBaseUrl()

    const mcpScope = resolveToolScope(params, executionContext)

    const headers: Record<string, string> = { 'Content-Type': 'application/json' }
    if (privateToolMetadataType) {
      headers[PRIVATE_TOOL_METADATA_REQUEST_HEADER] = privateToolMetadataType
    }

    if (typeof window === 'undefined') {
      try {
        const internalToken = await generateInternalToken(mcpScope.userId)
        headers.Authorization = `Bearer ${internalToken}`
      } catch (error) {
        logger.error(`[${actualRequestId}] Failed to generate internal token:`, error)
      }
    }

    // Handle two different parameter structures:
    // 1. Direct MCP blocks: arguments are stored as JSON string in 'arguments' field
    // 2. Agent blocks: arguments are passed directly as top-level parameters
    let toolArguments = {}

    // First check if we have the 'arguments' field (direct MCP block usage)
    if (params.arguments) {
      if (typeof params.arguments === 'string') {
        try {
          toolArguments = JSON.parse(params.arguments)
        } catch (error) {
          logger.warn(`[${actualRequestId}] Failed to parse MCP arguments JSON`, {
            errorName: toError(error).name,
            argumentsType: 'string',
            argumentsLength: params.arguments.length,
          })
          toolArguments = {}
        }
      } else {
        toolArguments = params.arguments
      }
    } else {
      // Agent block usage: extract MCP-specific arguments by filtering out system parameters
      toolArguments = Object.fromEntries(
        Object.entries(params).filter(([key]) => !MCP_SYSTEM_PARAMETERS.has(key))
      )
    }

    if (mcpScope.callChain && mcpScope.callChain.length > 0) {
      headers[SIM_VIA_HEADER] = serializeCallChain(mcpScope.callChain)
    }
    if (mcpScope.billingAttribution) {
      headers[BILLING_ATTRIBUTION_HEADER] = serializeBillingAttributionHeader(
        mcpScope.billingAttribution
      )
    }
    const executionDeadline = serializeExecutionDeadlineHeader(signal)
    if (executionDeadline) {
      headers[INTERNAL_EXECUTION_DEADLINE_HEADER] = executionDeadline
    }

    if (!mcpScope.workspaceId) {
      return {
        success: false,
        output: {},
        error: `Missing workspaceId in execution context for MCP tool ${toolName}`,
        timing: {
          startTime: actualStartTime,
          endTime: new Date().toISOString(),
          duration: Date.now() - new Date(actualStartTime).getTime(),
        },
      }
    }
    if (!mcpScope.billingAttribution) {
      return {
        success: false,
        output: {},
        error: `Missing billing attribution in execution context for MCP tool ${toolName}`,
        timing: {
          startTime: actualStartTime,
          endTime: new Date().toISOString(),
          duration: Date.now() - new Date(actualStartTime).getTime(),
        },
      }
    }

    const requestBody: Record<string, any> = {
      serverId,
      toolName,
      arguments: toolArguments,
      workflowId: mcpScope.workflowId,
      workspaceId: mcpScope.workspaceId,
    }

    const body = JSON.stringify(requestBody)

    // Check request body size before sending
    validateRequestBodySize(body, actualRequestId, `mcp:${toolId}`)

    logger.info(`[${actualRequestId}] Making MCP tool request to ${toolName} on ${serverId}`, {
      hasWorkspaceId: !!mcpScope.workspaceId,
      hasWorkflowId: !!mcpScope.workflowId,
    })

    const mcpUrl = new URL('/api/mcp/tools/execute', baseUrl)
    if (mcpScope.userId) {
      mcpUrl.searchParams.set('userId', mcpScope.userId)
    }

    let response = await fetch(mcpUrl.toString(), {
      method: 'POST',
      headers,
      body,
      signal,
    })
    response = (
      await consumePrivateToolResponseMetadata(
        response,
        privateToolMetadataType,
        params,
        resolvedSecretTraceRegistry
      )
    ).response

    const endTime = new Date()
    const endTimeISO = endTime.toISOString()
    const duration = endTime.getTime() - new Date(actualStartTime).getTime()

    if (!response.ok) {
      // Check for 413 (Entity Too Large) - body size limit exceeded
      if (response.status === 413) {
        logger.error(`[${actualRequestId}] Request body too large for mcp:${toolId} (HTTP 413)`)
        return {
          success: false,
          output: {},
          error: BODY_SIZE_LIMIT_ERROR_MESSAGE,
          timing: {
            startTime: actualStartTime,
            endTime: endTimeISO,
            duration,
          },
        }
      }

      let errorMessage = `MCP tool execution failed: ${response.status} ${response.statusText}`

      try {
        const errorData = await response.json()
        if (errorData.error) errorMessage = errorData.error
      } catch {
        // Failed to parse error response, use default message
      }

      return {
        success: false,
        output: {},
        error: errorMessage,
        timing: {
          startTime: actualStartTime,
          endTime: endTimeISO,
          duration,
        },
      }
    }

    const result = await response.json()
    if (!result.success) {
      return {
        success: false,
        output: {},
        error: result.error || 'MCP tool execution failed',
        timing: {
          startTime: actualStartTime,
          endTime: endTimeISO,
          duration,
        },
      }
    }

    logger.info(`[${actualRequestId}] MCP tool ${toolId} executed successfully`)

    return {
      success: true,
      output: result.data?.output || result.output || result.data || {},
      timing: {
        startTime: actualStartTime,
        endTime: endTimeISO,
        duration,
      },
    }
  } catch (error) {
    const endTime = new Date()
    const endTimeISO = endTime.toISOString()
    const duration = endTime.getTime() - new Date(actualStartTime).getTime()

    // Check if this is a body size limit error
    const errorMsg = toError(error).message
    if (isBodySizeLimitError(errorMsg)) {
      logger.error(
        `[${actualRequestId}] Request body size limit exceeded for mcp:${toolId}:`,
        projectToolLogMetadata({ originalError: errorMsg }, resolvedSecretTraceRegistry, {
          hasOriginalError: errorMsg.length > 0,
        })
      )
      return {
        success: false,
        output: {},
        error: BODY_SIZE_LIMIT_ERROR_MESSAGE,
        timing: {
          startTime: actualStartTime,
          endTime: endTimeISO,
          duration,
        },
      }
    }

    const normalizedError = toError(error)
    logger.error(
      `[${actualRequestId}] Error executing MCP tool ${toolId}:`,
      projectToolLogMetadata(
        {
          error: normalizedError.message,
          stack: error instanceof Error ? error.stack : undefined,
        },
        resolvedSecretTraceRegistry,
        {
          errorName: normalizedError.name,
          hasStack: Boolean(error instanceof Error && error.stack),
        }
      )
    )

    const errorMessage = getErrorMessage(error, `Failed to execute MCP tool ${toolId}`)

    return {
      success: false,
      output: {},
      error: errorMessage,
      timing: {
        startTime: actualStartTime,
        endTime: endTimeISO,
        duration,
      },
    }
  }
}