Merge PR #3907: fix(delegate-task): pause timeouts for active sessions
fix(delegate-task): pause timeouts for active sessions
This commit is contained in:
@@ -364,6 +364,29 @@ describe("BackgroundManager pollRunningTasks", () => {
|
|||||||
//#then
|
//#then
|
||||||
expect(task.status).toBe("running")
|
expect(task.status).toBe("running")
|
||||||
})
|
})
|
||||||
|
|
||||||
|
test("#when progress is older than prune TTL #then active status still keeps the task running", async () => {
|
||||||
|
//#given
|
||||||
|
const manager = createManagerWithClient({
|
||||||
|
status: async () => ({ data: { "ses-busy-stale": { type: "busy" } } }),
|
||||||
|
})
|
||||||
|
const task = createRunningTask("ses-busy-stale")
|
||||||
|
task.startedAt = new Date(Date.now() - 60 * 60 * 1000)
|
||||||
|
task.progress = {
|
||||||
|
toolCalls: 4,
|
||||||
|
lastUpdate: new Date(Date.now() - 35 * 60 * 1000),
|
||||||
|
}
|
||||||
|
injectTask(manager, task)
|
||||||
|
|
||||||
|
//#when
|
||||||
|
const poll = manager["pollRunningTasks"]
|
||||||
|
await poll.call(manager)
|
||||||
|
manager.shutdown()
|
||||||
|
|
||||||
|
//#then
|
||||||
|
expect(task.status).toBe("running")
|
||||||
|
expect(task.error).toBeUndefined()
|
||||||
|
})
|
||||||
})
|
})
|
||||||
|
|
||||||
describe("#given a running task whose session has terminal non-idle status", () => {
|
describe("#given a running task whose session has terminal non-idle status", () => {
|
||||||
|
|||||||
@@ -2285,11 +2285,12 @@ The task was re-queued on a fallback model after a retryable failure.
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
private pruneStaleTasksAndNotifications(): void {
|
private pruneStaleTasksAndNotifications(allStatuses?: SessionStatusMap): void {
|
||||||
pruneStaleTasksAndNotifications({
|
pruneStaleTasksAndNotifications({
|
||||||
tasks: this.tasks,
|
tasks: this.tasks,
|
||||||
notifications: this.notifications,
|
notifications: this.notifications,
|
||||||
taskTtlMs: this.config?.taskTtlMs,
|
taskTtlMs: this.config?.taskTtlMs,
|
||||||
|
sessionStatuses: allStatuses,
|
||||||
onTaskPruned: (taskId, task, errorMessage) => {
|
onTaskPruned: (taskId, task, errorMessage) => {
|
||||||
const wasPending = task.status === "pending"
|
const wasPending = task.status === "pending"
|
||||||
log("[background-agent] Pruning stale task:", { taskId, status: task.status, age: Math.round(((wasPending ? task.queuedAt?.getTime() : task.startedAt?.getTime()) ? (Date.now() - (wasPending ? task.queuedAt!.getTime() : task.startedAt!.getTime())) : 0) / 1000) + "s" })
|
log("[background-agent] Pruning stale task:", { taskId, status: task.status, age: Math.round(((wasPending ? task.queuedAt?.getTime() : task.startedAt?.getTime()) ? (Date.now() - (wasPending ? task.queuedAt!.getTime() : task.startedAt!.getTime())) : 0) / 1000) + "s" })
|
||||||
@@ -2412,8 +2413,6 @@ The task was re-queued on a fallback model after a retryable failure.
|
|||||||
if (this.pollingInFlight) return
|
if (this.pollingInFlight) return
|
||||||
this.pollingInFlight = true
|
this.pollingInFlight = true
|
||||||
try {
|
try {
|
||||||
this.pruneStaleTasksAndNotifications()
|
|
||||||
|
|
||||||
let allStatuses: SessionStatusMap | undefined
|
let allStatuses: SessionStatusMap | undefined
|
||||||
const sessionStatusMethod = this.client?.session?.status
|
const sessionStatusMethod = this.client?.session?.status
|
||||||
if (typeof sessionStatusMethod !== "function") {
|
if (typeof sessionStatusMethod !== "function") {
|
||||||
@@ -2435,6 +2434,8 @@ The task was re-queued on a fallback model after a retryable failure.
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
this.pruneStaleTasksAndNotifications(allStatuses)
|
||||||
|
|
||||||
await this.checkAndInterruptStaleTasks(allStatuses)
|
await this.checkAndInterruptStaleTasks(allStatuses)
|
||||||
|
|
||||||
for (const task of this.tasks.values()) {
|
for (const task of this.tasks.values()) {
|
||||||
|
|||||||
@@ -903,6 +903,42 @@ describe("pruneStaleTasksAndNotifications", () => {
|
|||||||
expect(pruned).toContain("stale-task")
|
expect(pruned).toContain("stale-task")
|
||||||
})
|
})
|
||||||
|
|
||||||
|
it("#given running task with stale progress and active session #when lastUpdate exceeds TTL #then should NOT prune", () => {
|
||||||
|
//#given
|
||||||
|
const tasks = new Map<string, BackgroundTask>()
|
||||||
|
const activeTask: BackgroundTask = {
|
||||||
|
id: "active-status-task",
|
||||||
|
sessionId: "ses-active-status",
|
||||||
|
parentSessionId: "parent",
|
||||||
|
parentMessageId: "msg",
|
||||||
|
description: "active status",
|
||||||
|
prompt: "active status",
|
||||||
|
agent: "oracle",
|
||||||
|
status: "running",
|
||||||
|
startedAt: new Date(Date.now() - 60 * 60 * 1000),
|
||||||
|
progress: {
|
||||||
|
toolCalls: 10,
|
||||||
|
lastUpdate: new Date(Date.now() - 35 * 60 * 1000),
|
||||||
|
},
|
||||||
|
}
|
||||||
|
tasks.set("active-status-task", activeTask)
|
||||||
|
|
||||||
|
const pruned: string[] = []
|
||||||
|
const notifications = new Map<string, BackgroundTask[]>()
|
||||||
|
|
||||||
|
//#when
|
||||||
|
pruneStaleTasksAndNotifications({
|
||||||
|
tasks,
|
||||||
|
notifications,
|
||||||
|
sessionStatuses: { "ses-active-status": { type: "busy" } },
|
||||||
|
onTaskPruned: (taskId) => pruned.push(taskId),
|
||||||
|
})
|
||||||
|
|
||||||
|
//#then
|
||||||
|
expect(pruned).toEqual([])
|
||||||
|
expect(tasks.has("active-status-task")).toBe(true)
|
||||||
|
})
|
||||||
|
|
||||||
it("#given custom taskTtlMs #when task exceeds custom TTL #then should prune", () => {
|
it("#given custom taskTtlMs #when task exceeds custom TTL #then should prune", () => {
|
||||||
//#given
|
//#given
|
||||||
const tasks = new Map<string, BackgroundTask>()
|
const tasks = new Map<string, BackgroundTask>()
|
||||||
|
|||||||
@@ -31,6 +31,7 @@ export function pruneStaleTasksAndNotifications(args: {
|
|||||||
notifications: Map<string, BackgroundTask[]>
|
notifications: Map<string, BackgroundTask[]>
|
||||||
onTaskPruned: (taskId: string, task: BackgroundTask, errorMessage: string) => void
|
onTaskPruned: (taskId: string, task: BackgroundTask, errorMessage: string) => void
|
||||||
taskTtlMs?: number
|
taskTtlMs?: number
|
||||||
|
sessionStatuses?: SessionStatusMap
|
||||||
}): void {
|
}): void {
|
||||||
const { tasks, notifications, onTaskPruned } = args
|
const { tasks, notifications, onTaskPruned } = args
|
||||||
const effectiveTtl = args.taskTtlMs ?? TASK_TTL_MS
|
const effectiveTtl = args.taskTtlMs ?? TASK_TTL_MS
|
||||||
@@ -62,6 +63,11 @@ export function pruneStaleTasksAndNotifications(args: {
|
|||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
|
||||||
|
const sessionStatus = task.sessionId ? args.sessionStatuses?.[task.sessionId]?.type : undefined
|
||||||
|
if (task.status === "running" && sessionStatus !== undefined && isActiveSessionStatus(sessionStatus)) {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
|
||||||
const lastActivity = task.status === "running" && task.progress?.lastUpdate
|
const lastActivity = task.status === "running" && task.progress?.lastUpdate
|
||||||
? task.progress.lastUpdate.getTime()
|
? task.progress.lastUpdate.getTime()
|
||||||
: undefined
|
: undefined
|
||||||
|
|||||||
@@ -79,6 +79,52 @@ describe("syncPollTimeoutMs threading", () => {
|
|||||||
expect(abortCount).toBe(1)
|
expect(abortCount).toBe(1)
|
||||||
})
|
})
|
||||||
})
|
})
|
||||||
|
|
||||||
|
test("#then active OpenCode statuses do not consume the inactivity timeout", async () => {
|
||||||
|
const { pollSyncSession } = require("./sync-session-poller")
|
||||||
|
let abortCount = 0
|
||||||
|
let statusCallCount = 0
|
||||||
|
let messageCallCount = 0
|
||||||
|
const mockClient = {
|
||||||
|
session: {
|
||||||
|
abort: async () => {
|
||||||
|
abortCount++
|
||||||
|
},
|
||||||
|
messages: async () => {
|
||||||
|
messageCallCount++
|
||||||
|
return {
|
||||||
|
data: [
|
||||||
|
{ info: { id: "msg_001", role: "user", time: { created: 1000 } } },
|
||||||
|
{
|
||||||
|
info: { id: "msg_002", role: "assistant", time: { created: 2000 }, finish: "stop" },
|
||||||
|
parts: [{ type: "text", text: "done" }],
|
||||||
|
},
|
||||||
|
],
|
||||||
|
}
|
||||||
|
},
|
||||||
|
status: async () => {
|
||||||
|
statusCallCount++
|
||||||
|
if (statusCallCount === 1) return { data: { ses_active: { type: "busy" } } }
|
||||||
|
if (statusCallCount === 2) return { data: { ses_active: { type: "retry" } } }
|
||||||
|
return { data: { ses_active: { type: "idle" } } }
|
||||||
|
},
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
|
await withMockedDateNow(60_000, async () => {
|
||||||
|
const result = await pollSyncSession(createMockCtx(), mockClient, {
|
||||||
|
sessionID: "ses_active",
|
||||||
|
agentToUse: "oracle",
|
||||||
|
toastManager: null,
|
||||||
|
taskId: undefined,
|
||||||
|
}, 120_000)
|
||||||
|
|
||||||
|
expect(result).toBeNull()
|
||||||
|
expect(abortCount).toBe(0)
|
||||||
|
expect(statusCallCount).toBe(3)
|
||||||
|
expect(messageCallCount).toBe(1)
|
||||||
|
})
|
||||||
|
})
|
||||||
})
|
})
|
||||||
|
|
||||||
describe("#when timeoutMs is omitted", () => {
|
describe("#when timeoutMs is omitted", () => {
|
||||||
|
|||||||
@@ -7,6 +7,7 @@ import { extractErrorMessage } from "../../features/background-agent/error-class
|
|||||||
|
|
||||||
const NON_TERMINAL_FINISH_REASONS = new Set(["tool-calls", "unknown"])
|
const NON_TERMINAL_FINISH_REASONS = new Set(["tool-calls", "unknown"])
|
||||||
const PENDING_TOOL_PART_TYPES = new Set(["tool", "tool_use", "tool-call"])
|
const PENDING_TOOL_PART_TYPES = new Set(["tool", "tool_use", "tool-call"])
|
||||||
|
const ACTIVE_SESSION_STATUSES = new Set(["busy", "retry", "running"])
|
||||||
|
|
||||||
function wait(milliseconds: number): Promise<void> {
|
function wait(milliseconds: number): Promise<void> {
|
||||||
const sharedBuffer = new SharedArrayBuffer(Int32Array.BYTES_PER_ELEMENT)
|
const sharedBuffer = new SharedArrayBuffer(Int32Array.BYTES_PER_ELEMENT)
|
||||||
@@ -24,6 +25,10 @@ function abortSyncSession(client: OpencodeClient, sessionID: string, reason: str
|
|||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
function isActiveSessionStatus(status: { type: string } | undefined): boolean {
|
||||||
|
return status !== undefined && ACTIVE_SESSION_STATUSES.has(status.type)
|
||||||
|
}
|
||||||
|
|
||||||
async function fetchSessionMessages(
|
async function fetchSessionMessages(
|
||||||
client: OpencodeClient,
|
client: OpencodeClient,
|
||||||
sessionID: string
|
sessionID: string
|
||||||
@@ -84,6 +89,7 @@ export async function pollSyncSession(
|
|||||||
const maxPollTimeMs = Math.max(timeoutMs ?? getDefaultSyncPollTimeoutMs(), 50)
|
const maxPollTimeMs = Math.max(timeoutMs ?? getDefaultSyncPollTimeoutMs(), 50)
|
||||||
const maxTurns = input.maxAssistantTurns ?? DEFAULT_MAX_ASSISTANT_TURNS
|
const maxTurns = input.maxAssistantTurns ?? DEFAULT_MAX_ASSISTANT_TURNS
|
||||||
const pollStart = Date.now()
|
const pollStart = Date.now()
|
||||||
|
let inactiveStart = pollStart
|
||||||
let pollCount = 0
|
let pollCount = 0
|
||||||
let timedOut = false
|
let timedOut = false
|
||||||
let assistantTurnCount = 0
|
let assistantTurnCount = 0
|
||||||
@@ -91,7 +97,13 @@ export async function pollSyncSession(
|
|||||||
|
|
||||||
log("[task] Starting poll loop", { sessionID: input.sessionID, agentToUse: input.agentToUse, maxTurns })
|
log("[task] Starting poll loop", { sessionID: input.sessionID, agentToUse: input.agentToUse, maxTurns })
|
||||||
|
|
||||||
while (Date.now() - pollStart < maxPollTimeMs) {
|
while (true) {
|
||||||
|
const inactiveElapsedMs = Date.now() - inactiveStart
|
||||||
|
if (inactiveElapsedMs >= maxPollTimeMs) {
|
||||||
|
timedOut = true
|
||||||
|
break
|
||||||
|
}
|
||||||
|
|
||||||
if (ctx.abort?.aborted) {
|
if (ctx.abort?.aborted) {
|
||||||
try {
|
try {
|
||||||
const messages = await fetchSessionMessages(client, input.sessionID)
|
const messages = await fetchSessionMessages(client, input.sessionID)
|
||||||
@@ -132,11 +144,13 @@ export async function pollSyncSession(
|
|||||||
sessionID: input.sessionID,
|
sessionID: input.sessionID,
|
||||||
pollCount,
|
pollCount,
|
||||||
elapsed: Math.floor((Date.now() - pollStart) / 1000) + "s",
|
elapsed: Math.floor((Date.now() - pollStart) / 1000) + "s",
|
||||||
|
inactiveElapsed: Math.floor(inactiveElapsedMs / 1000) + "s",
|
||||||
sessionStatus: sessionStatus?.type ?? "not_in_status",
|
sessionStatus: sessionStatus?.type ?? "not_in_status",
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
if (sessionStatus && sessionStatus.type !== "idle") {
|
if (isActiveSessionStatus(sessionStatus)) {
|
||||||
|
inactiveStart = Date.now()
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -199,8 +213,7 @@ export async function pollSyncSession(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
if (Date.now() - pollStart >= maxPollTimeMs) {
|
if (timedOut) {
|
||||||
timedOut = true
|
|
||||||
log("[task] Poll timeout reached", { sessionID: input.sessionID, pollCount })
|
log("[task] Poll timeout reached", { sessionID: input.sessionID, pollCount })
|
||||||
abortSyncSession(client, input.sessionID, "poll_timeout")
|
abortSyncSession(client, input.sessionID, "poll_timeout")
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user