Merge pull request #4235 from code-yeongyu/fix/subagent-timeout-active-output

fix(background-agent): track active subagent output
2026-05-21 16:15:53 +09:00
parent 53cabfe44d dbfde0b056
commit f29411a689
13 changed files with 744 additions and 124 deletions
@@ -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<never>(() => {}))
@@ -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)
@@ -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<unknown> = 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()
@@ -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<PollingManager>(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<PollingManager>(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<PollingManager>(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()
+  })
 })
@@ -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 } })
@@ -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<string, unknown>
-  state?: { status?: string; input?: Record<string, unknown> }
-}
-
 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<void> {
+  private async abortSessionWithLogging(sessionID: string, reason: string): Promise<boolean> {
    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<string>()
@@ -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
@@ -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)
+  })
+})
@@ -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<string, unknown> | undefined
+  readonly state: {
+    readonly status: string | undefined
+    readonly input: Record<string, unknown> | undefined
+  } | undefined
+  readonly field: string | undefined
+  readonly activityTime: Date | undefined
+}
+
+function getStringField(record: Record<string, unknown> | undefined, key: string): string | undefined {
+  const value = record?.[key]
+  return typeof value === "string" && value.length > 0 ? value : undefined
+}
+
+function getRecordField(record: Record<string, unknown> | undefined, key: string): Record<string, unknown> | undefined {
+  const value = record?.[key]
+  return isRecord(value) ? value : undefined
+}
+
+function getDateField(record: Record<string, unknown> | 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<string, unknown> | 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<string, unknown>,
+  fallback: Record<string, unknown> | 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"
+}
@@ -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({
@@ -113,6 +113,64 @@ export function pruneStaleTasksAndNotifications(args: {

 export type SessionStatusMap = Record<string, { type: string }>

+async function interruptStaleTask(args: {
+  task: BackgroundTask
+  client: OpencodeClient
+  concurrencyManager: ConcurrencyManager
+  notifyParentSession: (task: BackgroundTask) => Promise<void>
+  onTaskInterrupted: (task: BackgroundTask) => void
+  sessionID: string
+  reason: string
+  staleMinutes: number
+  timeoutConfigKey: "messageStalenessTimeoutMs" | "sessionGoneTimeoutMs" | "staleTimeoutMs"
+  errorSuffix: string
+  logReason: string
+}): Promise<void> {
+  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<BackgroundTask>
  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<Promise<unknown>> = []

  const messageStalenessMs = config?.messageStalenessTimeoutMs ?? DEFAULT_MESSAGE_STALENESS_TIMEOUT_MS
  const getSessionActivity = args.getSessionActivity
    ?? ((id: string) => getSessionActivityFromClient(client, id, directory))
+  const staleInterruptions: Array<Promise<void>> = []

  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)
  }
 }
@@ -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(() => {})
@@ -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)
  }

@@ -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<BackgroundManager>({
+      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" })