(threadId, runId, afterSeq = '0-0')
| 238 | } |
| 239 | |
| 240 | const startRunStream = async (threadId, runId, afterSeq = '0-0') => { |
| 241 | if (!threadId || !runId) return |
| 242 | const ts = getThreadState(threadId) |
| 243 | if (!ts) return |
| 244 | |
| 245 | stopRunStreamSubscription(threadId) |
| 246 | const runController = new AbortController() |
| 247 | ts.runStreamAbortController = runController |
| 248 | ts.activeRunId = runId |
| 249 | ts.runLastSeq = normalizeRunSeq(afterSeq) |
| 250 | ts.lastRetryableJobTry = null |
| 251 | ts.isStreaming = true |
| 252 | saveActiveRunSnapshot(threadId, runId, ts.runLastSeq) |
| 253 | const touchedThreadIds = new Set([threadId]) |
| 254 | let sawTerminalEvent = false |
| 255 | |
| 256 | try { |
| 257 | const response = await agentApi.streamAgentRunEvents(runId, ts.runLastSeq, { |
| 258 | signal: runController.signal |
| 259 | }) |
| 260 | if (!response.ok) { |
| 261 | throw new Error(`SSE response not ok: ${response.status}`) |
| 262 | } |
| 263 | |
| 264 | await processRunSseResponse(response, (event, data, eventId) => { |
| 265 | if (!data || ts.activeRunId !== runId) return |
| 266 | |
| 267 | if (eventId) { |
| 268 | const incomingSeq = normalizeRunSeq(eventId) |
| 269 | if (compareRunSeq(incomingSeq, ts.runLastSeq) <= 0) return |
| 270 | ts.runLastSeq = incomingSeq |
| 271 | saveActiveRunSnapshot(threadId, runId, incomingSeq) |
| 272 | } |
| 273 | |
| 274 | const payload = data.payload || {} |
| 275 | const terminalStatus = event === 'end' ? payload.status : data.status |
| 276 | const isRetryableError = |
| 277 | event === 'error' && (payload?.retryable === true || payload?.chunk?.retryable === true) |
| 278 | if (isRetryableError) { |
| 279 | const parsedJobTry = Number.parseInt(payload?.chunk?.job_try, 10) |
| 280 | const retryJobTry = Number.isNaN(parsedJobTry) ? null : parsedJobTry |
| 281 | if (retryJobTry !== null && ts.lastRetryableJobTry === retryJobTry) { |
| 282 | return |
| 283 | } |
| 284 | ts.lastRetryableJobTry = retryJobTry |
| 285 | console.warn('Run encountered retryable error, waiting for worker retry', { |
| 286 | threadId, |
| 287 | runId, |
| 288 | retryJobTry, |
| 289 | errorType: payload?.chunk?.error_type |
| 290 | }) |
| 291 | return |
| 292 | } |
| 293 | |
| 294 | if (Array.isArray(payload.items)) { |
| 295 | payload.items.forEach((chunk) => { |
| 296 | const routeThreadId = resolveChunkThreadId({ |
| 297 | envelope: data, |
no test coverage detected