From b68af25e41bd87bbbc0f5dc7c161aee76a6cfa82 Mon Sep 17 00:00:00 2001 From: YeonGyu-Kim Date: Thu, 21 May 2026 15:23:26 +0900 Subject: [PATCH] fix(background-agent): track session.next activity Convert OpenCode v2 session.next stream events into the existing message part activity path so child sessions that are still producing text, reasoning, or tool output refresh lastUpdate before stale polling runs. This keeps the timeout poller from cancelling active subagents and preserves tool-call progress for session.next.tool.called events. Plan: .omo/plans/subagent-timeout-active-output.md --- .../manager-session-activity.test.ts | 108 ++++++++++++++ src/features/background-agent/manager.ts | 113 ++++++--------- .../session-stream-activity.ts | 137 ++++++++++++++++++ 3 files changed, 293 insertions(+), 65 deletions(-) create mode 100644 src/features/background-agent/session-stream-activity.ts 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" +}