fix(session): handle Qoder requires_action idle state and SSE reconnection

When Qoder agents execute tools, the session transitions through
idle(requires_action) before resuming. Three issues caused session run
to produce no output or hang:

1. Qoder mapper treated session.status_idle with stop_reason
   requires_action as terminal idle, causing poll/stream to exit before
   tool results and the final reply arrived. Fixed by mapping it to
   running.

2. streamWithResume did a single yield* of the SSE stream. When Qoder
   closes the SSE connection after requires_action, the generator exited
   without reconnecting. Added a reconnect loop with exponential backoff
   that continues until a true terminal status event arrives.

3. SSE parser in base-client discarded the id: field from SSE frames,
   so event.id was always undefined and stream reconnection could not
   skip already-consumed events. Now extracts id: and injects it into
   the parsed payload.

Additionally, polling in collectEventsUntilTerminal now uses exponential
backoff (300ms initial, doubling to 2s cap) instead of a fixed 2s
interval, reducing first-response latency.

Change-Id: Ieefe7ad827516f92fcfff3023600564d9eff0eec
Co-developed-by: OpenCode <noreply@opencode.ai>

Author: heimanba

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<string, unknown> | undefined, agentName?: string): string {
 	if (agentName) return agentName;
@@ -258,8 +259,9 @@ export async function collectEventsUntilTerminal(
 	} = {},
 ): Promise<Omit<CollectedSessionEvents, "eventId">> {
 	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<ProviderSessionEvent> {
 	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.

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<string, unknown>;
+						const parsed = JSON.parse(json) as Record<string, unknown>;
+						if (sseId !== undefined && parsed.id === undefined) {
+							parsed.id = sseId;
+						}
+						yield parsed;
 					} catch {
 						// skip unparseable frames
 					}

packages/sdk/src/internal/providers/qoder/mapper.ts

@@ -551,6 +551,7 @@ export function toSessionEvent(raw: Record<string, unknown>): 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<string, unknown>): 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);
 	}

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");