diff --git a/src/features/background-agent/manager-session-activity.test.ts b/src/features/background-agent/manager-session-activity.test.ts index 4a7bbf3fe..229f38f11 100644 --- a/src/features/background-agent/manager-session-activity.test.ts +++ b/src/features/background-agent/manager-session-activity.test.ts @@ -151,4 +151,112 @@ describe("BackgroundManager persisted session activity stale checks", () => { await manager.shutdown() }) + + test("keeps a busy task running when session.next.text.delta refreshes activity", async () => { + //#given - live event progress is stale and session metadata cannot confirm freshness + spyOn(globalThis.Date, "now").mockReturnValue(fixedTime) + let abortCallCount = 0 + const client = { + session: { + status: async () => ({ data: { "ses-active": { type: "busy" } } }), + prompt: async () => ({}), + promptAsync: async () => ({}), + abort: async () => { + abortCallCount += 1 + return {} + }, + todo: async () => ({ data: [] }), + messages: async () => ({ data: [] }), + }, + } + const manager = new BackgroundManager({ + pluginContext: createPluginContext(client), + config: { staleTimeoutMs: 180_000 }, + enableParentSessionNotifications: false, + }) + const task = createRunningTask({ + startedAt: new Date(Date.now() - 45 * 60 * 1000), + progress: { + toolCalls: 3, + lastUpdate: new Date(Date.now() - 45 * 60 * 1000), + }, + }) + const pollingManager = unsafeTestValue(manager) + pollingManager.tasks.set(task.id, task) + + //#when - an OpenCode v2 stream delta arrives before polling checks staleness + manager.handleEvent({ + type: "session.next.text.delta", + properties: { + sessionID: "ses-active", + timestamp: new Date(fixedTime).toISOString(), + delta: "still producing output", + }, + }) + await pollingManager.pollRunningTasks() + + //#then - event activity refresh keeps the task running instead of aborting it + expect(task.status).toBe("running") + expect(task.error).toBeUndefined() + expect(task.progress?.lastUpdate.getTime()).toBe(fixedTime) + expect(abortCallCount).toBe(0) + + await manager.shutdown() + }) + + test("counts session.next.tool.called as activity before stale timeout", async () => { + //#given - live event progress is stale and no tool call has been counted + spyOn(globalThis.Date, "now").mockReturnValue(fixedTime) + let abortCallCount = 0 + const client = { + session: { + status: async () => ({ data: { "ses-active": { type: "busy" } } }), + prompt: async () => ({}), + promptAsync: async () => ({}), + abort: async () => { + abortCallCount += 1 + return {} + }, + todo: async () => ({ data: [] }), + messages: async () => ({ data: [] }), + }, + } + const manager = new BackgroundManager({ + pluginContext: createPluginContext(client), + config: { staleTimeoutMs: 180_000 }, + enableParentSessionNotifications: false, + }) + const task = createRunningTask({ + startedAt: new Date(Date.now() - 45 * 60 * 1000), + progress: { + toolCalls: 0, + lastUpdate: new Date(Date.now() - 45 * 60 * 1000), + }, + }) + const pollingManager = unsafeTestValue(manager) + pollingManager.tasks.set(task.id, task) + + //#when - an OpenCode v2 tool event arrives before polling checks staleness + manager.handleEvent({ + type: "session.next.tool.called", + properties: { + sessionID: "ses-active", + timestamp: new Date(fixedTime).toISOString(), + callID: "call-1", + tool: "bash", + input: { command: "printf ok" }, + }, + }) + await pollingManager.pollRunningTasks() + + //#then - tool activity keeps the task running and increments progress + expect(task.status).toBe("running") + expect(task.error).toBeUndefined() + expect(task.progress?.toolCalls).toBe(1) + expect(task.progress?.lastTool).toBe("bash") + expect(task.progress?.lastUpdate.getTime()).toBe(fixedTime) + expect(abortCallCount).toBe(0) + + await manager.shutdown() + }) }) diff --git a/src/features/background-agent/manager.ts b/src/features/background-agent/manager.ts index bdedc17cc..d1dbf7b1e 100644 --- a/src/features/background-agent/manager.ts +++ b/src/features/background-agent/manager.ts @@ -83,6 +83,12 @@ import { verifySessionExists as verifySessionStillExists, } from "./session-existence" import { handleSessionIdleBackgroundEvent } from "./session-idle-event-handler" +import { + hasOutputSignalFromPart, + resolveMessagePartInfo, + resolveSessionNextPartInfo, + SESSION_NEXT_EVENT_PREFIX, +} from "./session-stream-activity" import { isActiveSessionStatus, isTerminalSessionStatus } from "./session-status-classifier" import { buildFallbackBody, FALLBACK_AGENT, isAgentNotFoundError } from "./spawner" import { @@ -144,15 +150,6 @@ const PARENT_WAKE_TOOL_CALL_DEFER_MAX_MS = 5_000 const PARENT_WAKE_USER_MESSAGE_IN_PROGRESS_WINDOW_MS = 2_000 const PARENT_WAKE_SESSION_ACTIVITY_IN_PROGRESS_WINDOW_MS = 2_000 -interface MessagePartInfo { - id?: string - sessionID?: string - type?: string - tool?: string - input?: Record - state?: { status?: string; input?: Record } -} - interface EventProperties { sessionID?: string info?: { id?: string; sessionID?: string } @@ -164,19 +161,6 @@ interface Event { properties?: EventProperties } -function resolveMessagePartInfo(properties: EventProperties | undefined): MessagePartInfo | undefined { - if (!properties || typeof properties !== "object") { - return undefined - } - - const nestedPart = properties.part - if (nestedPart && typeof nestedPart === "object") { - return nestedPart as MessagePartInfo - } - - return properties as MessagePartInfo -} - interface Todo { content: string status: string @@ -1457,22 +1441,21 @@ The fallback retry session is now created and can be inspected directly. this.observedIncompleteTodosBySession.delete(sessionID) } - private hasOutputSignalFromPart(partInfo: MessagePartInfo | undefined, sessionID?: string): boolean { - if (!partInfo) return false - if (!partInfo.sessionID && !sessionID) return false - if (partInfo.tool) return true - if (partInfo.type === "tool" || partInfo.type === "tool_result") return true - if (partInfo.type === "text" || partInfo.type === "reasoning") return true - - const field = typeof (partInfo as { field?: unknown }).field === "string" - ? (partInfo as { field?: string }).field - : undefined - return field === "text" || field === "reasoning" - } - handleEvent(event: Event): void { const props = event.properties + if (event.type.startsWith(SESSION_NEXT_EVENT_PREFIX)) { + const sessionID = resolveSessionEventID(props) + const partInfo = resolveSessionNextPartInfo(event.type, props) + if (!sessionID || !partInfo) return + + this.handleEvent({ + type: "message.part.updated", + properties: { sessionID, part: partInfo }, + }) + return + } + if (event.type === "message.updated") { const info = props?.info if (!info || typeof info !== "object") return @@ -1523,7 +1506,7 @@ The fallback retry session is now created and can be inspected directly. const { task } = resolved - if (this.hasOutputSignalFromPart(partInfo, sessionID)) { + if (hasOutputSignalFromPart(partInfo, sessionID)) { this.markSessionOutputObserved(sessionID) } @@ -1537,10 +1520,10 @@ The fallback retry session is now created and can be inspected directly. if (!task.progress) { task.progress = { toolCalls: 0, - lastUpdate: new Date(), + lastUpdate: partInfo?.activityTime ?? new Date(), } } - task.progress.lastUpdate = new Date() + task.progress.lastUpdate = partInfo?.activityTime ?? new Date() if (partInfo?.type === "tool" || partInfo?.tool) { const countedToolPartIDs = task.progress.countedToolPartIDs ?? new Set() @@ -1560,34 +1543,34 @@ The fallback retry session is now created and can be inspected directly. task.progress.toolCalls += 1 task.progress.lastTool = partInfo.tool - const circuitBreaker = this.cachedCircuitBreakerSettings ?? resolveCircuitBreakerSettings(this.config) - this.cachedCircuitBreakerSettings = circuitBreaker - if (partInfo.tool) { - const toolInput = partInfo.state?.input ?? partInfo.input - task.progress.toolCallWindow = recordToolCall( - task.progress.toolCallWindow, - partInfo.tool, - circuitBreaker, - toolInput - ) + const circuitBreaker = this.cachedCircuitBreakerSettings ?? resolveCircuitBreakerSettings(this.config) + this.cachedCircuitBreakerSettings = circuitBreaker + if (partInfo.tool) { + const toolInput = partInfo.state?.input ?? partInfo.input + task.progress.toolCallWindow = recordToolCall( + task.progress.toolCallWindow, + partInfo.tool, + circuitBreaker, + toolInput + ) - if (circuitBreaker.enabled) { - const loopDetection = detectRepetitiveToolUse(task.progress.toolCallWindow) - if (loopDetection.triggered) { - log("[background-agent] Circuit breaker: consecutive tool usage detected", { - taskId: task.id, - agent: task.agent, - sessionID, - toolName: loopDetection.toolName, - repeatedCount: loopDetection.repeatedCount, - }) - void this.cancelTask(task.id, { - source: "circuit-breaker", - reason: `Subagent called ${loopDetection.toolName} ${loopDetection.repeatedCount} consecutive times (threshold: ${circuitBreaker.consecutiveThreshold}). This usually indicates an infinite loop. The task was automatically cancelled to prevent excessive token usage.`, - }) - return - } - } + if (circuitBreaker.enabled) { + const loopDetection = detectRepetitiveToolUse(task.progress.toolCallWindow) + if (loopDetection.triggered) { + log("[background-agent] Circuit breaker: consecutive tool usage detected", { + taskId: task.id, + agent: task.agent, + sessionID, + toolName: loopDetection.toolName, + repeatedCount: loopDetection.repeatedCount, + }) + void this.cancelTask(task.id, { + source: "circuit-breaker", + reason: `Subagent called ${loopDetection.toolName} ${loopDetection.repeatedCount} consecutive times (threshold: ${circuitBreaker.consecutiveThreshold}). This usually indicates an infinite loop. The task was automatically cancelled to prevent excessive token usage.`, + }) + return + } + } } const maxToolCalls = circuitBreaker.maxToolCalls diff --git a/src/features/background-agent/session-stream-activity.ts b/src/features/background-agent/session-stream-activity.ts new file mode 100644 index 000000000..2334c4e79 --- /dev/null +++ b/src/features/background-agent/session-stream-activity.ts @@ -0,0 +1,137 @@ +import { isRecord } from "../../shared" + +export const SESSION_NEXT_EVENT_PREFIX = "session.next." + +export interface MessagePartInfo { + readonly id: string | undefined + readonly sessionID: string | undefined + readonly type: string | undefined + readonly tool: string | undefined + readonly input: Record | undefined + readonly state: { + readonly status: string | undefined + readonly input: Record | undefined + } | undefined + readonly field: string | undefined + readonly activityTime: Date | undefined +} + +function getStringField(record: Record | undefined, key: string): string | undefined { + const value = record?.[key] + return typeof value === "string" && value.length > 0 ? value : undefined +} + +function getRecordField(record: Record | undefined, key: string): Record | undefined { + const value = record?.[key] + return isRecord(value) ? value : undefined +} + +function getDateField(record: Record | undefined, key: string): Date | undefined { + const value = record?.[key] + if (value instanceof Date) return Number.isFinite(value.getTime()) ? value : undefined + if (typeof value === "number" && Number.isFinite(value)) return new Date(value) + if (typeof value !== "string") return undefined + + const parsed = new Date(value) + return Number.isFinite(parsed.getTime()) ? parsed : undefined +} + +function resolveState(record: Record | undefined): MessagePartInfo["state"] { + const state = getRecordField(record, "state") + if (!state) return undefined + return { + status: getStringField(state, "status"), + input: getRecordField(state, "input"), + } +} + +function buildPartInfo( + source: Record, + fallback: Record | undefined, +): MessagePartInfo { + return { + id: getStringField(source, "id") ?? getStringField(source, "callID"), + sessionID: getStringField(source, "sessionID") ?? getStringField(fallback, "sessionID"), + type: getStringField(source, "type") ?? getStringField(fallback, "type"), + tool: getStringField(source, "tool") ?? getStringField(fallback, "tool"), + input: getRecordField(source, "input") ?? getRecordField(fallback, "input"), + state: resolveState(source) ?? resolveState(fallback), + field: getStringField(source, "field") ?? getStringField(fallback, "field"), + activityTime: getDateField(source, "activityTime") + ?? getDateField(source, "timestamp") + ?? getDateField(fallback, "activityTime") + ?? getDateField(fallback, "timestamp"), + } +} + +export function resolveMessagePartInfo(properties: unknown): MessagePartInfo | undefined { + const props = isRecord(properties) ? properties : undefined + if (!props) return undefined + + const nestedPart = getRecordField(props, "part") + return nestedPart ? buildPartInfo(nestedPart, props) : buildPartInfo(props, undefined) +} + +function sessionNextType(eventType: string): string { + if (eventType.startsWith("session.next.reasoning.")) return "reasoning" + if (eventType.startsWith("session.next.tool.") && eventType !== "session.next.tool.called") return "tool_result" + return "text" +} + +function isTrackedSessionNextActivityEvent(eventType: string): boolean { + return eventType === "session.next.synthetic" + || eventType === "session.next.retried" + || eventType.startsWith("session.next.shell.") + || eventType.startsWith("session.next.step.") + || eventType.startsWith("session.next.text.") + || eventType.startsWith("session.next.reasoning.") + || eventType.startsWith("session.next.tool.") + || eventType.startsWith("session.next.compaction.") +} + +export function resolveSessionNextPartInfo(eventType: string, properties: unknown): MessagePartInfo | undefined { + if (!eventType.startsWith(SESSION_NEXT_EVENT_PREFIX)) return undefined + if (!isTrackedSessionNextActivityEvent(eventType)) return undefined + + const props = isRecord(properties) ? properties : undefined + const sessionID = getStringField(props, "sessionID") + if (!props || !sessionID) return undefined + + const input = getRecordField(props, "input") + if (eventType === "session.next.tool.called") { + return { + id: getStringField(props, "callID"), + sessionID, + type: "tool", + tool: getStringField(props, "tool"), + input, + state: { + status: "running", + input, + }, + field: undefined, + activityTime: getDateField(props, "timestamp"), + } + } + + return { + id: getStringField(props, "callID"), + sessionID, + type: sessionNextType(eventType), + tool: undefined, + input: undefined, + state: undefined, + field: eventType.endsWith(".delta") ? sessionNextType(eventType) : undefined, + activityTime: getDateField(props, "timestamp"), + } +} + +export function hasOutputSignalFromPart(partInfo: MessagePartInfo | undefined, sessionID?: string): boolean { + if (!partInfo) return false + if (!partInfo.sessionID && !sessionID) return false + if (partInfo.tool) return true + if (partInfo.type === "tool" || partInfo.type === "tool_result") return true + if (partInfo.type === "text" || partInfo.type === "reasoning") return true + + return partInfo.field === "text" || partInfo.field === "reasoning" +}