opensource/sim
3
0
Fork 0
mirror of https://github.com/simstudioai/sim.git synced 2026-09-24 15:45:35 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(memory): fix O(n²) string concatenation and unconsumed fetch response leaks (#3399)

Browse Source 
* fix(monitoring): set MemoryTelemetry logger to INFO level for production visibility

Production defaults to ERROR-only logging. Without this override,
memory snapshots would be silently suppressed.

* fix(memory): fix O(n²) string concatenation and unconsumed fetch response leaks

* fix(tests): add text() mock to workflow-handler test fetch responses

* fix(memory): remove unused O(n²) join in onStreamChunk callback
This commit is contained in:
Waleed
2026-03-02 13:58:03 -08:00
committed by GitHub
parent 61a447aba5
commit ebb9a2bdd3
20 changed files with 131 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/sim/app/api/a2a/serve/[agentId]/route.ts
+3 -2
View File
@@ -711,7 +711,7 @@ async function handleMessageStream(
if (response.body && isStreamingResponse) {
const reader = response.body.getReader()
const decoder = new TextDecoder()
let accumulatedContent = ''
const contentChunks: string[] = []
let finalContent: string | undefined
while (true) {
@@ -722,7 +722,7 @@ async function handleMessageStream(
const parsed = parseWorkflowSSEChunk(rawChunk)
if (parsed.content) {
accumulatedContent += parsed.content
contentChunks.push(parsed.content)
sendEvent('message', {
kind: 'message',
taskId,
@@ -738,6 +738,7 @@ async function handleMessageStream(
}
}
const accumulatedContent = contentChunks.join('')
const messageContent =
(finalContent !== undefined && finalContent.length > 0
? finalContent
apps/sim/app/api/copilot/chat/route.ts
+6 -2
View File
@@ -405,13 +405,17 @@ export async function POST(req: NextRequest) {
},
})
} finally {
controller.close()
try {
controller.close()
} catch {
// controller may already be closed by cancel()
}
}
},
async cancel() {
clientDisconnected = true
if (eventWriter) {
await eventWriter.flush()
await eventWriter.close().catch(() => {})
}
},
})
apps/sim/app/api/tools/video/route.ts
+12
View File
@@ -342,6 +342,7 @@ async function generateWithRunway(
})
if (!statusResponse.ok) {
await statusResponse.text().catch(() => {})
throw new Error(`Runway status check failed: ${statusResponse.status}`)
}
@@ -352,6 +353,7 @@ async function generateWithRunway(
const videoResponse = await fetch(statusData.output[0])
if (!videoResponse.ok) {
await videoResponse.text().catch(() => {})
throw new Error(`Failed to download video: ${videoResponse.status}`)
}
@@ -448,6 +450,7 @@ async function generateWithVeo(
)
if (!statusResponse.ok) {
await statusResponse.text().catch(() => {})
throw new Error(`Veo status check failed: ${statusResponse.status}`)
}
@@ -472,6 +475,7 @@ async function generateWithVeo(
})
if (!videoResponse.ok) {
await videoResponse.text().catch(() => {})
throw new Error(`Failed to download video: ${videoResponse.status}`)
}
@@ -561,6 +565,7 @@ async function generateWithLuma(
)
if (!statusResponse.ok) {
await statusResponse.text().catch(() => {})
throw new Error(`Luma status check failed: ${statusResponse.status}`)
}
@@ -576,6 +581,7 @@ async function generateWithLuma(
const videoResponse = await fetch(videoUrl)
if (!videoResponse.ok) {
await videoResponse.text().catch(() => {})
throw new Error(`Failed to download video: ${videoResponse.status}`)
}
@@ -679,6 +685,7 @@ async function generateWithMiniMax(
)
if (!statusResponse.ok) {
await statusResponse.text().catch(() => {})
throw new Error(`MiniMax status check failed: ${statusResponse.status}`)
}
@@ -712,6 +719,7 @@ async function generateWithMiniMax(
)
if (!fileResponse.ok) {
await fileResponse.text().catch(() => {})
throw new Error(`Failed to download video: ${fileResponse.status}`)
}
@@ -725,6 +733,7 @@ async function generateWithMiniMax(
// Download the actual video file
const videoResponse = await fetch(videoUrl)
if (!videoResponse.ok) {
await videoResponse.text().catch(() => {})
throw new Error(`Failed to download video from URL: ${videoResponse.status}`)
}
@@ -881,6 +890,7 @@ async function generateWithFalAI(
)
if (!statusResponse.ok) {
await statusResponse.text().catch(() => {})
throw new Error(`Fal.ai status check failed: ${statusResponse.status}`)
}
@@ -899,6 +909,7 @@ async function generateWithFalAI(
)
if (!resultResponse.ok) {
await resultResponse.text().catch(() => {})
throw new Error(`Failed to fetch result: ${resultResponse.status}`)
}
@@ -911,6 +922,7 @@ async function generateWithFalAI(
const videoResponse = await fetch(videoUrl)
if (!videoResponse.ok) {
await videoResponse.text().catch(() => {})
throw new Error(`Failed to download video: ${videoResponse.status}`)
}
apps/sim/app/workspace/[workspaceId]/w/[workflowId]/hooks/use-workflow-execution.ts
+20 -12
View File
@@ -745,7 +745,7 @@ export function useWorkflowExecution() {
const stream = new ReadableStream({
async start(controller) {
const { encodeSSE } = await import('@/lib/core/utils/sse')
const streamedContent = new Map<string, string>()
const streamedChunks = new Map<string, string[]>()
const streamReadingPromises: Promise<void>[] = []
const safeEnqueue = (data: Uint8Array) => {
@@ -845,8 +845,8 @@ export function useWorkflowExecution() {
const reader = streamingExecution.stream.getReader()
const blockId = (streamingExecution.execution as any)?.blockId
if (blockId && !streamedContent.has(blockId)) {
streamedContent.set(blockId, '')
if (blockId && !streamedChunks.has(blockId)) {
streamedChunks.set(blockId, [])
}
try {
@@ -860,13 +860,13 @@ export function useWorkflowExecution() {
}
const chunk = new TextDecoder().decode(value)
if (blockId) {
streamedContent.set(blockId, (streamedContent.get(blockId) || '') + chunk)
streamedChunks.get(blockId)!.push(chunk)
}
let chunkToSend = chunk
if (blockId && !processedFirstChunk.has(blockId)) {
processedFirstChunk.add(blockId)
if (streamedContent.size > 1) {
if (streamedChunks.size > 1) {
chunkToSend = `\n\n${chunk}`
}
}
@@ -884,7 +884,7 @@ export function useWorkflowExecution() {
// Handle non-streaming blocks (like Function blocks)
const onBlockComplete = async (blockId: string, output: any) => {
// Skip if this block already had streaming content (avoid duplicates)
if (streamedContent.has(blockId)) {
if (streamedChunks.has(blockId)) {
logger.debug('[handleRunWorkflow] Skipping onBlockComplete for streaming block', {
blockId,
})
@@ -921,13 +921,13 @@ export function useWorkflowExecution() {
: JSON.stringify(outputValue, null, 2)
// Add separator if this isn't the first output
const separator = streamedContent.size > 0 ? '\n\n' : ''
const separator = streamedChunks.size > 0 ? '\n\n' : ''
// Send the non-streaming block output as a chunk
safeEnqueue(encodeSSE({ blockId, chunk: separator + formattedOutput }))
// Track that we've sent output for this block
streamedContent.set(blockId, formattedOutput)
streamedChunks.set(blockId, [formattedOutput])
}
}
}
@@ -969,6 +969,12 @@ export function useWorkflowExecution() {
})
}
// Resolve chunks to final strings for consumption
const streamedContent = new Map<string, string>()
for (const [id, chunks] of streamedChunks) {
streamedContent.set(id, chunks.join(''))
}
// Update streamed content and apply tokenization
if (result.logs) {
result.logs.forEach((log: BlockLog) => {
@@ -1316,7 +1322,7 @@ export function useWorkflowExecution() {
const activeBlocksSet = new Set<string>()
const activeBlockRefCounts = new Map<string, number>()
const streamedContent = new Map<string, string>()
const streamedChunks = new Map<string, string[]>()
const accumulatedBlockLogs: BlockLog[] = []
const accumulatedBlockStates = new Map<string, BlockState>()
const executedBlockIds = new Set<string>()
@@ -1374,8 +1380,10 @@ export function useWorkflowExecution() {
onBlockChildWorkflowStarted: blockHandlers.onBlockChildWorkflowStarted,
onStreamChunk: (data) => {
const existing = streamedContent.get(data.blockId) || ''
streamedContent.set(data.blockId, existing + data.chunk)
if (!streamedChunks.has(data.blockId)) {
streamedChunks.set(data.blockId, [])
}
streamedChunks.get(data.blockId)!.push(data.chunk)
// Call onStream callback if provided (create a fake StreamingExecution)
if (onStream && isExecutingFromChat) {
@@ -1390,7 +1398,7 @@ export function useWorkflowExecution() {
stream,
execution: {
success: true,
output: { content: existing + data.chunk },
output: { content: '' },
blockId: data.blockId,
} as any,
}
apps/sim/executor/handlers/agent/agent-handler.ts
+1
View File
@@ -357,6 +357,7 @@ export class AgentBlockHandler implements BlockHandler {
})
if (!response.ok) {
await response.text().catch(() => {})
logger.error(`Failed to fetch custom tools: ${response.status}`)
return null
}
apps/sim/executor/handlers/agent/memory.ts
+5 -4
View File
@@ -116,21 +116,22 @@ export class Memory {
ctx: ExecutionContext,
inputs: AgentInputs
): ReadableStream<Uint8Array> {
let accumulatedContent = ''
const chunks: string[] = []
const decoder = new TextDecoder()
const transformStream = new TransformStream<Uint8Array, Uint8Array>({
transform: (chunk, controller) => {
controller.enqueue(chunk)
const decoded = decoder.decode(chunk, { stream: true })
accumulatedContent += decoded
chunks.push(decoded)
},
flush: () => {
if (accumulatedContent.trim()) {
const content = chunks.join('')
if (content.trim()) {
this.appendToMemory(ctx, inputs, {
role: 'assistant',
content: accumulatedContent,
content,
}).catch((error) => logger.error('Failed to persist streaming response:', error))
}
},
apps/sim/executor/handlers/workflow/workflow-handler.test.ts
+2
View File
@@ -142,6 +142,7 @@ describe('WorkflowBlockHandler', () => {
ok: false,
status: 404,
statusText: 'Not Found',
text: () => Promise.resolve(''),
})
await expect(handler.execute(mockContext, mockBlock, inputs)).rejects.toThrow(
@@ -168,6 +169,7 @@ describe('WorkflowBlockHandler', () => {
ok: false,
status: 404,
statusText: 'Not Found',
text: () => Promise.resolve(''),
})
const result = await (handler as any).loadChildWorkflow(workflowId)
apps/sim/executor/handlers/workflow/workflow-handler.ts
+1
View File
@@ -375,6 +375,7 @@ export class WorkflowBlockHandler implements BlockHandler {
const response = await fetch(url.toString(), { headers })
if (!response.ok) {
await response.text().catch(() => {})
if (response.status === HTTP.STATUS.NOT_FOUND) {
logger.warn(`Child workflow ${workflowId} not found`)
return null
apps/sim/lib/a2a/push-notifications.ts
+1
View File
@@ -77,6 +77,7 @@ export async function deliverPushNotification(taskId: string, state: TaskState):
})
if (!response.ok) {
await response.text().catch(() => {})
logger.error('Push notification delivery failed', {
taskId,
url: config.url,
apps/sim/lib/auth/auth.ts
+35
View File
@@ -773,6 +773,7 @@ export const auth = betterAuth({
})
if (!profileResponse.ok) {
await profileResponse.text().catch(() => {})
logger.error('Failed to fetch GitHub profile', {
status: profileResponse.status,
statusText: profileResponse.statusText,
@@ -850,6 +851,7 @@ export const auth = betterAuth({
headers: { Authorization: `Bearer ${tokens.accessToken}` },
})
if (!response.ok) {
await response.text().catch(() => {})
logger.error('Failed to fetch Google user info', { status: response.status })
throw new Error(`Failed to fetch Google user info: ${response.statusText}`)
}
@@ -889,6 +891,7 @@ export const auth = betterAuth({
headers: { Authorization: `Bearer ${tokens.accessToken}` },
})
if (!response.ok) {
await response.text().catch(() => {})
logger.error('Failed to fetch Google user info', { status: response.status })
throw new Error(`Failed to fetch Google user info: ${response.statusText}`)
}
@@ -929,6 +932,7 @@ export const auth = betterAuth({
headers: { Authorization: `Bearer ${tokens.accessToken}` },
})
if (!response.ok) {
await response.text().catch(() => {})
logger.error('Failed to fetch Google user info', { status: response.status })
throw new Error(`Failed to fetch Google user info: ${response.statusText}`)
}
@@ -969,6 +973,7 @@ export const auth = betterAuth({
headers: { Authorization: `Bearer ${tokens.accessToken}` },
})
if (!response.ok) {
await response.text().catch(() => {})
logger.error('Failed to fetch Google user info', { status: response.status })
throw new Error(`Failed to fetch Google user info: ${response.statusText}`)
}
@@ -1009,6 +1014,7 @@ export const auth = betterAuth({
headers: { Authorization: `Bearer ${tokens.accessToken}` },
})
if (!response.ok) {
await response.text().catch(() => {})
logger.error('Failed to fetch Google user info', { status: response.status })
throw new Error(`Failed to fetch Google user info: ${response.statusText}`)
}
@@ -1049,6 +1055,7 @@ export const auth = betterAuth({
headers: { Authorization: `Bearer ${tokens.accessToken}` },
})
if (!response.ok) {
await response.text().catch(() => {})
logger.error('Failed to fetch Google user info', { status: response.status })
throw new Error(`Failed to fetch Google user info: ${response.statusText}`)
}
@@ -1090,6 +1097,7 @@ export const auth = betterAuth({
headers: { Authorization: `Bearer ${tokens.accessToken}` },
})
if (!response.ok) {
await response.text().catch(() => {})
logger.error('Failed to fetch Google user info', { status: response.status })
throw new Error(`Failed to fetch Google user info: ${response.statusText}`)
}
@@ -1129,6 +1137,7 @@ export const auth = betterAuth({
headers: { Authorization: `Bearer ${tokens.accessToken}` },
})
if (!response.ok) {
await response.text().catch(() => {})
logger.error('Failed to fetch Google user info', { status: response.status })
throw new Error(`Failed to fetch Google user info: ${response.statusText}`)
}
@@ -1170,6 +1179,7 @@ export const auth = betterAuth({
headers: { Authorization: `Bearer ${tokens.accessToken}` },
})
if (!response.ok) {
await response.text().catch(() => {})
logger.error('Failed to fetch Google user info', { status: response.status })
throw new Error(`Failed to fetch Google user info: ${response.statusText}`)
}
@@ -1211,6 +1221,7 @@ export const auth = betterAuth({
headers: { Authorization: `Bearer ${tokens.accessToken}` },
})
if (!response.ok) {
await response.text().catch(() => {})
logger.error('Failed to fetch Google user info', { status: response.status })
throw new Error(`Failed to fetch Google user info: ${response.statusText}`)
}
@@ -1251,6 +1262,7 @@ export const auth = betterAuth({
headers: { Authorization: `Bearer ${tokens.accessToken}` },
})
if (!response.ok) {
await response.text().catch(() => {})
logger.error('Failed to fetch Google user info', { status: response.status })
throw new Error(`Failed to fetch Google user info: ${response.statusText}`)
}
@@ -1291,6 +1303,7 @@ export const auth = betterAuth({
headers: { Authorization: `Bearer ${tokens.accessToken}` },
})
if (!response.ok) {
await response.text().catch(() => {})
logger.error('Failed to fetch Google user info', { status: response.status })
throw new Error(`Failed to fetch Google user info: ${response.statusText}`)
}
@@ -1352,6 +1365,7 @@ export const auth = betterAuth({
headers: { Authorization: `Bearer ${tokens.accessToken}` },
})
if (!response.ok) {
await response.text().catch(() => {})
logger.error('Failed to fetch Microsoft user info', { status: response.status })
throw new Error(`Failed to fetch Microsoft user info: ${response.statusText}`)
}
@@ -1391,6 +1405,7 @@ export const auth = betterAuth({
headers: { Authorization: `Bearer ${tokens.accessToken}` },
})
if (!response.ok) {
await response.text().catch(() => {})
logger.error('Failed to fetch Microsoft user info', { status: response.status })
throw new Error(`Failed to fetch Microsoft user info: ${response.statusText}`)
}
@@ -1485,6 +1500,7 @@ export const auth = betterAuth({
headers: { Authorization: `Bearer ${tokens.accessToken}` },
})
if (!response.ok) {
await response.text().catch(() => {})
logger.error('Failed to fetch Microsoft user info', { status: response.status })
throw new Error(`Failed to fetch Microsoft user info: ${response.statusText}`)
}
@@ -1533,6 +1549,7 @@ export const auth = betterAuth({
headers: { Authorization: `Bearer ${tokens.accessToken}` },
})
if (!response.ok) {
await response.text().catch(() => {})
logger.error('Failed to fetch Microsoft user info', { status: response.status })
throw new Error(`Failed to fetch Microsoft user info: ${response.statusText}`)
}
@@ -1572,6 +1589,7 @@ export const auth = betterAuth({
headers: { Authorization: `Bearer ${tokens.accessToken}` },
})
if (!response.ok) {
await response.text().catch(() => {})
logger.error('Failed to fetch Microsoft user info', { status: response.status })
throw new Error(`Failed to fetch Microsoft user info: ${response.statusText}`)
}
@@ -1619,6 +1637,7 @@ export const auth = betterAuth({
headers: { Authorization: `Bearer ${tokens.accessToken}` },
})
if (!response.ok) {
await response.text().catch(() => {})
logger.error('Failed to fetch Microsoft user info', { status: response.status })
throw new Error(`Failed to fetch Microsoft user info: ${response.statusText}`)
}
@@ -1701,6 +1720,7 @@ export const auth = betterAuth({
})
if (!response.ok) {
await response.text().catch(() => {})
logger.error('Failed to fetch Pipedrive user info', {
status: response.status,
})
@@ -1848,6 +1868,7 @@ export const auth = betterAuth({
)
if (!response.ok) {
await response.text().catch(() => {})
logger.error('Failed to fetch Salesforce user info', {
status: response.status,
})
@@ -1915,6 +1936,7 @@ export const auth = betterAuth({
)
if (!response.ok) {
await response.text().catch(() => {})
logger.error('Error fetching X user info:', {
status: response.status,
statusText: response.statusText,
@@ -2009,6 +2031,7 @@ export const auth = betterAuth({
})
if (!response.ok) {
await response.text().catch(() => {})
logger.error('Error fetching Confluence user info:', {
status: response.status,
statusText: response.statusText,
@@ -2120,6 +2143,7 @@ export const auth = betterAuth({
})
if (!response.ok) {
await response.text().catch(() => {})
logger.error('Error fetching Jira user info:', {
status: response.status,
statusText: response.statusText,
@@ -2177,6 +2201,7 @@ export const auth = betterAuth({
})
if (!response.ok) {
await response.text().catch(() => {})
logger.error('Error fetching Airtable user info:', {
status: response.status,
statusText: response.statusText,
@@ -2226,6 +2251,7 @@ export const auth = betterAuth({
})
if (!response.ok) {
await response.text().catch(() => {})
logger.error('Error fetching Notion user info:', {
status: response.status,
statusText: response.statusText,
@@ -2293,6 +2319,7 @@ export const auth = betterAuth({
})
if (!response.ok) {
await response.text().catch(() => {})
logger.error('Error fetching Reddit user info:', {
status: response.status,
statusText: response.statusText,
@@ -2540,6 +2567,7 @@ export const auth = betterAuth({
})
if (!response.ok) {
await response.text().catch(() => {})
logger.error('Error fetching Asana user info:', {
status: response.status,
statusText: response.statusText,
@@ -2606,6 +2634,7 @@ export const auth = betterAuth({
})
if (!response.ok) {
await response.text().catch(() => {})
logger.error('Slack auth.test failed', {
status: response.status,
statusText: response.statusText,
@@ -2665,6 +2694,7 @@ export const auth = betterAuth({
})
if (!response.ok) {
await response.text().catch(() => {})
logger.error('Error fetching Webflow user info:', {
status: response.status,
statusText: response.statusText,
@@ -2716,6 +2746,7 @@ export const auth = betterAuth({
})
if (!response.ok) {
await response.text().catch(() => {})
logger.error('Failed to fetch LinkedIn user info', {
status: response.status,
statusText: response.statusText,
@@ -2778,6 +2809,7 @@ export const auth = betterAuth({
})
if (!response.ok) {
await response.text().catch(() => {})
logger.error('Failed to fetch Zoom user info', {
status: response.status,
statusText: response.statusText,
@@ -2845,6 +2877,7 @@ export const auth = betterAuth({
})
if (!response.ok) {
await response.text().catch(() => {})
logger.error('Failed to fetch Spotify user info', {
status: response.status,
statusText: response.statusText,
@@ -2893,6 +2926,7 @@ export const auth = betterAuth({
})
if (!response.ok) {
await response.text().catch(() => {})
logger.error('Failed to fetch WordPress.com user info', {
status: response.status,
statusText: response.statusText,
@@ -2942,6 +2976,7 @@ export const auth = betterAuth({
})
if (!response.ok) {
await response.text().catch(() => {})
logger.error('Failed to fetch Cal.com user info', {
status: response.status,
statusText: response.statusText,
apps/sim/lib/billing/client/upgrade.ts
+1
View File
@@ -43,6 +43,7 @@ export function useSubscriptionUpgrade() {
try {
const orgsResponse = await fetch('/api/organizations')
if (!orgsResponse.ok) {
await orgsResponse.text().catch(() => {})
throw new Error('Failed to check organization status')
}
apps/sim/lib/copilot/messages/checkpoints.ts
+1
View File
@@ -71,6 +71,7 @@ export async function saveMessageCheckpoint(
})
if (!response.ok) {
await response.text().catch(() => {})
throw new Error(`Failed to create checkpoint: ${response.statusText}`)
}
apps/sim/lib/environment/api.ts
+2
View File
@@ -11,6 +11,7 @@ export async function fetchPersonalEnvironment(): Promise<Record<string, Environ
const response = await fetch(API_ENDPOINTS.ENVIRONMENT)
if (!response.ok) {
await response.text().catch(() => {})
throw new Error(`Failed to load environment variables: ${response.statusText}`)
}
@@ -29,6 +30,7 @@ export async function fetchWorkspaceEnvironment(
const response = await fetch(API_ENDPOINTS.WORKSPACE_ENVIRONMENT(workspaceId))
if (!response.ok) {
await response.text().catch(() => {})
throw new Error(`Failed to load workspace environment: ${response.statusText}`)
}
apps/sim/lib/webhooks/rss-polling-service.ts
+1
View File
@@ -278,6 +278,7 @@ async function fetchNewRssItems(
})
if (!response.ok) {
await response.text().catch(() => {})
throw new Error(`Failed to fetch RSS feed: ${response.status} ${response.statusText}`)
}
apps/sim/lib/workflows/streaming/streaming.ts
+35 -16
View File
@@ -52,11 +52,19 @@ export interface StreamingResponseOptions {
}
interface StreamingState {
streamedContent: Map<string, string>
streamedChunks: Map<string, string[]>
processedOutputs: Set<string>
streamCompletionTimes: Map<string, number>
}
function resolveStreamedContent(state: StreamingState): Map<string, string> {
const result = new Map<string, string>()
for (const [blockId, chunks] of state.streamedChunks) {
result.set(blockId, chunks.join(''))
}
return result
}
function extractOutputValue(output: unknown, path: string): unknown {
return traverseObjectPath(output, path)
}
@@ -125,17 +133,21 @@ async function buildMinimalResult(
return minimalResult
}
function updateLogsWithStreamedContent(logs: BlockLog[], state: StreamingState): BlockLog[] {
function updateLogsWithStreamedContent(
logs: BlockLog[],
streamedContent: Map<string, string>,
streamCompletionTimes: Map<string, number>
): BlockLog[] {
return logs.map((log: BlockLog) => {
if (!state.streamedContent.has(log.blockId)) {
if (!streamedContent.has(log.blockId)) {
return log
}
const content = state.streamedContent.get(log.blockId)
const content = streamedContent.get(log.blockId)
const updatedLog = { ...log }
if (state.streamCompletionTimes.has(log.blockId)) {
const completionTime = state.streamCompletionTimes.get(log.blockId)!
if (streamCompletionTimes.has(log.blockId)) {
const completionTime = streamCompletionTimes.get(log.blockId)!
const startTime = new Date(log.startedAt).getTime()
updatedLog.endedAt = new Date(completionTime).toISOString()
updatedLog.durationMs = completionTime - startTime
@@ -176,7 +188,7 @@ export async function createStreamingResponse(
return new ReadableStream({
async start(controller) {
const state: StreamingState = {
streamedContent: new Map(),
streamedChunks: new Map(),
processedOutputs: new Set(),
streamCompletionTimes: new Map(),
}
@@ -210,10 +222,10 @@ export async function createStreamingResponse(
}
const textChunk = decoder.decode(value, { stream: true })
state.streamedContent.set(
blockId,
(state.streamedContent.get(blockId) || '') + textChunk
)
if (!state.streamedChunks.has(blockId)) {
state.streamedChunks.set(blockId, [])
}
state.streamedChunks.get(blockId)!.push(textChunk)
if (isFirstChunk) {
sendChunk(blockId, textChunk)
@@ -242,7 +254,7 @@ export async function createStreamingResponse(
return
}
if (state.streamedContent.has(blockId)) {
if (state.streamedChunks.has(blockId)) {
return
}
@@ -292,9 +304,16 @@ export async function createStreamingResponse(
executionId
)
if (result.logs && state.streamedContent.size > 0) {
result.logs = updateLogsWithStreamedContent(result.logs, state)
processStreamingBlockLogs(result.logs, state.streamedContent)
const streamedContent =
state.streamedChunks.size > 0 ? resolveStreamedContent(state) : new Map<string, string>()
if (result.logs && streamedContent.size > 0) {
result.logs = updateLogsWithStreamedContent(
result.logs,
streamedContent,
state.streamCompletionTimes
)
processStreamingBlockLogs(result.logs, streamedContent)
}
if (
@@ -316,7 +335,7 @@ export async function createStreamingResponse(
const minimalResult = await buildMinimalResult(
result,
streamConfig.selectedOutputs,
state.streamedContent,
streamedContent,
requestId,
streamConfig.includeFileBase64 ?? true,
streamConfig.base64MaxBytes
apps/sim/providers/ollama/index.ts
+1
View File
@@ -37,6 +37,7 @@ export const ollamaProvider: ProviderConfig = {
try {
const response = await fetch(`${OLLAMA_HOST}/api/tags`)
if (!response.ok) {
await response.text().catch(() => {})
useProvidersStore.getState().setProviderModels('ollama', [])
logger.warn('Ollama service is not available. The provider will be disabled.')
return
apps/sim/providers/openrouter/utils.ts
+1
View File
@@ -29,6 +29,7 @@ async function fetchModelCapabilities(): Promise<Map<string, ModelCapabilities>>
})
if (!response.ok) {
await response.text().catch(() => {})
logger.warn('Failed to fetch OpenRouter model capabilities', {
status: response.status,
})
apps/sim/providers/utils.ts
+1
View File
@@ -73,6 +73,7 @@ async function fetchWorkflowMetadata(
const response = await fetch(url.toString(), { headers })
if (!response.ok) {
await response.text().catch(() => {})
logger.warn(`Failed to fetch workflow metadata for ${workflowId}`)
return null
}
apps/sim/providers/vllm/index.ts
+1
View File
@@ -57,6 +57,7 @@ export const vllmProvider: ProviderConfig = {
const response = await fetch(`${baseUrl}/v1/models`, { headers })
if (!response.ok) {
await response.text().catch(() => {})
useProvidersStore.getState().setProviderModels('vllm', [])
logger.warn('vLLM service is not available. The provider will be disabled.')
return
apps/sim/tools/utils.ts
+1
View File
@@ -401,6 +401,7 @@ async function fetchCustomToolFromAPI(
})
if (!response.ok) {
await response.text().catch(() => {})
logger.error(`Failed to fetch custom tools: ${response.statusText}`)
return undefined
}