fix(background-agent): defer active parent wakes
This commit is contained in:
@@ -63,6 +63,10 @@ import {
|
||||
} from "./attempt-lifecycle"
|
||||
import { registerManagerForCleanup, unregisterManagerForCleanup } from "./process-cleanup"
|
||||
import { setContinuationMarkerSource } from "../../features/run-continuation-state"
|
||||
import {
|
||||
isSessionActive as isOpenCodeSessionActive,
|
||||
settleAfterSessionIdle,
|
||||
} from "../../hooks/shared/session-idle-settle"
|
||||
import {
|
||||
findNearestMessageExcludingCompaction,
|
||||
resolvePromptContextFromSessionMessages,
|
||||
@@ -100,6 +104,13 @@ type ParentWakePromptContext = {
|
||||
tools?: Record<string, boolean>
|
||||
}
|
||||
|
||||
type PendingParentWake = {
|
||||
promptContext: ParentWakePromptContext
|
||||
notifications: string[]
|
||||
}
|
||||
|
||||
const PENDING_PARENT_WAKE_RETRY_MS = 1_000
|
||||
|
||||
interface MessagePartInfo {
|
||||
id?: string
|
||||
sessionID?: string
|
||||
@@ -221,6 +232,8 @@ export class BackgroundManager {
|
||||
private completedTaskSummaries: Map<string, BackgroundTaskNotificationTask[]> = new Map()
|
||||
private idleDeferralTimers: Map<string, ReturnType<typeof setTimeout>> = new Map()
|
||||
private notificationQueueByParent: Map<string, Promise<void>> = new Map()
|
||||
private pendingParentWakes: Map<string, PendingParentWake> = new Map()
|
||||
private pendingParentWakeTimers: Map<string, ReturnType<typeof setTimeout>> = new Map()
|
||||
private observedOutputSessions: Set<string> = new Set()
|
||||
private observedIncompleteTodosBySession: Map<string, boolean> = new Map()
|
||||
private rootDescendantCounts: Map<string, number>
|
||||
@@ -1410,6 +1423,12 @@ The fallback retry session is now created and can be inspected directly.
|
||||
|
||||
if (event.type === "session.idle") {
|
||||
if (!props || typeof props !== "object") return
|
||||
const sessionID = resolveSessionEventID(props)
|
||||
if (sessionID) {
|
||||
void this.enqueueNotificationForParent(sessionID, () => this.flushPendingParentWake(sessionID)).catch((error) => {
|
||||
log("[background-agent] Failed to flush pending parent wake:", { sessionID, error })
|
||||
})
|
||||
}
|
||||
handleSessionIdleBackgroundEvent({
|
||||
properties: props as Record<string, unknown>,
|
||||
findBySession: (id) => {
|
||||
@@ -2216,30 +2235,41 @@ The task was re-queued on a fallback model after a retryable failure.
|
||||
...(variant !== undefined ? { variant } : {}),
|
||||
...(resolvedTools ? { tools: resolvedTools } : {}),
|
||||
}
|
||||
try {
|
||||
await promptAsyncInDirectory(this.client, {
|
||||
path: { id: task.parentSessionId },
|
||||
body: {
|
||||
noReply: !shouldReply,
|
||||
...parentPromptContext,
|
||||
parts: [createInternalAgentTextPart(notification)],
|
||||
},
|
||||
}, this.directory)
|
||||
log("[background-agent] Sent notification to parent session:", {
|
||||
const shouldDeferReply = shouldReply && await this.isSessionActive(task.parentSessionId)
|
||||
|
||||
if (shouldDeferReply) {
|
||||
this.queuePendingParentWake(task.parentSessionId, notification, parentPromptContext)
|
||||
log("[background-agent] Deferred notification until parent session is idle:", {
|
||||
taskId: task.id,
|
||||
allComplete,
|
||||
isTaskFailure,
|
||||
noReply: !shouldReply,
|
||||
})
|
||||
} catch (error) {
|
||||
if (isAbortedSessionError(error)) {
|
||||
log("[background-agent] Parent session aborted while sending notification; continuing cleanup:", {
|
||||
} else {
|
||||
try {
|
||||
await promptAsyncInDirectory(this.client, {
|
||||
path: { id: task.parentSessionId },
|
||||
body: {
|
||||
noReply: !shouldReply,
|
||||
...parentPromptContext,
|
||||
parts: [createInternalAgentTextPart(notification)],
|
||||
},
|
||||
}, this.directory)
|
||||
log("[background-agent] Sent notification to parent session:", {
|
||||
taskId: task.id,
|
||||
parentSessionID: task.parentSessionId,
|
||||
allComplete,
|
||||
isTaskFailure,
|
||||
noReply: !shouldReply,
|
||||
})
|
||||
this.queuePendingNotification(task.parentSessionId, notification)
|
||||
} else {
|
||||
log("[background-agent] Failed to send notification:", error)
|
||||
} catch (error) {
|
||||
if (isAbortedSessionError(error)) {
|
||||
log("[background-agent] Parent session aborted while sending notification; continuing cleanup:", {
|
||||
taskId: task.id,
|
||||
parentSessionID: task.parentSessionId,
|
||||
})
|
||||
this.queuePendingNotification(task.parentSessionId, notification)
|
||||
} else {
|
||||
log("[background-agent] Failed to send notification:", error)
|
||||
}
|
||||
}
|
||||
}
|
||||
} else {
|
||||
@@ -2254,6 +2284,93 @@ The task was re-queued on a fallback model after a retryable failure.
|
||||
}
|
||||
}
|
||||
|
||||
private async isSessionActive(sessionID: string): Promise<boolean> {
|
||||
return isOpenCodeSessionActive(this.client, sessionID)
|
||||
}
|
||||
|
||||
private queuePendingParentWake(
|
||||
sessionID: string,
|
||||
notification: string,
|
||||
promptContext: ParentWakePromptContext,
|
||||
): void {
|
||||
const pendingWake = this.pendingParentWakes.get(sessionID)
|
||||
if (pendingWake) {
|
||||
pendingWake.notifications.push(notification)
|
||||
pendingWake.promptContext = promptContext
|
||||
} else {
|
||||
this.pendingParentWakes.set(sessionID, {
|
||||
promptContext,
|
||||
notifications: [notification],
|
||||
})
|
||||
}
|
||||
this.schedulePendingParentWakeFlush(sessionID)
|
||||
}
|
||||
|
||||
private async flushPendingParentWake(sessionID: string): Promise<void> {
|
||||
const pendingWake = this.pendingParentWakes.get(sessionID)
|
||||
if (!pendingWake) {
|
||||
this.clearPendingParentWakeTimer(sessionID)
|
||||
return
|
||||
}
|
||||
|
||||
if (await this.isSessionActive(sessionID)) {
|
||||
this.schedulePendingParentWakeFlush(sessionID)
|
||||
return
|
||||
}
|
||||
|
||||
this.pendingParentWakes.delete(sessionID)
|
||||
this.clearPendingParentWakeTimer(sessionID)
|
||||
await settleAfterSessionIdle()
|
||||
|
||||
if (await this.isSessionActive(sessionID)) {
|
||||
this.pendingParentWakes.set(sessionID, pendingWake)
|
||||
this.schedulePendingParentWakeFlush(sessionID)
|
||||
return
|
||||
}
|
||||
|
||||
const notificationContent = pendingWake.notifications.join("\n\n")
|
||||
|
||||
try {
|
||||
await promptAsyncInDirectory(this.client, {
|
||||
path: { id: sessionID },
|
||||
body: {
|
||||
noReply: false,
|
||||
...pendingWake.promptContext,
|
||||
parts: [createInternalAgentTextPart(notificationContent)],
|
||||
},
|
||||
}, this.directory)
|
||||
log("[background-agent] Sent deferred parent wake:", { sessionID })
|
||||
} catch (error) {
|
||||
this.queuePendingNotification(sessionID, notificationContent)
|
||||
log("[background-agent] Failed to send deferred parent wake:", { sessionID, error })
|
||||
}
|
||||
}
|
||||
|
||||
private schedulePendingParentWakeFlush(sessionID: string): void {
|
||||
if (this.pendingParentWakeTimers.has(sessionID)) {
|
||||
return
|
||||
}
|
||||
|
||||
const timer = setTimeout(() => {
|
||||
this.pendingParentWakeTimers.delete(sessionID)
|
||||
void this.enqueueNotificationForParent(sessionID, () => this.flushPendingParentWake(sessionID)).catch((error) => {
|
||||
log("[background-agent] Failed to retry pending parent wake:", { sessionID, error })
|
||||
})
|
||||
}, PENDING_PARENT_WAKE_RETRY_MS)
|
||||
|
||||
this.pendingParentWakeTimers.set(sessionID, timer)
|
||||
}
|
||||
|
||||
private clearPendingParentWakeTimer(sessionID: string): void {
|
||||
const timer = this.pendingParentWakeTimers.get(sessionID)
|
||||
if (!timer) {
|
||||
return
|
||||
}
|
||||
|
||||
clearTimeout(timer)
|
||||
this.pendingParentWakeTimers.delete(sessionID)
|
||||
}
|
||||
|
||||
private hasRunningTasks(): boolean {
|
||||
for (const task of this.tasks.values()) {
|
||||
if (task.status === "running") return true
|
||||
@@ -2574,6 +2691,11 @@ The task was re-queued on a fallback model after a retryable failure.
|
||||
}
|
||||
this.idleDeferralTimers.clear()
|
||||
|
||||
for (const timer of this.pendingParentWakeTimers.values()) {
|
||||
clearTimeout(timer)
|
||||
}
|
||||
this.pendingParentWakeTimers.clear()
|
||||
|
||||
for (const sessionID of trackedSessionIDs) {
|
||||
subagentSessions.delete(sessionID)
|
||||
SessionCategoryRegistry.remove(sessionID)
|
||||
@@ -2585,6 +2707,7 @@ The task was re-queued on a fallback model after a retryable failure.
|
||||
this.notifications.clear()
|
||||
this.pendingNotifications.clear()
|
||||
this.pendingByParent.clear()
|
||||
this.pendingParentWakes.clear()
|
||||
this.notificationQueueByParent.clear()
|
||||
this.rootDescendantCounts.clear()
|
||||
this.queuesByKey.clear()
|
||||
|
||||
@@ -163,6 +163,14 @@ async function notifyParentSessionForTest(manager: BackgroundManager, task: Back
|
||||
return notifyParentSession.call(manager, task)
|
||||
}
|
||||
|
||||
function waitForDeferredWake(): Promise<void> {
|
||||
return new Promise((resolve) => setTimeout(resolve, 180))
|
||||
}
|
||||
|
||||
function waitForDeferredWakeRetry(): Promise<void> {
|
||||
return new Promise((resolve) => setTimeout(resolve, 1_180))
|
||||
}
|
||||
|
||||
function getRequiredTimer(manager: BackgroundManager, taskID: string): ReturnType<typeof setTimeout> {
|
||||
const timer = getCompletionTimers(manager).get(taskID)
|
||||
expect(timer).toBeDefined()
|
||||
@@ -255,7 +263,7 @@ describe("BackgroundManager.notifyParentSession cleanup scheduling", () => {
|
||||
expect(allCompletePayload).toContain(taskB.description)
|
||||
})
|
||||
|
||||
test("#when parent session is busy #then all-complete notification keeps the direct 4.0.0 parent prompt behavior", async () => {
|
||||
test("#when parent session is busy #then all-complete notification does not start an overlapping parent reply", async () => {
|
||||
// given
|
||||
const sessionStatuses: Record<string, { type: string }> = {
|
||||
"parent-1": { type: "busy" },
|
||||
@@ -270,11 +278,7 @@ describe("BackgroundManager.notifyParentSession cleanup scheduling", () => {
|
||||
await notifyParentSessionForTest(manager, task)
|
||||
|
||||
// then
|
||||
expect(promptAsyncCalls).toHaveLength(1)
|
||||
expect(promptAsyncCalls[0]?.body.noReply).toBe(false)
|
||||
const notificationPayload = JSON.stringify(promptAsyncCalls[0]?.body.parts)
|
||||
expect(notificationPayload).toContain("ALL BACKGROUND TASKS COMPLETE")
|
||||
expect(notificationPayload).toContain(OMO_INTERNAL_INITIATOR_MARKER)
|
||||
expect(promptAsyncCalls).toHaveLength(0)
|
||||
})
|
||||
|
||||
test("#when all-complete notification wakes parent #then prompt stays in the same OpenCode directory instance", async () => {
|
||||
@@ -295,7 +299,7 @@ describe("BackgroundManager.notifyParentSession cleanup scheduling", () => {
|
||||
expect(promptAsyncCalls[0]?.query).toEqual({ directory })
|
||||
})
|
||||
|
||||
test("#when busy parent later becomes idle #then completion notification is not replayed as a second parent prompt", async () => {
|
||||
test("#when busy parent later becomes idle #then completion notification wakes the parent once", async () => {
|
||||
// given
|
||||
const sessionStatuses: Record<string, { type: string }> = {
|
||||
"parent-1": { type: "busy" },
|
||||
@@ -306,12 +310,12 @@ describe("BackgroundManager.notifyParentSession cleanup scheduling", () => {
|
||||
getTasks(manager).set(task.id, task)
|
||||
getPendingByParent(manager).set(task.parentSessionId, new Set([task.id]))
|
||||
await notifyParentSessionForTest(manager, task)
|
||||
expect(promptAsyncCalls).toHaveLength(1)
|
||||
expect(promptAsyncCalls).toHaveLength(0)
|
||||
|
||||
// when
|
||||
sessionStatuses["parent-1"] = { type: "idle" }
|
||||
manager.handleEvent({ type: "session.idle", properties: { sessionID: "parent-1" } })
|
||||
await Promise.resolve()
|
||||
await waitForDeferredWake()
|
||||
|
||||
// then
|
||||
expect(promptAsyncCalls).toHaveLength(1)
|
||||
@@ -321,7 +325,7 @@ describe("BackgroundManager.notifyParentSession cleanup scheduling", () => {
|
||||
expect(notificationPayload).not.toContain("BACKGROUND TASK NOTIFICATION READY")
|
||||
})
|
||||
|
||||
test("#when a single background task finishes during a stale busy parent status #then no deferred wake is scheduled", async () => {
|
||||
test("#when a single background task finishes during a stale busy parent status #then completion notification is retried after the parent becomes idle", async () => {
|
||||
// given
|
||||
const sessionStatuses: Record<string, { type: string }> = {
|
||||
"parent-1": { type: "busy" },
|
||||
@@ -335,7 +339,7 @@ describe("BackgroundManager.notifyParentSession cleanup scheduling", () => {
|
||||
// when
|
||||
await notifyParentSessionForTest(manager, task)
|
||||
sessionStatuses["parent-1"] = { type: "idle" }
|
||||
await new Promise((resolve) => setTimeout(resolve, 1_180))
|
||||
await waitForDeferredWakeRetry()
|
||||
|
||||
// then
|
||||
expect(promptAsyncCalls).toHaveLength(1)
|
||||
@@ -362,6 +366,9 @@ describe("BackgroundManager.notifyParentSession cleanup scheduling", () => {
|
||||
|
||||
// when
|
||||
await notifyParentSessionForTest(manager, task)
|
||||
sessionStatuses["parent-1"] = { type: "idle" }
|
||||
manager.handleEvent({ type: "session.idle", properties: { sessionID: "parent-1" } })
|
||||
await waitForDeferredWake()
|
||||
|
||||
// then
|
||||
expect(promptAsyncCalls).toHaveLength(1)
|
||||
|
||||
@@ -0,0 +1,55 @@
|
||||
import { describe, expect, test } from "bun:test"
|
||||
|
||||
import {
|
||||
isSessionActive,
|
||||
shouldPromptAfterSessionIdle,
|
||||
} from "./session-idle-settle"
|
||||
|
||||
describe("session idle prompt guard", () => {
|
||||
test("#given session.status reports busy #when checking active session #then it returns true", async () => {
|
||||
// given
|
||||
const client = {
|
||||
session: {
|
||||
status: async () => ({
|
||||
data: {
|
||||
"ses-active": { type: "busy" },
|
||||
},
|
||||
}),
|
||||
},
|
||||
}
|
||||
|
||||
// when
|
||||
const active = await isSessionActive(client, "ses-active")
|
||||
|
||||
// then
|
||||
expect(active).toBe(true)
|
||||
})
|
||||
|
||||
test("#given a stale idle event but session became busy #when settling before prompt #then it blocks the wake", async () => {
|
||||
// given
|
||||
const client = {
|
||||
session: {
|
||||
status: async () => ({
|
||||
"ses-active": { type: "busy" },
|
||||
}),
|
||||
},
|
||||
}
|
||||
|
||||
// when
|
||||
const shouldPrompt = await shouldPromptAfterSessionIdle(client, "ses-active", 0)
|
||||
|
||||
// then
|
||||
expect(shouldPrompt).toBe(false)
|
||||
})
|
||||
|
||||
test("#given session.status is unavailable #when settling before prompt #then it preserves legacy prompt behavior", async () => {
|
||||
// given
|
||||
const client = { session: {} }
|
||||
|
||||
// when
|
||||
const shouldPrompt = await shouldPromptAfterSessionIdle(client, "ses-legacy", 0)
|
||||
|
||||
// then
|
||||
expect(shouldPrompt).toBe(true)
|
||||
})
|
||||
})
|
||||
@@ -3,3 +3,59 @@ export const DEFAULT_SESSION_IDLE_SETTLE_MS = 150
|
||||
export function settleAfterSessionIdle(ms = DEFAULT_SESSION_IDLE_SETTLE_MS): Promise<void> {
|
||||
return ms > 0 ? new Promise((resolve) => setTimeout(resolve, ms)) : Promise.resolve()
|
||||
}
|
||||
|
||||
type SessionStatusClient = {
|
||||
session?: {
|
||||
status?: () => Promise<unknown>
|
||||
}
|
||||
}
|
||||
|
||||
const ACTIVE_SESSION_STATUSES = new Set(["busy", "retry", "running"])
|
||||
|
||||
function isRecord(value: unknown): value is Record<string, unknown> {
|
||||
return typeof value === "object" && value !== null
|
||||
}
|
||||
|
||||
function getSessionStatusPayload(response: unknown): Record<string, unknown> {
|
||||
if (isRecord(response) && isRecord(response.data)) {
|
||||
return response.data
|
||||
}
|
||||
|
||||
if (isRecord(response)) {
|
||||
return response
|
||||
}
|
||||
|
||||
return {}
|
||||
}
|
||||
|
||||
export function isActiveSessionStatusType(statusType: string): boolean {
|
||||
return ACTIVE_SESSION_STATUSES.has(statusType)
|
||||
}
|
||||
|
||||
export async function isSessionActive(client: SessionStatusClient, sessionID: string): Promise<boolean> {
|
||||
if (typeof client.session?.status !== "function") {
|
||||
return false
|
||||
}
|
||||
|
||||
try {
|
||||
const statusResult = await client.session.status()
|
||||
const status = getSessionStatusPayload(statusResult)[sessionID]
|
||||
if (!isRecord(status)) {
|
||||
return false
|
||||
}
|
||||
|
||||
const statusType = status.type
|
||||
return typeof statusType === "string" && isActiveSessionStatusType(statusType)
|
||||
} catch {
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
export async function shouldPromptAfterSessionIdle(
|
||||
client: SessionStatusClient,
|
||||
sessionID: string,
|
||||
settleMs = DEFAULT_SESSION_IDLE_SETTLE_MS,
|
||||
): Promise<boolean> {
|
||||
await settleAfterSessionIdle(settleMs)
|
||||
return !(await isSessionActive(client, sessionID))
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user