feat(background-agent): adapt task poller for team-mode work distribution
This commit is contained in:
@@ -107,6 +107,57 @@ describe("checkAndInterruptStaleTasks", () => {
|
|||||||
expect(task.status).toBe("running")
|
expect(task.status).toBe("running")
|
||||||
})
|
})
|
||||||
|
|
||||||
|
it("should NOT interrupt idle team-member tasks just because lastUpdate is old", async () => {
|
||||||
|
//#given
|
||||||
|
const task = createRunningTask({
|
||||||
|
teamRunId: "team-run-1",
|
||||||
|
progress: {
|
||||||
|
toolCalls: 1,
|
||||||
|
lastUpdate: new Date(Date.now() - 200_000),
|
||||||
|
},
|
||||||
|
})
|
||||||
|
|
||||||
|
//#when
|
||||||
|
await checkAndInterruptStaleTasks({
|
||||||
|
tasks: [task],
|
||||||
|
client: mockClient as never,
|
||||||
|
config: { staleTimeoutMs: 180_000 },
|
||||||
|
concurrencyManager: mockConcurrencyManager as never,
|
||||||
|
notifyParentSession: mockNotify,
|
||||||
|
sessionStatuses: { "ses-1": { type: "idle" } },
|
||||||
|
})
|
||||||
|
|
||||||
|
//#then
|
||||||
|
expect(task.status).toBe("running")
|
||||||
|
})
|
||||||
|
|
||||||
|
it("should still interrupt team-member tasks when the session is gone", async () => {
|
||||||
|
//#given
|
||||||
|
const task = createRunningTask({
|
||||||
|
teamRunId: "team-run-1",
|
||||||
|
progress: {
|
||||||
|
toolCalls: 1,
|
||||||
|
lastUpdate: new Date(Date.now() - 200_000),
|
||||||
|
},
|
||||||
|
consecutiveMissedPolls: 2,
|
||||||
|
})
|
||||||
|
mockClient.session.get.mockRejectedValueOnce(new Error("missing"))
|
||||||
|
|
||||||
|
//#when
|
||||||
|
await checkAndInterruptStaleTasks({
|
||||||
|
tasks: [task],
|
||||||
|
client: mockClient as never,
|
||||||
|
config: { staleTimeoutMs: 180_000, sessionGoneTimeoutMs: 180_000 },
|
||||||
|
concurrencyManager: mockConcurrencyManager as never,
|
||||||
|
notifyParentSession: mockNotify,
|
||||||
|
sessionStatuses: {},
|
||||||
|
})
|
||||||
|
|
||||||
|
//#then
|
||||||
|
expect(task.status).toBe("cancelled")
|
||||||
|
expect(task.error).toContain("session gone from status registry")
|
||||||
|
})
|
||||||
|
|
||||||
it("should interrupt tasks with NO progress.lastUpdate that exceeded messageStalenessTimeoutMs since startedAt", async () => {
|
it("should interrupt tasks with NO progress.lastUpdate that exceeded messageStalenessTimeoutMs since startedAt", async () => {
|
||||||
//#given - task started 15 minutes ago, never received any progress update
|
//#given - task started 15 minutes ago, never received any progress update
|
||||||
const task = createRunningTask({
|
const task = createRunningTask({
|
||||||
@@ -912,6 +963,41 @@ describe("pruneStaleTasksAndNotifications", () => {
|
|||||||
expect(pruned).toEqual([])
|
expect(pruned).toEqual([])
|
||||||
})
|
})
|
||||||
|
|
||||||
|
it("#given active team-member task with stale progress #when prune runs #then should NOT prune", () => {
|
||||||
|
//#given
|
||||||
|
const tasks = new Map<string, BackgroundTask>()
|
||||||
|
const task: BackgroundTask = {
|
||||||
|
id: "team-task",
|
||||||
|
sessionID: "ses-team-1",
|
||||||
|
parentSessionID: "parent",
|
||||||
|
parentMessageID: "msg",
|
||||||
|
teamRunId: "team-run-1",
|
||||||
|
description: "team member",
|
||||||
|
prompt: "team member",
|
||||||
|
agent: "sisyphus-junior",
|
||||||
|
status: "running",
|
||||||
|
startedAt: new Date(Date.now() - 60 * 60 * 1000),
|
||||||
|
progress: {
|
||||||
|
toolCalls: 1,
|
||||||
|
lastUpdate: new Date(Date.now() - 35 * 60 * 1000),
|
||||||
|
},
|
||||||
|
}
|
||||||
|
tasks.set(task.id, task)
|
||||||
|
|
||||||
|
const pruned: string[] = []
|
||||||
|
|
||||||
|
//#when
|
||||||
|
pruneStaleTasksAndNotifications({
|
||||||
|
tasks,
|
||||||
|
notifications: new Map<string, BackgroundTask[]>(),
|
||||||
|
onTaskPruned: (taskId) => pruned.push(taskId),
|
||||||
|
})
|
||||||
|
|
||||||
|
//#then
|
||||||
|
expect(pruned).toEqual([])
|
||||||
|
expect(tasks.has(task.id)).toBe(true)
|
||||||
|
})
|
||||||
|
|
||||||
it("should prune terminal tasks when completion time exceeds terminal TTL", () => {
|
it("should prune terminal tasks when completion time exceeds terminal TTL", () => {
|
||||||
//#given
|
//#given
|
||||||
const tasks = new Map<string, BackgroundTask>()
|
const tasks = new Map<string, BackgroundTask>()
|
||||||
|
|||||||
@@ -58,6 +58,10 @@ export function pruneStaleTasksAndNotifications(args: {
|
|||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if (task.teamRunId) {
|
||||||
|
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
|
||||||
@@ -146,8 +150,10 @@ export async function checkAndInterruptStaleTasks(args: {
|
|||||||
}
|
}
|
||||||
|
|
||||||
const sessionGone = sessionMissing && (task.consecutiveMissedPolls ?? 0) >= MIN_SESSION_GONE_POLLS
|
const sessionGone = sessionMissing && (task.consecutiveMissedPolls ?? 0) >= MIN_SESSION_GONE_POLLS
|
||||||
|
const shouldSkipInactivityTimeout = task.teamRunId !== undefined && !sessionGone
|
||||||
|
|
||||||
if (!task.progress?.lastUpdate) {
|
if (!task.progress?.lastUpdate) {
|
||||||
|
if (shouldSkipInactivityTimeout) continue
|
||||||
if (sessionIsRunning) continue
|
if (sessionIsRunning) continue
|
||||||
if (sessionMissing && !sessionGone) continue
|
if (sessionMissing && !sessionGone) continue
|
||||||
const effectiveTimeout = sessionGone ? sessionGoneTimeoutMs : messageStalenessMs
|
const effectiveTimeout = sessionGone ? sessionGoneTimeoutMs : messageStalenessMs
|
||||||
@@ -183,6 +189,7 @@ export async function checkAndInterruptStaleTasks(args: {
|
|||||||
}
|
}
|
||||||
|
|
||||||
if (sessionIsRunning) continue
|
if (sessionIsRunning) continue
|
||||||
|
if (shouldSkipInactivityTimeout) continue
|
||||||
|
|
||||||
if (runtime < MIN_RUNTIME_BEFORE_STALE_MS) continue
|
if (runtime < MIN_RUNTIME_BEFORE_STALE_MS) continue
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user