From b68af25e41bd87bbbc0f5dc7c161aee76a6cfa82 Mon Sep 17 00:00:00 2001 From: YeonGyu-Kim Date: Thu, 21 May 2026 15:23:26 +0900 Subject: [PATCH 1/6] 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" +} From 0bf8a9df25d11e04e6436fed0d8812faf838c7d3 Mon Sep 17 00:00:00 2001 From: YeonGyu-Kim Date: Thu, 21 May 2026 15:23:52 +0900 Subject: [PATCH 2/6] fix(background-agent): fail abort on SDK errors Treat resolved abort responses with a non-null error payload the same as rejected aborts. This prevents stale-timeout cancellation bookkeeping from reporting success when the child session was not actually aborted. Plan: .omo/plans/subagent-timeout-active-output.md --- .../abort-with-timeout.test.ts | 16 ++++++++++ .../background-agent/abort-with-timeout.ts | 30 +++++++++++++++++-- 2 files changed, 44 insertions(+), 2 deletions(-) diff --git a/src/features/background-agent/abort-with-timeout.test.ts b/src/features/background-agent/abort-with-timeout.test.ts index 3655a4172..46b34e917 100644 --- a/src/features/background-agent/abort-with-timeout.test.ts +++ b/src/features/background-agent/abort-with-timeout.test.ts @@ -37,6 +37,22 @@ describe("abortWithTimeout", () => { expect(logMock).not.toHaveBeenCalled() }) + test("#given abort resolves with an SDK error response #when abortWithTimeout runs #then it reports cancellation failure", async () => { + // given + const error = { message: "session not found" } + const abort = mock(async () => ({ error })) + + // when + const result = await abortWithTimeout(createClient(abort), "session-error-response", 10) + + // then + expect(result).toBe(false) + expect(logMock).toHaveBeenCalledWith( + "[background-agent] Session abort returned an error response:", + { sessionID: "session-error-response", error }, + ) + }) + test("#given abort hangs indefinitely #when abortWithTimeout runs #then it logs warning and continues", async () => { // given const abort = mock(() => new Promise(() => {})) diff --git a/src/features/background-agent/abort-with-timeout.ts b/src/features/background-agent/abort-with-timeout.ts index 49f1170f2..9e4d9d7a6 100644 --- a/src/features/background-agent/abort-with-timeout.ts +++ b/src/features/background-agent/abort-with-timeout.ts @@ -1,6 +1,13 @@ import { log } from "../../shared" +import { isRecord } from "../../shared/record-type-guard" import type { OpencodeClient } from "./opencode-client" +function getAbortResponseError(response: unknown): unknown | undefined { + if (!isRecord(response)) return undefined + const error = response.error + return error === undefined || error === null ? undefined : error +} + export async function abortWithTimeout( client: OpencodeClient, sessionID: string, @@ -10,7 +17,26 @@ export async function abortWithTimeout( try { const result = await Promise.race([ - client.session.abort({ path: { id: sessionID } }).then(() => "aborted" as const), + client.session.abort({ path: { id: sessionID } }).then( + (response) => { + const error = getAbortResponseError(response) + if (error !== undefined) { + log("[background-agent] Session abort returned an error response:", { + sessionID, + error, + }) + return "failed" as const + } + return "aborted" as const + }, + (error) => { + log("[background-agent] Session abort failed:", { + sessionID, + error, + }) + return "failed" as const + }, + ), new Promise<"timed_out">((resolve) => { timeoutHandle = setTimeout(() => { resolve("timed_out") @@ -26,7 +52,7 @@ export async function abortWithTimeout( return false } - return true + return result === "aborted" } finally { if (timeoutHandle) { clearTimeout(timeoutHandle) From bd1a6e3d3bb0b6179f56cf7d5e340523882c5d0c Mon Sep 17 00:00:00 2001 From: YeonGyu-Kim Date: Thu, 21 May 2026 15:49:58 +0900 Subject: [PATCH 3/6] fix(background-agent): forward session stream activity --- .../manager-session-activity.test.ts | 61 +++++++++++++++++++ src/features/background-agent/manager.ts | 2 + .../session-stream-activity.ts | 5 ++ .../background-notification/hook.test.ts | 17 ++++++ src/hooks/background-notification/hook.ts | 9 ++- 5 files changed, 93 insertions(+), 1 deletion(-) diff --git a/src/features/background-agent/manager-session-activity.test.ts b/src/features/background-agent/manager-session-activity.test.ts index 229f38f11..1e8bc07bd 100644 --- a/src/features/background-agent/manager-session-activity.test.ts +++ b/src/features/background-agent/manager-session-activity.test.ts @@ -204,6 +204,67 @@ describe("BackgroundManager persisted session activity stale checks", () => { await manager.shutdown() }) + test("ignores nested message part activity from a different session", async () => { + //#given - live event progress is stale and a nested part belongs to another session + spyOn(globalThis.Date, "now").mockReturnValue(fixedTime) + const staleTime = fixedTime - 45 * 60 * 1000 + let abortCallCount = 0 + const client = { + session: { + status: async () => ({ data: { "ses-active": { type: "busy" } } }), + get: async () => ({ + data: { + id: "ses-active", + time: { updated: staleTime }, + }, + }), + 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(staleTime), + progress: { + toolCalls: 3, + lastUpdate: new Date(staleTime), + }, + }) + const pollingManager = unsafeTestValue(manager) + pollingManager.tasks.set(task.id, task) + + //#when - an inconsistent event carries a fresh part for a different session + manager.handleEvent({ + type: "message.part.updated", + properties: { + sessionID: "ses-active", + part: { + sessionID: "ses-other", + type: "text", + activityTime: new Date(fixedTime).toISOString(), + }, + }, + }) + await pollingManager.pollRunningTasks() + + //#then - the wrong-session part does not refresh activity or prevent stale cancellation + expect(task.status).toBe("cancelled") + expect(task.progress?.lastUpdate.getTime()).toBe(staleTime) + expect(abortCallCount).toBe(1) + + 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) diff --git a/src/features/background-agent/manager.ts b/src/features/background-agent/manager.ts index d1dbf7b1e..d5fa9533d 100644 --- a/src/features/background-agent/manager.ts +++ b/src/features/background-agent/manager.ts @@ -85,6 +85,7 @@ import { import { handleSessionIdleBackgroundEvent } from "./session-idle-event-handler" import { hasOutputSignalFromPart, + isMessagePartForSession, resolveMessagePartInfo, resolveSessionNextPartInfo, SESSION_NEXT_EVENT_PREFIX, @@ -1498,6 +1499,7 @@ The fallback retry session is now created and can be inspected directly. const partInfo = resolveMessagePartInfo(props) const sessionID = resolveMessageEventSessionID(props) if (!sessionID) return + if (!isMessagePartForSession(partInfo, sessionID)) return this.clearDispatchedParentWake(sessionID) this.parentWakeNotifier.recordParentSessionActivity(sessionID) diff --git a/src/features/background-agent/session-stream-activity.ts b/src/features/background-agent/session-stream-activity.ts index 2334c4e79..ad6c1f0a4 100644 --- a/src/features/background-agent/session-stream-activity.ts +++ b/src/features/background-agent/session-stream-activity.ts @@ -126,8 +126,13 @@ export function resolveSessionNextPartInfo(eventType: string, properties: unknow } } +export function isMessagePartForSession(partInfo: MessagePartInfo | undefined, sessionID: string): boolean { + return !partInfo?.sessionID || partInfo.sessionID === sessionID +} + export function hasOutputSignalFromPart(partInfo: MessagePartInfo | undefined, sessionID?: string): boolean { if (!partInfo) return false + if (partInfo.sessionID && sessionID && partInfo.sessionID !== sessionID) return false if (!partInfo.sessionID && !sessionID) return false if (partInfo.tool) return true if (partInfo.type === "tool" || partInfo.type === "tool_result") return true diff --git a/src/hooks/background-notification/hook.test.ts b/src/hooks/background-notification/hook.test.ts index f32ce14c1..3887092e0 100644 --- a/src/hooks/background-notification/hook.test.ts +++ b/src/hooks/background-notification/hook.test.ts @@ -35,6 +35,23 @@ describe("createBackgroundNotificationHook", () => { expect(handleEvent).toHaveBeenCalledWith(event) }) + test("#given session.next stream event #when event handler runs #then it forwards to manager", async () => { + //#given + const handleEvent = mock(() => {}) + const hook = createBackgroundNotificationHook({ + handleEvent, + injectPendingNotificationsIntoChatMessage: () => {}, + } as never) + + const event = { type: "session.next.text.delta", properties: { sessionID: "ses-1", delta: "x" } } + + //#when + await hook.event({ event }) + + //#then + expect(handleEvent).toHaveBeenCalledWith(event) + }) + test("#given todo.updated event #when event handler runs #then it forwards to manager", async () => { //#given const handleEvent = mock(() => {}) diff --git a/src/hooks/background-notification/hook.ts b/src/hooks/background-notification/hook.ts index d963fa4f6..f83939757 100644 --- a/src/hooks/background-notification/hook.ts +++ b/src/hooks/background-notification/hook.ts @@ -28,9 +28,16 @@ const FORWARDED_EVENT_TYPES = new Set([ "session.status", ]) +const FORWARDED_EVENT_PREFIXES = ["session.next."] + +function shouldForwardEvent(type: string): boolean { + return FORWARDED_EVENT_TYPES.has(type) + || FORWARDED_EVENT_PREFIXES.some((prefix) => type.startsWith(prefix)) +} + export function createBackgroundNotificationHook(manager: BackgroundManager) { const eventHandler = async ({ event }: EventInput) => { - if (!FORWARDED_EVENT_TYPES.has(event.type)) return + if (!shouldForwardEvent(event.type)) return manager.handleEvent(event) } From 6d15ab86ab41872506e097ba06f6672327a47f5a Mon Sep 17 00:00:00 2001 From: YeonGyu-Kim Date: Thu, 21 May 2026 15:50:28 +0900 Subject: [PATCH 4/6] fix(background-agent): fail cancellation when abort fails --- .../cancel-task-cleanup.test.ts | 32 +++++++++++++++++-- src/features/background-agent/manager.ts | 26 +++++++++------ src/tools/background-task/tools.test.ts | 18 +++++++++++ 3 files changed, 64 insertions(+), 12 deletions(-) diff --git a/src/features/background-agent/cancel-task-cleanup.test.ts b/src/features/background-agent/cancel-task-cleanup.test.ts index 19bb03351..11e64d12a 100644 --- a/src/features/background-agent/cancel-task-cleanup.test.ts +++ b/src/features/background-agent/cancel-task-cleanup.test.ts @@ -11,11 +11,14 @@ afterEach(() => { while (managersToShutdown.length > 0) managersToShutdown.pop()?.shutdown() }) -function createBackgroundManager(config?: { defaultConcurrency?: number }): BackgroundManager { +function createBackgroundManager( + config?: { defaultConcurrency?: number }, + abortSession: () => Promise = async () => ({ data: true }), +): BackgroundManager { const directory = tmpdir() const client = { session: {} as PluginInput["client"]["session"] } as PluginInput["client"] - Reflect.set(client.session, "abort", async () => ({ data: true })) + Reflect.set(client.session, "abort", abortSession) Reflect.set(client.session, "create", async () => ({ data: { id: `session-${crypto.randomUUID().slice(0, 8)}` } })) Reflect.set(client.session, "get", async () => ({ data: { directory } })) Reflect.set(client.session, "messages", async () => ({ data: [] })) @@ -111,6 +114,31 @@ describe("BackgroundManager.cancelTask cleanup", () => { expect(manager.getTask(task.id)?.sessionId).toBe(task.sessionId) }) + test("#given running task abort returns SDK error #when cancelTask runs #then cancellation fails and task stays running", async () => { + // given + const manager = createBackgroundManager(undefined, async () => ({ error: { message: "session still active" } })) + const task = createMockTask({ + id: "task-abort-error", + parentSessionId: "parent-session-abort-error", + sessionId: "session-abort-error", + }) + + getTaskMap(manager).set(task.id, task) + getPendingByParent(manager).set(task.parentSessionId, new Set([task.id])) + + // when + const cancelled = await manager.cancelTask(task.id, { + skipNotification: true, + source: "test", + }) + + // then + expect(cancelled).toBe(false) + expect(task.status).toBe("running") + expect(getTaskMap(manager).get(task.id)).toBe(task) + expect(getPendingByParent(manager).get(task.parentSessionId)).toEqual(new Set([task.id])) + }) + test("#given a running task #when cancelTask called with skipNotification=false #then task is also eventually removed", async () => { // given const manager = createBackgroundManager() diff --git a/src/features/background-agent/manager.ts b/src/features/background-agent/manager.ts index d5fa9533d..e13e58aae 100644 --- a/src/features/background-agent/manager.ts +++ b/src/features/background-agent/manager.ts @@ -301,14 +301,21 @@ export class BackgroundManager { this.registerProcessCleanup() } - private async abortSessionWithLogging(sessionID: string, reason: string): Promise { + private async abortSessionWithLogging(sessionID: string, reason: string): Promise { try { - await abortWithTimeout(this.client, sessionID) + const aborted = await abortWithTimeout(this.client, sessionID) + if (!aborted) { + log(`[background-agent] Session abort did not complete during ${reason}:`, { + sessionID, + }) + } + return aborted } catch (error) { log(`[background-agent] Failed to abort session during ${reason}:`, { sessionID, error, }) + return false } } @@ -2179,6 +2186,13 @@ The task was re-queued on a fallback model after a retryable failure. } const wasRunning = task.status === "running" + if (wasRunning && abortSession && task.sessionId) { + const aborted = await this.abortSessionWithLogging(task.sessionId, `task cancellation (${source})`) + if (!aborted) return false + + clearDelegatedChildSessionBootstrap(task.sessionId) + SessionCategoryRegistry.remove(task.sessionId) + } if (task.currentAttemptID) { finalizeAttempt(task, task.currentAttemptID, "cancelled", reason) } else { @@ -2210,14 +2224,6 @@ The task was re-queued on a fallback model after a retryable failure. this.idleDeferralTimers.delete(task.id) } - if (abortSession && task.sessionId) { - // Awaited to prevent dangling promise during subagent teardown (Bun/WebKit SIGABRT) - await this.abortSessionWithLogging(task.sessionId, `task cancellation (${source})`) - - clearDelegatedChildSessionBootstrap(task.sessionId) - SessionCategoryRegistry.remove(task.sessionId) - } - removeTaskToastTracking(task.id) // Update continuation marker for CLI run mode diff --git a/src/tools/background-task/tools.test.ts b/src/tools/background-task/tools.test.ts index 81969b431..045230623 100644 --- a/src/tools/background-task/tools.test.ts +++ b/src/tools/background-task/tools.test.ts @@ -408,6 +408,24 @@ describe("background_cancel", () => { expect(output).toContain("Task cancelled successfully") }) + test("reports an error when manager cannot cancel a running task", async () => { + // #given + const task = createTask({ status: "running" }) + const manager = unsafeTestValue({ + getTask: (id: string) => (id === task.id ? task : undefined), + getAllDescendantTasks: () => [task], + cancelTask: async () => false, + }) + const client = { session: { abort: async () => ({}) } } as BackgroundCancelClient + const tool = createBackgroundCancel(manager, client) + + // #when + const output = await tool.execute({ taskId: task.id }, mockContext) + + // #then + expect(output).toContain(`[ERROR] Failed to cancel task: ${task.id}`) + }) + test("cancels all running or pending tasks", async () => { // #given const taskA = createTask({ id: "task-a", status: "running" }) From 282010f97d398ed698ca33c8592d2eccaf64133e Mon Sep 17 00:00:00 2001 From: YeonGyu-Kim Date: Thu, 21 May 2026 15:50:44 +0900 Subject: [PATCH 5/6] fix(background-agent): gate stale timeout on abort success --- src/features/background-agent/manager.test.ts | 4 +- .../background-agent/task-poller.test.ts | 115 +++++++++++++++ src/features/background-agent/task-poller.ts | 132 ++++++++++++------ 3 files changed, 207 insertions(+), 44 deletions(-) diff --git a/src/features/background-agent/manager.test.ts b/src/features/background-agent/manager.test.ts index d5e6e99ab..8098cdf26 100644 --- a/src/features/background-agent/manager.test.ts +++ b/src/features/background-agent/manager.test.ts @@ -4821,9 +4821,7 @@ describe("BackgroundManager.checkAndInterruptStaleTasks", () => { prompt: async () => ({}), promptAsync: async () => ({}), abort: async () => ({}), - get: async () => { - throw new Error("missing") - }, + get: async () => ({ data: { id: "session-running", time: { updated: fixedTime - 300_000 } } }), }, } const manager = new BackgroundManager({ pluginContext: createPluginInput(client), config: { staleTimeoutMs: 180_000 } }) diff --git a/src/features/background-agent/task-poller.test.ts b/src/features/background-agent/task-poller.test.ts index da4d2da1e..19038cca5 100644 --- a/src/features/background-agent/task-poller.test.ts +++ b/src/features/background-agent/task-poller.test.ts @@ -180,6 +180,36 @@ describe("checkAndInterruptStaleTasks", () => { expect(task.error).toContain("messageStalenessTimeoutMs") }) + it("should keep never-updated task running when stale abort returns SDK error", async () => { + //#given + const task = createRunningTask({ + startedAt: new Date(Date.now() - 15 * 60 * 1000), + progress: undefined, + concurrencyKey: "anthropic/claude-opus-4-7", + }) + const releaseMock = mock(() => {}) + const onTaskInterrupted = mock(() => {}) + mockClient.session.abort.mockImplementationOnce(() => Promise.resolve({ error: { message: "still running" } })) + + //#when + await checkAndInterruptStaleTasks({ + tasks: [task], + client: mockClient as never, + config: { messageStalenessTimeoutMs: 600_000 }, + concurrencyManager: { release: releaseMock } as never, + notifyParentSession: mockNotify, + onTaskInterrupted, + }) + + //#then + expect(task.status).toBe("running") + expect(task.error).toBeUndefined() + expect(task.concurrencyKey).toBe("anthropic/claude-opus-4-7") + expect(releaseMock).not.toHaveBeenCalled() + expect(onTaskInterrupted).not.toHaveBeenCalled() + expect(mockNotify).not.toHaveBeenCalled() + }) + it("should await abort before resolving for no-progress stale interruption", async () => { //#given const task = createRunningTask({ @@ -303,6 +333,91 @@ describe("checkAndInterruptStaleTasks", () => { expect(task.error).toContain("Stale timeout") }) + it("should keep stale-progress task running when abort returns SDK error", async () => { + //#given + const task = createRunningTask({ + startedAt: new Date(Date.now() - 900_000), + progress: { + toolCalls: 2, + lastUpdate: new Date(Date.now() - 900_000), + }, + concurrencyKey: "anthropic/claude-opus-4-7", + }) + const releaseMock = mock(() => {}) + const onTaskInterrupted = mock(() => {}) + mockClient.session.abort.mockImplementationOnce(() => Promise.resolve({ error: { message: "still running" } })) + + //#when + await checkAndInterruptStaleTasks({ + tasks: [task], + client: mockClient as never, + config: { staleTimeoutMs: 180_000, messageStalenessTimeoutMs: 600_000 }, + concurrencyManager: { release: releaseMock } as never, + notifyParentSession: mockNotify, + sessionStatuses: { "ses-1": { type: "busy" } }, + onTaskInterrupted, + }) + + //#then + expect(task.status).toBe("running") + expect(task.error).toBeUndefined() + expect(task.concurrencyKey).toBe("anthropic/claude-opus-4-7") + expect(releaseMock).not.toHaveBeenCalled() + expect(onTaskInterrupted).not.toHaveBeenCalled() + expect(mockNotify).not.toHaveBeenCalled() + }) + + it("should abort multiple stale-progress tasks concurrently before marking them cancelled", async () => { + //#given + const firstAbort = createDeferredPromise() + const secondAbort = createDeferredPromise() + const abortSessionIDs: string[] = [] + const taskA = createRunningTask({ + id: "task-stale-a", + sessionId: "ses-stale-a", + parentSessionId: "parent-stale-a", + progress: { + toolCalls: 1, + lastUpdate: new Date(Date.now() - 900_000), + }, + }) + const taskB = createRunningTask({ + id: "task-stale-b", + sessionId: "ses-stale-b", + parentSessionId: "parent-stale-b", + progress: { + toolCalls: 1, + lastUpdate: new Date(Date.now() - 900_000), + }, + }) + mockClient.session.abort.mockImplementation(({ path }: { path: { id: string } }) => { + abortSessionIDs.push(path.id) + return path.id === "ses-stale-a" ? firstAbort.promise : secondAbort.promise + }) + + //#when + const interruption = checkAndInterruptStaleTasks({ + tasks: [taskA, taskB], + client: mockClient as never, + config: { staleTimeoutMs: 180_000 }, + concurrencyManager: mockConcurrencyManager as never, + notifyParentSession: mockNotify, + }) + await Promise.resolve() + + //#then + expect(abortSessionIDs).toEqual(["ses-stale-a", "ses-stale-b"]) + expect(taskA.status).toBe("running") + expect(taskB.status).toBe("running") + + firstAbort.resolve() + secondAbort.resolve() + await interruption + + expect(taskA.status).toBe("cancelled") + expect(taskB.status).toBe("cancelled") + }) + it("should NOT interrupt busy session with no progress within message staleness timeout", async () => { //#given - task has no progress yet, but it is still inside the configured first-progress window const task = createRunningTask({ diff --git a/src/features/background-agent/task-poller.ts b/src/features/background-agent/task-poller.ts index d80a16cfa..9537e43da 100644 --- a/src/features/background-agent/task-poller.ts +++ b/src/features/background-agent/task-poller.ts @@ -113,6 +113,64 @@ export function pruneStaleTasksAndNotifications(args: { export type SessionStatusMap = Record +async function interruptStaleTask(args: { + task: BackgroundTask + client: OpencodeClient + concurrencyManager: ConcurrencyManager + notifyParentSession: (task: BackgroundTask) => Promise + onTaskInterrupted: (task: BackgroundTask) => void + sessionID: string + reason: string + staleMinutes: number + timeoutConfigKey: "messageStalenessTimeoutMs" | "sessionGoneTimeoutMs" | "staleTimeoutMs" + errorSuffix: string + logReason: string +}): Promise { + const { + task, + client, + concurrencyManager, + notifyParentSession, + onTaskInterrupted, + sessionID, + reason, + staleMinutes, + timeoutConfigKey, + errorSuffix, + logReason, + } = args + + const aborted = await abortWithTimeout(client, sessionID) + if (!aborted) { + log("[background-agent] Task stale interruption skipped because session abort failed:", { + taskId: task.id, + sessionID, + reason, + }) + return + } + + if (task.status !== "running" || task.sessionId !== sessionID) return + + task.status = "cancelled" + task.error = `Stale timeout (${reason} for ${staleMinutes}min${errorSuffix}). This is a FINAL cancellation - do NOT create a replacement task. If the timeout is too short, increase 'background_task.${timeoutConfigKey}' in .opencode/${CONFIG_BASENAME}.json.` + task.completedAt = new Date() + + if (task.concurrencyKey) { + concurrencyManager.release(task.concurrencyKey) + task.concurrencyKey = undefined + } + + onTaskInterrupted(task) + log(`[background-agent] Task ${task.id} interrupted: ${logReason}`) + + try { + await notifyParentSession(task) + } catch (err) { + log("[background-agent] Error in notifyParentSession for stale task:", { taskId: task.id, error: err }) + } +} + export async function checkAndInterruptStaleTasks(args: { tasks: Iterable client: OpencodeClient @@ -137,11 +195,11 @@ export async function checkAndInterruptStaleTasks(args: { const staleTimeoutMs = config?.staleTimeoutMs ?? DEFAULT_STALE_TIMEOUT_MS const sessionGoneTimeoutMs = config?.sessionGoneTimeoutMs ?? DEFAULT_SESSION_GONE_TIMEOUT_MS const now = Date.now() - const abortPromises: Array> = [] const messageStalenessMs = config?.messageStalenessTimeoutMs ?? DEFAULT_MESSAGE_STALENESS_TIMEOUT_MS const getSessionActivity = args.getSessionActivity ?? ((id: string) => getSessionActivityFromClient(client, id, directory)) + const staleInterruptions: Array> = [] for (const task of tasks) { if (task.status !== "running") continue @@ -189,25 +247,21 @@ export async function checkAndInterruptStaleTasks(args: { const staleMinutes = Math.round(runtime / 60000) const reason = sessionGone ? "session gone from status registry" : "no activity" - task.status = "cancelled" - task.error = `Stale timeout (${reason} for ${staleMinutes}min since start). This is a FINAL cancellation - do NOT create a replacement task. If the timeout is too short, increase 'background_task.${sessionGone ? "sessionGoneTimeoutMs" : "messageStalenessTimeoutMs"}' in .opencode/${CONFIG_BASENAME}.json.` - task.completedAt = new Date() - - if (task.concurrencyKey) { - concurrencyManager.release(task.concurrencyKey) - task.concurrencyKey = undefined - } - - onTaskInterrupted(task) - - abortPromises.push(abortWithTimeout(client, sessionID)) - log(`[background-agent] Task ${task.id} interrupted: no progress since start`) - - try { - await notifyParentSession(task) - } catch (err) { - log("[background-agent] Error in notifyParentSession for stale task:", { taskId: task.id, error: err }) - } + staleInterruptions.push( + interruptStaleTask({ + task, + client, + concurrencyManager, + notifyParentSession, + onTaskInterrupted, + sessionID, + reason, + staleMinutes, + timeoutConfigKey: sessionGone ? "sessionGoneTimeoutMs" : "messageStalenessTimeoutMs", + errorSuffix: " since start", + logReason: "no progress since start", + }), + ) continue } @@ -243,28 +297,24 @@ export async function checkAndInterruptStaleTasks(args: { const staleMinutes = Math.round(timeSinceLastUpdate / 60000) const reason = sessionGone ? "session gone from status registry" : "no activity" - task.status = "cancelled" - task.error = `Stale timeout (${reason} for ${staleMinutes}min). This is a FINAL cancellation - do NOT create a replacement task. If the timeout is too short, increase 'background_task.${sessionGone ? "sessionGoneTimeoutMs" : "staleTimeoutMs"}' in .opencode/${CONFIG_BASENAME}.json.` - task.completedAt = new Date() - - if (task.concurrencyKey) { - concurrencyManager.release(task.concurrencyKey) - task.concurrencyKey = undefined - } - - onTaskInterrupted(task) - - abortPromises.push(abortWithTimeout(client, sessionID)) - log(`[background-agent] Task ${task.id} interrupted: stale timeout`) - - try { - await notifyParentSession(task) - } catch (err) { - log("[background-agent] Error in notifyParentSession for stale task:", { taskId: task.id, error: err }) - } + staleInterruptions.push( + interruptStaleTask({ + task, + client, + concurrencyManager, + notifyParentSession, + onTaskInterrupted, + sessionID, + reason, + staleMinutes, + timeoutConfigKey: sessionGone ? "sessionGoneTimeoutMs" : "staleTimeoutMs", + errorSuffix: "", + logReason: "stale timeout", + }), + ) } - if (abortPromises.length > 0) { - await Promise.allSettled(abortPromises) + if (staleInterruptions.length > 0) { + await Promise.all(staleInterruptions) } } From dbfde0b05628b09185eba56327b571198f9f4785 Mon Sep 17 00:00:00 2001 From: YeonGyu-Kim Date: Thu, 21 May 2026 16:12:53 +0900 Subject: [PATCH 6/6] fix(background-agent): ignore metadata stream output --- .../session-stream-activity.test.ts | 41 +++++++++++++++++++ .../session-stream-activity.ts | 10 +++-- 2 files changed, 47 insertions(+), 4 deletions(-) create mode 100644 src/features/background-agent/session-stream-activity.test.ts diff --git a/src/features/background-agent/session-stream-activity.test.ts b/src/features/background-agent/session-stream-activity.test.ts new file mode 100644 index 000000000..76e4751c1 --- /dev/null +++ b/src/features/background-agent/session-stream-activity.test.ts @@ -0,0 +1,41 @@ +import { describe, expect, test } from "bun:test" +import { + hasOutputSignalFromPart, + resolveSessionNextPartInfo, +} from "./session-stream-activity" + +describe("session.next stream activity", () => { + test("#given text delta event #when resolving part info #then it counts as output activity", () => { + // given + const timestamp = "2026-05-21T03:00:00.000Z" + + // when + const partInfo = resolveSessionNextPartInfo("session.next.text.delta", { + sessionID: "ses-active", + timestamp, + }) + + // then + expect(partInfo?.type).toBe("text") + expect(partInfo?.field).toBe("text") + expect(partInfo?.activityTime).toEqual(new Date(timestamp)) + expect(hasOutputSignalFromPart(partInfo, "ses-active")).toBe(true) + }) + + test("#given metadata stream event #when resolving part info #then it refreshes activity without counting as output", () => { + // given + const timestamp = "2026-05-21T03:00:00.000Z" + + // when + const partInfo = resolveSessionNextPartInfo("session.next.compaction.started", { + sessionID: "ses-active", + timestamp, + }) + + // then + expect(partInfo?.type).toBeUndefined() + expect(partInfo?.field).toBeUndefined() + expect(partInfo?.activityTime).toEqual(new Date(timestamp)) + expect(hasOutputSignalFromPart(partInfo, "ses-active")).toBe(false) + }) +}) diff --git a/src/features/background-agent/session-stream-activity.ts b/src/features/background-agent/session-stream-activity.ts index ad6c1f0a4..7ba5d429b 100644 --- a/src/features/background-agent/session-stream-activity.ts +++ b/src/features/background-agent/session-stream-activity.ts @@ -72,10 +72,11 @@ export function resolveMessagePartInfo(properties: unknown): MessagePartInfo | u return nestedPart ? buildPartInfo(nestedPart, props) : buildPartInfo(props, undefined) } -function sessionNextType(eventType: string): string { +function sessionNextType(eventType: string): string | undefined { + if (eventType.startsWith("session.next.text.")) return "text" if (eventType.startsWith("session.next.reasoning.")) return "reasoning" if (eventType.startsWith("session.next.tool.") && eventType !== "session.next.tool.called") return "tool_result" - return "text" + return undefined } function isTrackedSessionNextActivityEvent(eventType: string): boolean { @@ -114,14 +115,15 @@ export function resolveSessionNextPartInfo(eventType: string, properties: unknow } } + const type = sessionNextType(eventType) return { id: getStringField(props, "callID"), sessionID, - type: sessionNextType(eventType), + type, tool: undefined, input: undefined, state: undefined, - field: eventType.endsWith(".delta") ? sessionNextType(eventType) : undefined, + field: eventType.endsWith(".delta") ? type : undefined, activityTime: getDateField(props, "timestamp"), } }