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) 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-session-activity.test.ts b/src/features/background-agent/manager-session-activity.test.ts index 4a7bbf3fe..1e8bc07bd 100644 --- a/src/features/background-agent/manager-session-activity.test.ts +++ b/src/features/background-agent/manager-session-activity.test.ts @@ -151,4 +151,173 @@ 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("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) + 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.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/manager.ts b/src/features/background-agent/manager.ts index bdedc17cc..e13e58aae 100644 --- a/src/features/background-agent/manager.ts +++ b/src/features/background-agent/manager.ts @@ -83,6 +83,13 @@ import { verifySessionExists as verifySessionStillExists, } from "./session-existence" import { handleSessionIdleBackgroundEvent } from "./session-idle-event-handler" +import { + hasOutputSignalFromPart, + isMessagePartForSession, + 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 +151,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 +162,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 @@ -316,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 } } @@ -1457,22 +1449,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 @@ -1515,6 +1506,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) @@ -1523,7 +1515,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 +1529,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 +1552,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 @@ -2194,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 { @@ -2225,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/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 new file mode 100644 index 000000000..7ba5d429b --- /dev/null +++ b/src/features/background-agent/session-stream-activity.ts @@ -0,0 +1,144 @@ +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 | 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 undefined +} + +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"), + } + } + + const type = sessionNextType(eventType) + return { + id: getStringField(props, "callID"), + sessionID, + type, + tool: undefined, + input: undefined, + state: undefined, + field: eventType.endsWith(".delta") ? type : undefined, + activityTime: getDateField(props, "timestamp"), + } +} + +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 + if (partInfo.type === "text" || partInfo.type === "reasoning") return true + + return partInfo.field === "text" || partInfo.field === "reasoning" +} 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) } } 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) } 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" })