import type { ToolContextWithMetadata, OpencodeClient } from "./types" import type { SessionMessage } from "./executor-types" import { getDefaultSyncPollTimeoutMs, getTimingConfig } from "./timing" import { log } from "../../shared/logger" import { normalizeSDKResponse } from "../../shared" import { extractErrorMessage } from "../../features/background-agent/error-classifier" const NON_TERMINAL_FINISH_REASONS = new Set(["tool-calls", "unknown"]) const PENDING_TOOL_PART_TYPES = new Set(["tool", "tool_use", "tool-call"]) const ACTIVE_SESSION_STATUSES = new Set(["busy", "retry", "running"]) function wait(milliseconds: number): Promise { const sharedBuffer = new SharedArrayBuffer(Int32Array.BYTES_PER_ELEMENT) const typedArray = new Int32Array(sharedBuffer) const result = Atomics.waitAsync(typedArray, 0, 0, milliseconds) return result.async ? result.value.then(() => undefined) : Promise.resolve() } function abortSyncSession(client: OpencodeClient, sessionID: string, reason: string): void { log("[task] Aborting sync session", { sessionID, reason }) void client.session.abort({ path: { id: sessionID }, }).catch((error: unknown) => { log("[task] Failed to abort sync session", { sessionID, reason, error: String(error) }) }) } function isActiveSessionStatus(status: { type: string } | undefined): boolean { return status !== undefined && ACTIVE_SESSION_STATUSES.has(status.type) } async function fetchSessionMessages( client: OpencodeClient, sessionID: string ): Promise { const messagesResult = await client.session.messages({ path: { id: sessionID } }) const rawData = (messagesResult as { data?: unknown })?.data ?? messagesResult return Array.isArray(rawData) ? (rawData as SessionMessage[]) : [] } function getTerminalSessionError(messages: SessionMessage[]): string | null { const lastAssistant = [...messages].reverse().find((msg) => msg.info?.role === "assistant") const lastUser = [...messages].reverse().find((msg) => msg.info?.role === "user") if (lastUser?.info?.id && lastAssistant?.info?.id && lastAssistant.info.id <= lastUser.info.id) { return null } if (!lastAssistant?.info || !("error" in lastAssistant.info)) { return null } const errorMessage = extractErrorMessage((lastAssistant.info as { error?: unknown }).error) return errorMessage && errorMessage.length > 0 ? errorMessage : "Session error" } export function isSessionComplete(messages: SessionMessage[]): boolean { let lastUser: SessionMessage | undefined let lastAssistant: SessionMessage | undefined for (let i = messages.length - 1; i >= 0; i--) { const msg = messages[i] if (!lastAssistant && msg.info?.role === "assistant") lastAssistant = msg if (!lastUser && msg.info?.role === "user") lastUser = msg if (lastUser && lastAssistant) break } if (!lastAssistant?.info?.finish) return false if (NON_TERMINAL_FINISH_REASONS.has(lastAssistant.info.finish)) return false if (lastAssistant.parts?.some((part) => part.type && PENDING_TOOL_PART_TYPES.has(part.type))) return false if (!lastUser?.info?.id || !lastAssistant?.info?.id) return false return lastUser.info.id < lastAssistant.info.id } const DEFAULT_MAX_ASSISTANT_TURNS = 300 export async function pollSyncSession( ctx: ToolContextWithMetadata, client: OpencodeClient, input: { sessionID: string agentToUse: string toastManager: { removeTask: (id: string) => void } | null | undefined taskId: string | undefined anchorMessageCount?: number maxAssistantTurns?: number }, timeoutMs?: number ): Promise { const syncTiming = getTimingConfig() const maxPollTimeMs = Math.max(timeoutMs ?? getDefaultSyncPollTimeoutMs(), 50) const maxTurns = input.maxAssistantTurns ?? DEFAULT_MAX_ASSISTANT_TURNS const pollStart = Date.now() let inactiveStart = pollStart let pollCount = 0 let timedOut = false let assistantTurnCount = 0 let lastSeenAssistantId: string | undefined log("[task] Starting poll loop", { sessionID: input.sessionID, agentToUse: input.agentToUse, maxTurns }) while (true) { const inactiveElapsedMs = Date.now() - inactiveStart if (inactiveElapsedMs >= maxPollTimeMs) { timedOut = true break } if (ctx.abort?.aborted) { try { const messages = await fetchSessionMessages(client, input.sessionID) const hasNewMessages = input.anchorMessageCount === undefined || messages.length > input.anchorMessageCount if (hasNewMessages && isSessionComplete(messages)) { log("[task] Abort detected after session already completed", { sessionID: input.sessionID }) return null } } catch (error) { log("[task] Final messages fetch failed after abort, continuing with abort", { sessionID: input.sessionID, error: String(error), }) } log("[task] Aborted by user", { sessionID: input.sessionID }) abortSyncSession(client, input.sessionID, "parent_abort") if (input.toastManager && input.taskId) input.toastManager.removeTask(input.taskId) return `Task aborted.\n\nSession ID: ${input.sessionID}` } await wait(syncTiming.POLL_INTERVAL_MS) pollCount++ let statusResult: { data?: Record } try { statusResult = await client.session.status() } catch (error) { log("[task] Poll status fetch failed, retrying", { sessionID: input.sessionID, error: String(error) }) continue } const allStatuses = normalizeSDKResponse(statusResult, {} as Record) const sessionStatus = allStatuses[input.sessionID] if (pollCount % 10 === 0) { log("[task] Poll status", { sessionID: input.sessionID, pollCount, elapsed: Math.floor((Date.now() - pollStart) / 1000) + "s", inactiveElapsed: Math.floor(inactiveElapsedMs / 1000) + "s", sessionStatus: sessionStatus?.type ?? "not_in_status", }) } if (isActiveSessionStatus(sessionStatus)) { inactiveStart = Date.now() continue } let messages: SessionMessage[] try { messages = await fetchSessionMessages(client, input.sessionID) } catch (error) { log("[task] Poll messages fetch failed, retrying", { sessionID: input.sessionID, error: String(error) }) continue } if (input.anchorMessageCount !== undefined && messages.length <= input.anchorMessageCount) { continue } const sessionError = getTerminalSessionError(messages) if (sessionError) { log("[task] Poll detected terminal session error", { sessionID: input.sessionID, sessionError }) return sessionError } if (isSessionComplete(messages)) { log("[task] Poll complete - terminal finish detected", { sessionID: input.sessionID, pollCount }) break } // Count new assistant turns to circuit-break infinite loops const lastAssistant = [...messages].reverse().find((m) => m.info?.role === "assistant") if (lastAssistant?.info?.id && lastAssistant.info.id !== lastSeenAssistantId) { lastSeenAssistantId = lastAssistant.info.id assistantTurnCount++ if (assistantTurnCount >= maxTurns) { log("[task] Max assistant turns reached, aborting to prevent infinite loop", { sessionID: input.sessionID, assistantTurnCount, maxTurns, }) abortSyncSession(client, input.sessionID, "max_turns_exceeded") if (input.toastManager && input.taskId) input.toastManager.removeTask(input.taskId) return `Task aborted: subagent exceeded ${maxTurns} assistant turns without completing. This usually indicates an infinite tool-call loop. Session ID: ${input.sessionID}` } } const hasAssistantText = messages.some((m) => { if (m.info?.role !== "assistant") return false const parts = m.parts ?? [] return parts.some((p) => { if (p.type !== "text" && p.type !== "reasoning") return false const text = (p.text ?? "").trim() return text.length > 0 }) }) if (!lastAssistant?.info?.finish && hasAssistantText) { log("[task] Poll complete - assistant text detected (fallback)", { sessionID: input.sessionID, pollCount, }) break } } if (timedOut) { log("[task] Poll inactivity timeout reached", { sessionID: input.sessionID, pollCount }) abortSyncSession(client, input.sessionID, "poll_timeout") } return timedOut ? `Poll inactivity timeout reached after ${maxPollTimeMs}ms without active OpenCode status for session ${input.sessionID}` : null }