diff --git a/apps/sim/app/api/guardrails/validate/route.test.ts b/apps/sim/app/api/guardrails/validate/route.test.ts index cdb5977536a..3f6963a9f9d 100644 --- a/apps/sim/app/api/guardrails/validate/route.test.ts +++ b/apps/sim/app/api/guardrails/validate/route.test.ts @@ -435,4 +435,31 @@ describe('POST /api/guardrails/validate', () => { await expect(res.json()).resolves.toEqual({ error: 'Failed to resolve billing attribution' }) expect(mockValidateHallucination).not.toHaveBeenCalled() }) + + /** + * The signal now reaches the scoring model, so cancellation is reachable here. + * `passed: false` would read to a consumer as the guardrail rejecting the content, + * blocking a run that was abandoned rather than judged. + */ + it('reports a cancelled run as cancellation rather than a failed guardrail', async () => { + mockAuthorizeCredentialUse.mockResolvedValue({ ok: true }) + mockValidateHallucination.mockRejectedValueOnce( + Object.assign(new Error('The operation was aborted.'), { name: 'AbortError' }) + ) + + const res = await POST( + createMockRequest('POST', { + validationType: 'hallucination', + input: 'test input', + knowledgeBaseId: 'kb-1', + model: 'openai/gpt-4o', + workflowId: 'wf-1', + }) + ) + + expect(res.status).toBe(499) + const json = await res.json() + expect(json.success).toBe(false) + expect(json.output?.passed).toBeUndefined() + }) }) diff --git a/apps/sim/app/api/guardrails/validate/route.ts b/apps/sim/app/api/guardrails/validate/route.ts index be97ae0e52e..9a3f36df36b 100644 --- a/apps/sim/app/api/guardrails/validate/route.ts +++ b/apps/sim/app/api/guardrails/validate/route.ts @@ -28,6 +28,7 @@ import { ProviderNotAllowedError, } from '@/ee/access-control/utils/permission-check' import type { ResolvedSecretTraceRegistry } from '@/executor/utils/resolved-secret-trace-registry' +import { isAbortError } from '@/providers/streaming-tool-loop-shared' import { getProviderFromModel } from '@/providers/utils' const logger = createLogger('GuardrailsValidateAPI') @@ -316,7 +317,8 @@ export const POST = withRouteHandler(async (request: NextRequest) => { auth.userId, billingAttribution, requestId, - resolvedSecretTraceRegistry + resolvedSecretTraceRegistry, + request.signal ) /** @@ -371,6 +373,19 @@ export const POST = withRouteHandler(async (request: NextRequest) => { }, }) } catch (error: any) { + /** + * A cancelled run must not be reshaped into a verdict. `passed: false` here reads + * to a consumer as the guardrail rejecting the content, so an abandoned run would + * block content that was never actually judged. 499 matches the convention the + * workflow execute route already uses for a client-cancelled request. + */ + if (isAbortError(error)) { + logger.info(`[${requestId}] Guardrails validation cancelled by client`) + return NextResponse.json( + { success: false, error: 'Client cancelled request' }, + { status: 499 } + ) + } logger.error(`[${requestId}] Guardrails validation failed`, { error }) return NextResponse.json({ success: true, @@ -431,7 +446,8 @@ async function executeValidation( actorUserId: string, billingAttribution: BillingAttributionSnapshot | undefined, requestId: string, - resolvedSecretTraceRegistry: ResolvedSecretTraceRegistry | undefined + resolvedSecretTraceRegistry: ResolvedSecretTraceRegistry | undefined, + abortSignal: AbortSignal | undefined ): Promise<{ passed: boolean error?: string @@ -488,6 +504,7 @@ async function executeValidation( billingAttribution, requestId, resolvedSecretTraceRegistry, + abortSignal, }) } if (validationType === 'pii') { diff --git a/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/panel/components/editor/components/sub-block/components/text/text.tsx b/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/panel/components/editor/components/sub-block/components/text/text.tsx index cd7cd6b780d..85f045f84be 100644 --- a/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/panel/components/editor/components/sub-block/components/text/text.tsx +++ b/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/panel/components/editor/components/sub-block/components/text/text.tsx @@ -33,7 +33,7 @@ export function Text({ blockId, subBlockId, content, className }: TextProps) { className={`rounded-md border bg-[var(--surface-2)] p-4 shadow-sm ${className || ''}`} >
diff --git a/apps/sim/lib/guardrails/validate_hallucination.test.ts b/apps/sim/lib/guardrails/validate_hallucination.test.ts index db42ef6ca8f..e522db31b2e 100644 --- a/apps/sim/lib/guardrails/validate_hallucination.test.ts +++ b/apps/sim/lib/guardrails/validate_hallucination.test.ts @@ -182,4 +182,25 @@ describe('validateHallucination', () => { expect(mockExecuteProviderRequest).not.toHaveBeenCalled() expect(registry.isComplete()).toBe(true) }) + + /** + * Forwarding the caller's signal means the scoring model can now be aborted. A + * cancelled run must not be reported as a guardrail verdict — `passed: false` would + * block content on a run the caller abandoned, which is indistinguishable to a + * consumer from the model actually hallucinating. + */ + it('surfaces a cancelled run as cancellation, not as a failed guardrail', async () => { + const registry = new ResolvedSecretTraceRegistry() + const fetchMock = vi.fn(async () => + createPrivateKnowledgeResponse({ data: { results: [{ content: 'reference' }] } }) + ) + vi.stubGlobal('fetch', fetchMock) + + const abort = Object.assign(new Error('The operation was aborted.'), { name: 'AbortError' }) + mockExecuteProviderRequest.mockRejectedValueOnce(abort) + + await expect(validateHallucination(createInput(registry))).rejects.toMatchObject({ + name: 'AbortError', + }) + }) }) diff --git a/apps/sim/lib/guardrails/validate_hallucination.ts b/apps/sim/lib/guardrails/validate_hallucination.ts index 3927003b490..d97cf94bf8a 100644 --- a/apps/sim/lib/guardrails/validate_hallucination.ts +++ b/apps/sim/lib/guardrails/validate_hallucination.ts @@ -24,6 +24,7 @@ import { refreshTokenIfNeeded } from '@/app/api/auth/oauth/utils' import { projectResolvedSecretModelContent } from '@/executor/utils/resolved-secret-content-projection' import type { ResolvedSecretTraceRegistry } from '@/executor/utils/resolved-secret-trace-registry' import { executeProviderRequest } from '@/providers' +import { isAbortError } from '@/providers/streaming-tool-loop-shared' import { getProviderFromModel } from '@/providers/utils' const logger = createLogger('HallucinationValidator') @@ -68,6 +69,12 @@ export interface HallucinationValidationInput { billingAttribution: BillingAttributionSnapshot requestId: string resolvedSecretTraceRegistry: ResolvedSecretTraceRegistry + /** + * The caller's cancellation signal, forwarded to the scoring model exactly as the + * agent handler forwards `ctx.abortSignal`. Without it the scoring request outlives + * a cancelled request and keeps burning a provider slot until the transport gives up. + */ + abortSignal?: AbortSignal } /** @@ -176,7 +183,8 @@ async function scoreHallucinationWithLLM( providerCredentials: HallucinationValidationInput['providerCredentials'], workspaceId: string | undefined, requestId: string, - resolvedSecretTraceRegistry: ResolvedSecretTraceRegistry + resolvedSecretTraceRegistry: ResolvedSecretTraceRegistry, + abortSignal: AbortSignal | undefined ): Promise<{ score: number; reasoning: string; cost: number }> { try { const contextText = ragContext.join('\n\n---\n\n') @@ -251,6 +259,7 @@ Evaluate the consistency and provide your score and reasoning in JSON format.` bedrockSecretKey: providerCredentials?.bedrockSecretKey, bedrockRegion: providerCredentials?.bedrockRegion, workspaceId, + abortSignal, }, { resolvedSecretTraceRegistry } ) @@ -290,6 +299,11 @@ Evaluate the consistency and provide your score and reasoning in JSON format.` cost, } } catch (error: any) { + /** + * A cancelled run is not a scoring failure. Rewrapping it would erase the + * `AbortError` name the outer handler classifies on, so it propagates as-is. + */ + if (isAbortError(error)) throw error logger.error(`[${requestId}] Error scoring with LLM`, { error: error.message, }) @@ -317,6 +331,7 @@ export async function validateHallucination( billingAttribution, requestId, resolvedSecretTraceRegistry, + abortSignal, } = input try { @@ -371,7 +386,8 @@ export async function validateHallucination( providerCredentials, workspaceId, requestId, - providerRegistry + providerRegistry, + abortSignal ) logger.info(`[${requestId}] Confidence score: ${score}`, { @@ -392,6 +408,12 @@ export async function validateHallucination( : `Low confidence: score ${score}/10 is below threshold ${threshold}`, } } catch (error: any) { + /** + * Cancellation is surfaced as cancellation, not as a guardrail verdict. Returning + * `passed: false` here would fail content on a run the caller abandoned, which is + * indistinguishable to a consumer from the model actually hallucinating. + */ + if (isAbortError(error)) throw error logger.error(`[${requestId}] Hallucination validation error`, { error: error.message, }) diff --git a/apps/sim/providers/baseten/index.ts b/apps/sim/providers/baseten/index.ts index f03050e735e..47a5d65a7d0 100644 --- a/apps/sim/providers/baseten/index.ts +++ b/apps/sim/providers/baseten/index.ts @@ -19,6 +19,7 @@ import { createStreamingExecution } from '@/providers/streaming-execution' import { isAbortError, parseToolArguments } from '@/providers/streaming-tool-loop-shared' import { adaptOpenAIChatToolSchema } from '@/providers/tool-schema-adapter' import { enrichLastModelSegmentFromChatCompletions } from '@/providers/trace-enrichment' +import { openAICompatTransport } from '@/providers/transport' import type { FunctionCallResponse, Message, @@ -86,6 +87,7 @@ export const basetenProvider: ProviderConfig = { } const client = new OpenAI({ + ...openAICompatTransport(), apiKey: request.apiKey, baseURL: 'https://inference.baseten.co/v1', }) diff --git a/apps/sim/providers/cerebras/index.ts b/apps/sim/providers/cerebras/index.ts index 641b9e6d9f6..f4e9085f3f7 100644 --- a/apps/sim/providers/cerebras/index.ts +++ b/apps/sim/providers/cerebras/index.ts @@ -15,6 +15,7 @@ import { createStreamingExecution } from '@/providers/streaming-execution' import { isAbortError, parseToolArguments } from '@/providers/streaming-tool-loop-shared' import { adaptOpenAIChatToolSchema } from '@/providers/tool-schema-adapter' import { enrichLastModelSegmentFromChatCompletions } from '@/providers/trace-enrichment' +import { openAICompatTransport } from '@/providers/transport' import type { ProviderConfig, ProviderRequest, @@ -54,6 +55,7 @@ export const cerebrasProvider: ProviderConfig = { try { const client = new Cerebras({ apiKey: request.apiKey, + ...openAICompatTransport(), }) const allMessages = [] diff --git a/apps/sim/providers/deepseek/index.ts b/apps/sim/providers/deepseek/index.ts index 6b269df7a78..4a348003669 100644 --- a/apps/sim/providers/deepseek/index.ts +++ b/apps/sim/providers/deepseek/index.ts @@ -13,6 +13,7 @@ import { createStreamingExecution } from '@/providers/streaming-execution' import { isAbortError, parseToolArguments } from '@/providers/streaming-tool-loop-shared' import { adaptOpenAIChatToolSchema } from '@/providers/tool-schema-adapter' import { enrichLastModelSegmentFromChatCompletions } from '@/providers/trace-enrichment' +import { openAICompatTransport } from '@/providers/transport' import type { ProviderConfig, ProviderRequest, @@ -50,6 +51,7 @@ export const deepseekProvider: ProviderConfig = { try { const deepseek = new OpenAI({ + ...openAICompatTransport(), apiKey: request.apiKey, baseURL: 'https://api.deepseek.com', }) diff --git a/apps/sim/providers/fireworks/index.ts b/apps/sim/providers/fireworks/index.ts index 37f7b3fce28..e19be0f2940 100644 --- a/apps/sim/providers/fireworks/index.ts +++ b/apps/sim/providers/fireworks/index.ts @@ -20,6 +20,7 @@ import { createStreamingExecution } from '@/providers/streaming-execution' import { isAbortError, parseToolArguments } from '@/providers/streaming-tool-loop-shared' import { adaptOpenAIChatToolSchema } from '@/providers/tool-schema-adapter' import { enrichLastModelSegmentFromChatCompletions } from '@/providers/trace-enrichment' +import { openAICompatTransport } from '@/providers/transport' import type { FunctionCallResponse, Message, @@ -87,6 +88,7 @@ export const fireworksProvider: ProviderConfig = { } const client = new OpenAI({ + ...openAICompatTransport(), apiKey: request.apiKey, baseURL: 'https://api.fireworks.ai/inference/v1', }) diff --git a/apps/sim/providers/groq/index.ts b/apps/sim/providers/groq/index.ts index 09158078c6f..e80e1f0a1a6 100644 --- a/apps/sim/providers/groq/index.ts +++ b/apps/sim/providers/groq/index.ts @@ -18,6 +18,7 @@ import { createStreamingExecution } from '@/providers/streaming-execution' import { isAbortError, parseToolArguments } from '@/providers/streaming-tool-loop-shared' import { adaptOpenAIChatToolSchema } from '@/providers/tool-schema-adapter' import { enrichLastModelSegmentFromChatCompletions } from '@/providers/trace-enrichment' +import { openAICompatTransport } from '@/providers/transport' import type { ProviderConfig, ProviderRequest, @@ -50,7 +51,7 @@ export const groqProvider: ProviderConfig = { throw new Error('API key is required for Groq') } - const groq = new Groq({ apiKey: request.apiKey }) + const groq = new Groq({ apiKey: request.apiKey, ...openAICompatTransport() }) const allMessages = [] diff --git a/apps/sim/providers/kimi/index.ts b/apps/sim/providers/kimi/index.ts index eb464830ed3..6e1d91e9d99 100644 --- a/apps/sim/providers/kimi/index.ts +++ b/apps/sim/providers/kimi/index.ts @@ -19,6 +19,7 @@ import { createStreamingExecution } from '@/providers/streaming-execution' import { isAbortError, parseToolArguments } from '@/providers/streaming-tool-loop-shared' import { adaptOpenAIChatToolSchema } from '@/providers/tool-schema-adapter' import { enrichLastModelSegmentFromChatCompletions } from '@/providers/trace-enrichment' +import { openAICompatTransport } from '@/providers/transport' import type { ProviderConfig, ProviderRequest, @@ -100,6 +101,7 @@ export const kimiProvider: ProviderConfig = { try { const kimi = new OpenAI({ + ...openAICompatTransport(), apiKey: request.apiKey, baseURL: KIMI_BASE_URL, }) diff --git a/apps/sim/providers/litellm/index.ts b/apps/sim/providers/litellm/index.ts index ff02c415fec..f3cd0d10b44 100644 --- a/apps/sim/providers/litellm/index.ts +++ b/apps/sim/providers/litellm/index.ts @@ -16,6 +16,7 @@ import { createStreamingExecution } from '@/providers/streaming-execution' import { isAbortError, parseToolArguments } from '@/providers/streaming-tool-loop-shared' import { adaptOpenAIChatToolSchema } from '@/providers/tool-schema-adapter' import { enrichLastModelSegmentFromChatCompletions } from '@/providers/trace-enrichment' +import { openAICompatTransport } from '@/providers/transport' import type { Message, ProviderConfig, @@ -110,6 +111,7 @@ export const litellmProvider: ProviderConfig = { const apiKey = request.apiKey || env.LITELLM_API_KEY || 'empty' const litellm = new OpenAI({ + ...openAICompatTransport(), apiKey, baseURL: `${baseUrl}/v1`, }) diff --git a/apps/sim/providers/meta/index.ts b/apps/sim/providers/meta/index.ts index e94b82ed997..dacce56641b 100644 --- a/apps/sim/providers/meta/index.ts +++ b/apps/sim/providers/meta/index.ts @@ -14,6 +14,7 @@ import { createStreamingExecution } from '@/providers/streaming-execution' import { isAbortError, parseToolArguments } from '@/providers/streaming-tool-loop-shared' import { adaptOpenAIChatToolSchema } from '@/providers/tool-schema-adapter' import { enrichLastModelSegmentFromChatCompletions } from '@/providers/trace-enrichment' +import { openAICompatTransport } from '@/providers/transport' import type { ProviderConfig, ProviderRequest, @@ -53,6 +54,7 @@ export const metaProvider: ProviderConfig = { try { const meta = new OpenAI({ + ...openAICompatTransport(), apiKey: request.apiKey, baseURL: META_BASE_URL, }) diff --git a/apps/sim/providers/mistral/index.ts b/apps/sim/providers/mistral/index.ts index 90d78f02d38..d63a6021d5e 100644 --- a/apps/sim/providers/mistral/index.ts +++ b/apps/sim/providers/mistral/index.ts @@ -14,6 +14,7 @@ import { createStreamingExecution } from '@/providers/streaming-execution' import { isAbortError, parseToolArguments } from '@/providers/streaming-tool-loop-shared' import { adaptOpenAIChatToolSchema } from '@/providers/tool-schema-adapter' import { enrichLastModelSegmentFromChatCompletions } from '@/providers/trace-enrichment' +import { openAICompatTransport } from '@/providers/transport' import type { ProviderConfig, ProviderRequest, @@ -61,6 +62,7 @@ export const mistralProvider: ProviderConfig = { } const mistral = new OpenAI({ + ...openAICompatTransport(), apiKey: request.apiKey, baseURL: 'https://api.mistral.ai/v1', }) diff --git a/apps/sim/providers/nvidia/index.ts b/apps/sim/providers/nvidia/index.ts index b2cda914a5e..f35d08042b1 100644 --- a/apps/sim/providers/nvidia/index.ts +++ b/apps/sim/providers/nvidia/index.ts @@ -15,6 +15,7 @@ import { createStreamingExecution } from '@/providers/streaming-execution' import { isAbortError, parseToolArguments } from '@/providers/streaming-tool-loop-shared' import { adaptOpenAIChatToolSchema } from '@/providers/tool-schema-adapter' import { enrichLastModelSegmentFromChatCompletions } from '@/providers/trace-enrichment' +import { openAICompatTransport } from '@/providers/transport' import type { ProviderConfig, ProviderRequest, @@ -60,6 +61,7 @@ export const nvidiaProvider: ProviderConfig = { try { const nvidia = new OpenAI({ + ...openAICompatTransport(), apiKey: request.apiKey, baseURL: NVIDIA_BASE_URL, }) diff --git a/apps/sim/providers/ollama-cloud/index.ts b/apps/sim/providers/ollama-cloud/index.ts index 0f782f7e24d..11d7bb4372b 100644 --- a/apps/sim/providers/ollama-cloud/index.ts +++ b/apps/sim/providers/ollama-cloud/index.ts @@ -4,6 +4,7 @@ import type { StreamingExecution } from '@/executor/types' import { getProviderDefaultModel, getProviderModels } from '@/providers/models' import { executeOllamaProviderRequest } from '@/providers/ollama/core' import { createReadableStreamFromOllamaCloudStream } from '@/providers/ollama-cloud/utils' +import { openAICompatTransport } from '@/providers/transport' import type { ProviderConfig, ProviderRequest, ProviderResponse } from '@/providers/types' const logger = createLogger('OllamaCloudProvider') @@ -36,6 +37,7 @@ export const ollamaCloudProvider: ProviderConfig = { providerLabel: 'Ollama Cloud', createClient: () => new OpenAI({ + ...openAICompatTransport(), apiKey, baseURL: OLLAMA_CLOUD_BASE_URL, }), diff --git a/apps/sim/providers/ollama/index.ts b/apps/sim/providers/ollama/index.ts index 755c11fdb36..cd0e16f8bd7 100644 --- a/apps/sim/providers/ollama/index.ts +++ b/apps/sim/providers/ollama/index.ts @@ -6,6 +6,7 @@ import type { StreamingExecution } from '@/executor/types' import { executeOllamaProviderRequest } from '@/providers/ollama/core' import type { ModelsObject } from '@/providers/ollama/types' import { createReadableStreamFromOllamaStream } from '@/providers/ollama/utils' +import { openAICompatTransport } from '@/providers/transport' import type { ProviderConfig, ProviderRequest, ProviderResponse } from '@/providers/types' import { useProvidersStore } from '@/stores/providers' @@ -52,6 +53,7 @@ export const ollamaProvider: ProviderConfig = { providerLabel: 'Ollama', createClient: () => new OpenAI({ + ...openAICompatTransport(), apiKey: 'empty', baseURL: `${OLLAMA_HOST}/v1`, }), diff --git a/apps/sim/providers/openrouter/index.ts b/apps/sim/providers/openrouter/index.ts index 1ec0369ef77..42173e83c55 100644 --- a/apps/sim/providers/openrouter/index.ts +++ b/apps/sim/providers/openrouter/index.ts @@ -29,6 +29,7 @@ import { createStreamingExecution } from '@/providers/streaming-execution' import { isAbortError, parseToolArguments } from '@/providers/streaming-tool-loop-shared' import { adaptOpenAIChatToolSchema } from '@/providers/tool-schema-adapter' import { enrichLastModelSegmentFromChatCompletions } from '@/providers/trace-enrichment' +import { openAICompatTransport } from '@/providers/transport' import type { FunctionCallResponse, Message, @@ -104,6 +105,7 @@ export const openRouterProvider: ProviderConfig = { } const client = new OpenAI({ + ...openAICompatTransport(), apiKey: request.apiKey, baseURL: 'https://openrouter.ai/api/v1', }) diff --git a/apps/sim/providers/sakana/index.ts b/apps/sim/providers/sakana/index.ts index 5548ac8dd3e..b85bbbfe964 100644 --- a/apps/sim/providers/sakana/index.ts +++ b/apps/sim/providers/sakana/index.ts @@ -14,6 +14,7 @@ import { createStreamingExecution } from '@/providers/streaming-execution' import { isAbortError, parseToolArguments } from '@/providers/streaming-tool-loop-shared' import { adaptOpenAIChatToolSchema } from '@/providers/tool-schema-adapter' import { enrichLastModelSegmentFromChatCompletions } from '@/providers/trace-enrichment' +import { openAICompatTransport } from '@/providers/transport' import type { ProviderConfig, ProviderRequest, @@ -54,6 +55,7 @@ export const sakanaProvider: ProviderConfig = { try { const sakana = new OpenAI({ + ...openAICompatTransport(), apiKey: request.apiKey, baseURL: SAKANA_BASE_URL, }) diff --git a/apps/sim/providers/together/index.ts b/apps/sim/providers/together/index.ts index 2725ae2d425..6c5718c0299 100644 --- a/apps/sim/providers/together/index.ts +++ b/apps/sim/providers/together/index.ts @@ -19,6 +19,7 @@ import { } from '@/providers/together/utils' import { adaptOpenAIChatToolSchema } from '@/providers/tool-schema-adapter' import { enrichLastModelSegmentFromChatCompletions } from '@/providers/trace-enrichment' +import { openAICompatTransport } from '@/providers/transport' import type { FunctionCallResponse, Message, @@ -86,6 +87,7 @@ export const togetherProvider: ProviderConfig = { } const client = new OpenAI({ + ...openAICompatTransport(), apiKey: request.apiKey, baseURL: 'https://api.together.ai/v1', }) diff --git a/apps/sim/providers/transport.test.ts b/apps/sim/providers/transport.test.ts new file mode 100644 index 00000000000..bd743b6423b --- /dev/null +++ b/apps/sim/providers/transport.test.ts @@ -0,0 +1,65 @@ +/** + * @vitest-environment node + * + * Pinned against the vendored SDKs rather than against numbers typed from memory: if an + * SDK bump moves a default, these fail and the divergence becomes a decision instead of + * a surprise. + */ +import Cerebras from '@cerebras/cerebras_cloud_sdk' +import Groq from 'groq-sdk' +import OpenAI from 'openai' +import { describe, expect, it } from 'vitest' +import { + openAICompatTransport, + PROVIDER_HEADERS_TIMEOUT_MS, + PROVIDER_MAX_RETRIES, +} from '@/providers/transport' + +describe('provider transport policy', () => { + it('pins the headers budget to the vendored OpenAI client default', () => { + expect(PROVIDER_HEADERS_TIMEOUT_MS).toBe(OpenAI.DEFAULT_TIMEOUT) + }) + + /** + * Not lowered to 0 in favour of a hand-rolled loop: a chat completion is + * non-idempotent and carries no idempotency key, and on the non-streaming path the + * response exists only once the generation is already billed, so a replay re-bills + * completed work. + */ + it('keeps the vendor retry default rather than hand-rolling one', () => { + expect(PROVIDER_MAX_RETRIES).toBe(OpenAI.DEFAULT_MAX_RETRIES ?? 2) + }) + + /** + * The assertion that matters: object spread gets no excess-property checking, so a + * renamed option in either SDK would become a silent no-op with a green typecheck. + * These read the value back off a constructed client. + */ + it('actually reaches the Groq client, which defaults to 60s', () => { + const client = new Groq({ apiKey: 'test', ...openAICompatTransport() }) + expect(client.timeout).toBe(PROVIDER_HEADERS_TIMEOUT_MS) + expect(client.maxRetries).toBe(PROVIDER_MAX_RETRIES) + }) + + it('actually reaches the Cerebras client, which defaults to 60s', () => { + const client = new Cerebras({ apiKey: 'test', ...openAICompatTransport() }) + expect(client.timeout).toBe(PROVIDER_HEADERS_TIMEOUT_MS) + expect(client.maxRetries).toBe(PROVIDER_MAX_RETRIES) + }) + + it('actually reaches an OpenAI-compatible client', () => { + const client = new OpenAI({ + apiKey: 'test', + baseURL: 'https://example.invalid', + ...openAICompatTransport(), + }) + expect(client.timeout).toBe(PROVIDER_HEADERS_TIMEOUT_MS) + expect(client.maxRetries).toBe(PROVIDER_MAX_RETRIES) + }) + + /** Stamping a `fetch` wrapper here would convert a header-only timer into a total + * deadline that truncates live streams, so the shape is pinned too. */ + it('stamps only constructor-safe options', () => { + expect(Object.keys(openAICompatTransport()).sort()).toEqual(['maxRetries', 'timeout']) + }) +}) diff --git a/apps/sim/providers/transport.ts b/apps/sim/providers/transport.ts new file mode 100644 index 00000000000..3755d3cf8ce --- /dev/null +++ b/apps/sim/providers/transport.ts @@ -0,0 +1,62 @@ +/** + * Transport policy for provider requests, in one place. + * + * These pin what the vendor SDKs already default to, so an SDK bump cannot silently + * move production behaviour. For 16 of 18 providers this is a no-op. Groq and Cerebras + * are the exception and are marked at {@link PROVIDER_HEADERS_TIMEOUT_MS}. + * + * What these deliberately do NOT do: bound a stalled stream. Bun's `fetch` is native + * and appears to impose a socket-scoped idle wall of roughly 300s, reduced by however + * long a pooled socket sat idle before reuse. That figure is observed, not documented, + * and is not something these constants can override — raising a number above it changes + * nothing. Only keeping bytes on the socket can rescue a long silent generation, and + * that work belongs in the stream pump. + */ + +/** + * Time-to-headers budget for a single attempt, matching `openai@7`'s own + * `DEFAULT_TIMEOUT`. + * + * Behaviour-preserving for the 16 providers already on that client. It is a deliberate + * divergence for **Groq and Cerebras**, whose SDKs default to 60s: on a non-streaming + * call headers do not arrive until the generation completes, so 60s caps every + * generation at a minute and then retries it twice, re-billing. The cost of the raise is + * that a genuinely hung call now fails slower — see the PR for the worst-case numbers. + * + * It must stay generous: `deepseek-reasoner`, `kimi-k3`, `grok-4.5-reasoning` and + * every dynamic-catalog provider can legitimately generate for minutes with zero + * bytes on the wire, and on a non-streaming call headers do not arrive until the + * generation completes. A tighter value converts today's successes into failures. + */ +export const PROVIDER_HEADERS_TIMEOUT_MS = 600_000 + +/** + * Vendor default, pinned rather than changed. + * + * Deliberately not lowered to 0 in favour of a hand-rolled loop: a chat completion + * is non-idempotent and carries no idempotency key, and on the non-streaming path + * the response only exists once the generation has already been billed — so a + * replay re-bills completed work, multiplied by every turn of the tool loop. + */ +export const PROVIDER_MAX_RETRIES = 2 + +export interface OpenAICompatTransport { + timeout: number + maxRetries: number +} + +/** + * Transport options for the OpenAI-compatible clients. + * + * Stamps `timeout` and `maxRetries` only. Both are process-wide constants, which is + * what makes them safe to set in a constructor that {@link getCachedProviderClient} + * memoises — anything varying per request must be passed per call instead. + * + * Deliberately stamps no `fetch`: a wrapper would turn the SDK's header-only timer + * into a total deadline that truncates a live stream, and would drop the caller's + * abort signal. + */ +export const openAICompatTransport = (): OpenAICompatTransport => ({ + timeout: PROVIDER_HEADERS_TIMEOUT_MS, + maxRetries: PROVIDER_MAX_RETRIES, +}) diff --git a/apps/sim/providers/vllm/index.ts b/apps/sim/providers/vllm/index.ts index bc419b70da5..751535eded5 100644 --- a/apps/sim/providers/vllm/index.ts +++ b/apps/sim/providers/vllm/index.ts @@ -17,6 +17,7 @@ import { createStreamingExecution } from '@/providers/streaming-execution' import { isAbortError, parseToolArguments } from '@/providers/streaming-tool-loop-shared' import { adaptOpenAIChatToolSchema } from '@/providers/tool-schema-adapter' import { enrichLastModelSegmentFromChatCompletions } from '@/providers/trace-enrichment' +import { openAICompatTransport } from '@/providers/transport' import type { Message, ProviderConfig, @@ -144,6 +145,7 @@ export const vllmProvider: ProviderConfig = { `vllm::${apiKey}::${baseUrl}::${pinnedIP ?? 'no-pin'}`, () => new OpenAI({ + ...openAICompatTransport(), apiKey, baseURL: `${baseUrl}/v1`, ...(pinnedFetch ? { fetch: pinnedFetch } : {}), diff --git a/apps/sim/providers/xai/index.ts b/apps/sim/providers/xai/index.ts index 74eee7568e1..4f1685fb3c5 100644 --- a/apps/sim/providers/xai/index.ts +++ b/apps/sim/providers/xai/index.ts @@ -14,6 +14,7 @@ import { createStreamingExecution } from '@/providers/streaming-execution' import { isAbortError, parseToolArguments } from '@/providers/streaming-tool-loop-shared' import { adaptOpenAIChatToolSchema } from '@/providers/tool-schema-adapter' import { enrichLastModelSegmentFromChatCompletions } from '@/providers/trace-enrichment' +import { openAICompatTransport } from '@/providers/transport' import type { Message, ProviderConfig, @@ -53,6 +54,7 @@ export const xAIProvider: ProviderConfig = { } const xai = new OpenAI({ + ...openAICompatTransport(), apiKey: request.apiKey, baseURL: 'https://api.x.ai/v1', }) diff --git a/apps/sim/providers/zai/index.ts b/apps/sim/providers/zai/index.ts index 7b3d008b22c..56a3f3a25a8 100644 --- a/apps/sim/providers/zai/index.ts +++ b/apps/sim/providers/zai/index.ts @@ -14,6 +14,7 @@ import { createStreamingExecution } from '@/providers/streaming-execution' import { isAbortError, parseToolArguments } from '@/providers/streaming-tool-loop-shared' import { adaptOpenAIChatToolSchema } from '@/providers/tool-schema-adapter' import { enrichLastModelSegmentFromChatCompletions } from '@/providers/trace-enrichment' +import { openAICompatTransport } from '@/providers/transport' import type { ProviderConfig, ProviderRequest, @@ -83,6 +84,7 @@ export const zaiProvider: ProviderConfig = { try { const zai = new OpenAI({ + ...openAICompatTransport(), apiKey: request.apiKey, baseURL: ZAI_BASE_URL, })