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

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

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<string, unknown>
-  state?: { status?: string; input?: Record<string, unknown> }
-}
-
 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<string>()
@@ -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

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<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 {
+  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"
+}