diff --git a/packages/sdk/src/internal/core/session-runtime.ts b/packages/sdk/src/internal/core/session-runtime.ts index f765302..e6767bc 100644 --- a/packages/sdk/src/internal/core/session-runtime.ts +++ b/packages/sdk/src/internal/core/session-runtime.ts @@ -110,6 +110,7 @@ const TERMINAL_SESSION_STATUSES = new Set(["idle", "completed", "failed", "termi const DEFAULT_POLL_INTERVAL_MS = 2000; const DEFAULT_POLL_TIMEOUT_MS = 10 * 60 * 1000; +const POLL_INITIAL_INTERVAL_MS = 300; export function resolveAgentName(agents: Record | undefined, agentName?: string): string { if (agentName) return agentName; @@ -258,8 +259,9 @@ export async function collectEventsUntilTerminal( } = {}, ): Promise> { const start = Date.now(); - const pollIntervalMs = options.pollIntervalMs ?? DEFAULT_POLL_INTERVAL_MS; + const maxPollIntervalMs = options.pollIntervalMs ?? DEFAULT_POLL_INTERVAL_MS; const pollTimeoutMs = options.pollTimeoutMs ?? DEFAULT_POLL_TIMEOUT_MS; + let currentIntervalMs = Math.min(POLL_INITIAL_INTERVAL_MS, maxPollIntervalMs); let terminalStatus = "idle"; let result: ProviderSessionEventList | undefined; @@ -277,7 +279,8 @@ export async function collectEventsUntilTerminal( terminalStatus = terminalEvent.status; break; } - await delay(pollIntervalMs); + await delay(currentIntervalMs); + currentIntervalMs = Math.min(currentIntervalMs * 2, maxPollIntervalMs); } } else { while (true) { @@ -287,7 +290,8 @@ export async function collectEventsUntilTerminal( terminalStatus = session.status; break; } - await delay(pollIntervalMs); + await delay(currentIntervalMs); + currentIntervalMs = Math.min(currentIntervalMs * 2, maxPollIntervalMs); } result = await adapter.listSessionEvents(sessionId, { limit: 100 }); @@ -540,13 +544,40 @@ function buildAgentNameByRemoteId(ctx: ProjectRuntimeContext, provider: string): } // Qoder: send first (returns event ID), then stream from that ID to avoid missing events. +// When the provider closes the SSE connection mid-turn (e.g. after emitting a +// `session.status_idle` with `stop_reason=requires_action` for tool execution), +// we reconnect automatically until the stream delivers a true terminal status. async function* streamWithResume( adapter: SessionWorkflowAdapter, sessionId: string, message: string, ): AsyncIterable { const eventId = await adapter.sendSessionMessage(sessionId, message); - yield* adapter.streamSessionEvents(sessionId, eventId ? { after_id: eventId } : undefined); + let lastEventId: string | undefined = eventId; + let reachedTerminal = false; + let reconnectIntervalMs = POLL_INITIAL_INTERVAL_MS; + const start = Date.now(); + + while (!reachedTerminal) { + assertNotTimedOut(start, DEFAULT_POLL_TIMEOUT_MS); + for await (const event of adapter.streamSessionEvents( + sessionId, + lastEventId ? { after_id: lastEventId } : undefined, + )) { + if (event.id) lastEventId = event.id; + yield event; + if (event.type === "status" && isTerminalSessionStatus(event.status)) { + reachedTerminal = true; + break; + } + } + if (!reachedTerminal) { + // SSE closed without a terminal event — back off before reconnecting + // using exponential backoff capped at DEFAULT_POLL_INTERVAL_MS. + await delay(reconnectIntervalMs); + reconnectIntervalMs = Math.min(reconnectIntervalMs * 2, DEFAULT_POLL_INTERVAL_MS); + } + } } // Claude/Bailian: connect stream first, then send — provider pushes events immediately on send. diff --git a/packages/sdk/src/internal/providers/base-client.ts b/packages/sdk/src/internal/providers/base-client.ts index a506f3a..db47ccf 100644 --- a/packages/sdk/src/internal/providers/base-client.ts +++ b/packages/sdk/src/internal/providers/base-client.ts @@ -104,6 +104,7 @@ export abstract class BaseApiClient { boundary = buffer.indexOf("\n\n"); const dataLines: string[] = []; + let sseId: string | undefined; for (const line of frame.split("\n")) { if (line.startsWith(":")) continue; if (line.startsWith("event:") && line.slice(6).trim() === "heartbeat") { @@ -113,12 +114,19 @@ export abstract class BaseApiClient { if (line.startsWith("data:")) { dataLines.push(line.slice(5).trimStart()); } + if (line.startsWith("id:")) { + sseId = line.slice(3).trimStart(); + } } if (dataLines.length === 0) continue; const json = dataLines.join("\n"); try { - yield JSON.parse(json) as Record; + const parsed = JSON.parse(json) as Record; + if (sseId !== undefined && parsed.id === undefined) { + parsed.id = sseId; + } + yield parsed; } catch { // skip unparseable frames } diff --git a/packages/sdk/src/internal/providers/qoder/mapper.ts b/packages/sdk/src/internal/providers/qoder/mapper.ts index e06d552..a29a8be 100644 --- a/packages/sdk/src/internal/providers/qoder/mapper.ts +++ b/packages/sdk/src/internal/providers/qoder/mapper.ts @@ -551,6 +551,7 @@ export function toSessionEvent(raw: Record): ProviderSessionEve const type: SessionEventType = QODER_EVENT_MAP[rawType] ?? "unknown"; const event: ProviderSessionEvent = { type, raw_type: rawType, raw }; + if (typeof raw.id === "string") event.id = raw.id; if (typeof raw.role === "string") event.role = raw.role; if (type === "message") { @@ -567,12 +568,17 @@ export function toSessionEvent(raw: Record): ProviderSessionEve } else if (type === "status") { // Only session-level idle/terminated are terminal; thread-level idle // (session.thread_status_idle) is non-terminal and should not stop the stream. + // Additionally, session.status_idle with stop_reason "requires_action" means + // the agent paused for tool execution and will resume — treat it as non-terminal. + const stopReason = extractStopReason(raw.stop_reason); if (rawType === "session.thread_status_idle") { event.status = "running"; + } else if (rawType === "session.status_idle" && stopReason === "requires_action") { + event.status = "running"; } else { event.status = rawType.includes("idle") ? "idle" : rawType.includes("terminated") ? "terminated" : "running"; } - event.stop_reason = extractStopReason(raw.stop_reason); + event.stop_reason = stopReason; } else if (type === "error") { event.content = extractErrorMessage(raw); } diff --git a/packages/sdk/tests/unit/session-event-mappers.test.ts b/packages/sdk/tests/unit/session-event-mappers.test.ts index 9de96d6..ee87449 100644 --- a/packages/sdk/tests/unit/session-event-mappers.test.ts +++ b/packages/sdk/tests/unit/session-event-mappers.test.ts @@ -205,6 +205,11 @@ describe("Qoder mapper", () => { expect(qoderToEvent({ type: "agent.tool_result", content: "result text" }).type).toBe("tool_result"); expect(qoderToEvent({ type: "session.status_idle", stop_reason: "end_turn" }).status).toBe("idle"); expect(qoderToEvent({ type: "session.status_running" }).status).toBe("running"); + // requires_action means the agent paused for tool execution; not truly terminal + expect(qoderToEvent({ type: "session.status_idle", stop_reason: "requires_action" }).status).toBe("running"); + expect(qoderToEvent({ type: "session.status_idle", stop_reason: { type: "requires_action" } }).status).toBe( + "running", + ); expect(qoderToEvent({ type: "agent.thinking" }).type).toBe("thinking"); const error = qoderToEvent({ type: "session.error", error: "timeout" }); expect(error.type).toBe("error");