Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
39 changes: 35 additions & 4 deletions packages/sdk/src/internal/core/session-runtime.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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;

Expand All @@ -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) {
Expand All @@ -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 });
Expand Down Expand Up @@ -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.
Expand Down
10 changes: 9 additions & 1 deletion packages/sdk/src/internal/providers/base-client.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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") {
Expand All @@ -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
}
Expand Down
8 changes: 7 additions & 1 deletion packages/sdk/src/internal/providers/qoder/mapper.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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") {
Expand All @@ -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);
}
Expand Down
5 changes: 5 additions & 0 deletions packages/sdk/tests/unit/session-event-mappers.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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");
Expand Down