diff --git a/packages/api/src/agent-side/route.ts b/packages/api/src/agent-side/route.ts index 19fdb2df..8aa59168 100644 --- a/packages/api/src/agent-side/route.ts +++ b/packages/api/src/agent-side/route.ts @@ -30,8 +30,9 @@ export function handleAgentListTasksRoute(request: AgentRequest): Promise> { return getAgentOpenTask(request.agent, request.params.taskId ?? ''); diff --git a/packages/api/src/agent-side/tasks.test.ts b/packages/api/src/agent-side/tasks.test.ts index 9d3e2533..8a9a6b3b 100644 --- a/packages/api/src/agent-side/tasks.test.ts +++ b/packages/api/src/agent-side/tasks.test.ts @@ -113,7 +113,8 @@ describe('startAgentTask', () => { }); test('a parent left open has the sub-task started under it, which takes it up again, not before', async () => { - getAgentTask.mockResolvedValue(Ok({...parentRow, lastActiveAt: new Date('2026-10-16T08:30:00Z')})); + // Left open at 08:55, a quarter of an hour ago. + getAgentTask.mockResolvedValue(Ok({...parentRow, lastActiveAt: new Date('2026-10-16T08:35:00Z')})); createAgentTask.mockResolvedValue(Ok({...row, parentId: 'task-0'})); const {startAgentTask} = await import('./tasks'); const result = await startAgentTask(caller, {title: 'Triage inbox', goal: 'Archive the newsletters', parent: 'task-0'}); @@ -123,6 +124,16 @@ describe('startAgentTask', () => { expect(result.unwrap()).toEqual({...response, parent: 'task-0'}); }); + test('a parent left open more than 20 minutes ago is refused as a conflict, saying to start a new task', async () => { + // Left open at 08:20, 50 minutes ago. + getAgentTask.mockResolvedValue(Ok({...parentRow, lastActiveAt: new Date('2026-10-16T08:00:00Z')})); + const {startAgentTask} = await import('./tasks'); + const result = await startAgentTask(caller, {title: 'Triage inbox', goal: 'Archive the newsletters', parent: 'task-0'}); + + expect(result.unwrapErr()).toMatchObject({kind: 'conflict', message: expect.stringContaining('Task task-0 (Plan the week) has ended: it was left open')}); + expect(createAgentTask).not.toHaveBeenCalled(); + }); + test('a parent finished between the check and the start is refused as a conflict too', async () => { getAgentTask.mockResolvedValue(Ok(parentRow)); createAgentTask.mockResolvedValue(Ok(null)); @@ -146,6 +157,10 @@ describe('listAgentOpenTasks', () => { }); describe('finishAgentTask', () => { + // Ten minutes after the task started. + beforeEach(() => setSystemTime(new Date('2026-10-16T09:10:00Z'))); + afterEach(() => setSystemTime()); + test('marks the agent’s task finished, with the outcome trimmed, and answers with when and what was done', async () => { const {finishAgentTask: finish} = await import('./tasks'); const result = await finish(caller, {task: 'task-1', outcome: ' Archived 12 newsletters\n'}); @@ -186,6 +201,20 @@ describe('finishAgentTask', () => { message: "Task task-1 (Triage inbox) was left open, and a task it is part of was finished without it, so it can't be finished any more.", }); }); + + test('a task left open more than 20 minutes ago is refused as a conflict, saying to start a new task', async () => { + // Left open at 09:20, 40 minutes ago, and not taken up again to be finished. + setSystemTime(new Date('2026-10-16T10:00:00Z')); + finishAgentTask.mockResolvedValue(Ok(row)); + const {finishAgentTask: finish} = await import('./tasks'); + const result = await finish(caller, {task: 'task-1', outcome: 'Archived 12 newsletters'}); + + expect(result.unwrapErr()).toEqual({ + kind: 'conflict', + statusCode: 409, + message: 'Task task-1 (Triage inbox) has ended: it was left open, with nothing done for it for too long. Start a new task for what you are doing now, and give its id as task.', + }); + }); }); describe('requireOpenTask', () => { @@ -218,15 +247,29 @@ describe('requireOpenTask', () => { expect(touchAgentTask).not.toHaveBeenCalled(); }); - test('a task left open, 20 minutes or more with nothing done for it, is taken up again by the call', async () => { + test('a task left open, 20 minutes or more with nothing done for it, is taken up again by the call within 20 minutes more', async () => { const {requireOpenTask} = await import('./tasks'); - setSystemTime(new Date('2026-10-16T11:00:00Z')); + // Left open at 09:20. + setSystemTime(new Date('2026-10-16T09:40:00Z')); const result = await requireOpenTask(agent, 'task-1'); expect(touchAgentTask).toHaveBeenCalledWith('org-1', 'agent-1', row); expect(result.unwrap()).toBe(row); }); + test('a task left open more than 20 minutes ago is refused as a conflict, saying to start a new one', async () => { + const {requireOpenTask} = await import('./tasks'); + setSystemTime(new Date('2026-10-16T09:40:01Z')); + const result = await requireOpenTask(agent, 'task-1'); + + expect(result.unwrapErr()).toEqual({ + kind: 'conflict', + statusCode: 409, + message: 'Task task-1 (Triage inbox) has ended: it was left open, with nothing done for it for too long. Start a new task for what you are doing now, and give its id as task.', + }); + expect(touchAgentTask).not.toHaveBeenCalled(); + }); + test('a task finished since it was read, or left open under a task finished, is refused as a conflict', async () => { touchAgentTask.mockResolvedValue(Ok(false)); const {requireOpenTask} = await import('./tasks'); @@ -256,7 +299,7 @@ describe('getAgentOpenTask', () => { test('only reads the task: one left open stays so, and one under a task finished is refused', async () => { const {getAgentOpenTask} = await import('./tasks'); - setSystemTime(new Date('2026-10-16T11:00:00Z')); + setSystemTime(new Date('2026-10-16T09:30:00Z')); const leftOpen = await getAgentOpenTask(agent, 'task-1'); isAgentTaskOpen.mockResolvedValue(Ok(false)); const under = await getAgentOpenTask(agent, 'task-1'); @@ -270,4 +313,13 @@ describe('getAgentOpenTask', () => { }); expect(touchAgentTask).not.toHaveBeenCalled(); }); + + test('refuses a task left open more than 20 minutes ago', async () => { + const {getAgentOpenTask} = await import('./tasks'); + setSystemTime(new Date('2026-10-16T11:00:00Z')); + const result = await getAgentOpenTask(agent, 'task-1'); + + expect(result.unwrapErr()).toMatchObject({kind: 'conflict', message: expect.stringContaining('has ended: it was left open')}); + expect(isAgentTaskOpen).not.toHaveBeenCalled(); + }); }); diff --git a/packages/api/src/agent-side/tasks.ts b/packages/api/src/agent-side/tasks.ts index 9b17d81d..c504425a 100644 --- a/packages/api/src/agent-side/tasks.ts +++ b/packages/api/src/agent-side/tasks.ts @@ -3,6 +3,7 @@ import type {z} from 'zod'; import {ApiErr, type ApiError, Do, parseSchema, requirePresent} from '@proxy/utils'; import type {SignedInAgent} from '@proxy/db/agent'; import {type AgentTask, type agentTaskListSchema, finishTaskBodySchema, startTaskBodySchema} from '@proxy/contracts/agent-side'; +import {taskEndedLongAgo} from '@proxy/utils/tasks'; import {type AgentTaskRow, createAgentTask, finishAgentTask as markAgentTaskFinished, getAgentTask, isAgentTaskOpen, listOpenAgentTasks, touchAgentTask} from '@proxy/db/task'; import type {AgentCaller} from './tools'; @@ -18,17 +19,22 @@ import type {AgentCaller} from './tools'; * sub-task for each part, with the task as its parent, so the tasks of one job form a tree the * person can follow. The parent is one of the agent's own tasks not finished. Finishing a task * finishes its sub-tasks still under way too. A task the agent left open, never finished with - * nothing done for it in 20 minutes, is taken up again by the next call made for it - * (@proxy/utils/tasks). + * nothing done for it in 20 minutes, is taken up again by the next call made for it within 20 + * minutes more; after that it stays ended, and a call for it is refused as one for a task finished + * is (@proxy/utils/tasks). */ const MAX_TITLE = 120; const MAX_GOAL = 2000; const MAX_OUTCOME = 2000; // Why a task can't be worked on any more, where it, or a task above it, was finished since it was -// read; or, left open, a task above it was finished without it (@proxy/utils/tasks). +// read, or left open too long ago to be taken up again; or, left open, a task above it was +// finished without it (@proxy/utils/tasks). const finishedAbove = (task: Pick) => `Task ${task.id} (${task.title}), or a task it is part of, is finished. Start a new task for what you are doing now, and give its id as task.`; +// Why a task left open more than 20 minutes ago can't be worked on, nor finished (@proxy/utils/tasks). +const leftOpenLongAgo = (task: Pick) => + `Task ${task.id} (${task.title}) has ended: it was left open, with nothing done for it for too long. Start a new task for what you are doing now, and give its id as task.`; function toResponse(task: AgentTaskRow): AgentTask { return { @@ -102,8 +108,9 @@ export function getAgentOpenTask(agent: SignedInAgent, taskId: string): Promise< * Marks the agent's task `{task}` finished, with `outcome`, what was done or found, where the * agent gives one; its sub-tasks still under way are finished with it. One finished already keeps * when it was, taking the outcome only where it has none, such as a sub-task finished with its - * parent. One left open (@proxy/utils/tasks) is taken up again to be finished, unless a task it is - * part of was finished without it, when it is refused as a conflict. + * parent. One left open (@proxy/utils/tasks) is taken up again to be finished, unless it was left + * open more than 20 minutes ago, or a task it is part of was finished without it, when it is + * refused as a conflict. */ export function finishAgentTask(caller: AgentCaller, args: unknown): Promise> { return Do(async ($) => { @@ -113,6 +120,9 @@ export function finishAgentTask(caller: AgentCaller, args: unknown): Promise> { return Do(async ($) => { @@ -138,14 +148,17 @@ export function requireOpenTask(agent: SignedInAgent, taskId: string): Promise> { return Do(async ($) => { const task = $(requirePresent($(await getAgentTask(agent.orgId, agent.agentId, taskId)), ApiErr.notFound('task', taskId))); if (task.finishedAt !== null) { return $(Err(ApiErr.conflict(`Task ${taskId} (${task.title}) is finished. Start a new task for what you are doing now, and give its id as task.`))); } + if (taskEndedLongAgo(task, new Date())) { + return $(Err(ApiErr.conflict(leftOpenLongAgo(task)))); + } return task; }); } diff --git a/packages/api/src/harness/thread-task.test.ts b/packages/api/src/harness/thread-task.test.ts index 9157bfec..89c11aa0 100644 --- a/packages/api/src/harness/thread-task.test.ts +++ b/packages/api/src/harness/thread-task.test.ts @@ -8,6 +8,7 @@ import {type FakeStore, fakeStore} from './fake-store'; import {ensureThreadTask} from './thread-task'; const now = new Date('2026-10-21T10:00:00Z'); +const MINUTE = 60 * 1000; // A sub-agent's thread, at work on its task. const thread: ThreadRow = { id: 'thread-1', @@ -44,14 +45,23 @@ beforeEach(() => { }); describe('ensureThreadTask', () => { - test('keeps the task it has while it is not finished, left open included', async () => { - store.getAgentTask.mockResolvedValue(Ok({...task, lastActiveAt: new Date('2026-10-20T10:00:00Z')})); + test('keeps the task it has while it is not finished, left open 20 minutes ago or less included', async () => { + store.getAgentTask.mockResolvedValue(Ok({...task, lastActiveAt: new Date(Date.now() - 30 * MINUTE)})); const result = await ensureThreadTask(store, thread); expect(result.unwrap()).toBe('task-1'); expect(store.createAgentTask).not.toHaveBeenCalled(); }); + test('one left open more than 20 minutes ago is replaced with a new one for the same work', async () => { + store.getAgentTask.mockResolvedValue(Ok({...task, lastActiveAt: new Date(Date.now() - 41 * MINUTE)})); + const result = await ensureThreadTask(store, thread); + + expect(result.unwrap()).toBe('task-2'); + expect(store.createAgentTask).toHaveBeenCalledWith(expect.objectContaining({title: 'Find the invoices', parentId: null})); + expect(store.setThreadTask).toHaveBeenCalledWith('thread-1', 'task-2'); + }); + test('one finished while the sub-agent is at work is replaced with a new one for the same work', async () => { store.getAgentTask.mockResolvedValue(Ok({...task, finishedAt: new Date('2026-10-21T09:00:00Z')})); const result = await ensureThreadTask(store, thread); diff --git a/packages/api/src/harness/thread-task.ts b/packages/api/src/harness/thread-task.ts index 0f6d57d4..5c69a7c6 100644 --- a/packages/api/src/harness/thread-task.ts +++ b/packages/api/src/harness/thread-task.ts @@ -1,6 +1,7 @@ import {Err, type Result} from 'ts-results-es'; import {type ApiError, ApiErr, Do, requirePresent} from '@proxy/utils'; +import {taskEndedLongAgo} from '@proxy/utils/tasks'; import type {ThreadRow} from '@proxy/db/harness'; import {taskFor} from './conversation'; @@ -12,8 +13,9 @@ import type {HarnessStore} from './store'; * started from is, so what the sub-agent did is theirs alone to see. A sub-agent starts with one, * made as it is (conversation.ts subAgentStart). It is finished each time the sub-agent stops, and * under way again once a message takes the sub-agent up (db/harness). One left open, as when it - * slept long, is taken up again by its next call (@proxy/utils/tasks); one finished by finish_task - * from the agent's login elsewhere while the sub-agent is still at work is replaced with a new one + * slept a while, is taken up again by its next call, unless it was left open more than 20 minutes + * before, as when it slept long (@proxy/utils/tasks); that one, and one finished by finish_task + * from the agent's login elsewhere while the sub-agent is still at work, is replaced with a new one * for the same piece of work, the title and goal the one before had, once the sub-agent makes a * call again, since the gate refuses calls for it and the sub-agent has no start_task of its own * (tool-surface). The agent talking with the person has none: it makes no calls but to @@ -23,14 +25,14 @@ import type {HarnessStore} from './store'; /** What a sub-agent's task needs of the store. */ export type ThreadTaskStore = Pick; -/** The sub-agent's task, not finished: the one it has, or a new one in its place. */ +/** The sub-agent's task, not finished nor left open long ago: the one it has, or a new one in its place. */ export function ensureThreadTask(store: ThreadTaskStore, thread: ThreadRow): Promise> { return Do(async ($) => { if (thread.taskId === null) { return $(Err(ApiErr.conflict('The agent talking with the person makes no calls of its own'))); } const task = $(requirePresent($(await store.getAgentTask(thread.orgId, thread.agentId, thread.taskId)), ApiErr.notFound('task', thread.taskId))); - if (task.finishedAt === null) { + if (task.finishedAt === null && !taskEndedLongAgo(task, new Date())) { return task.id; } const made = $(await store.createAgentTask(taskFor(thread, {title: task.title, goal: task.goal}))); diff --git a/packages/api/src/harness/tool-job.test.ts b/packages/api/src/harness/tool-job.test.ts index 15642cab..c5d60e62 100644 --- a/packages/api/src/harness/tool-job.test.ts +++ b/packages/api/src/harness/tool-job.test.ts @@ -90,7 +90,8 @@ beforeEach(() => { createdAt: now, finishedAt: null, outcome: null, - lastActiveAt: now, + // Just now, whenever the test runs. + lastActiveAt: new Date(), }), ); getJobThread.mockResolvedValue(Ok(thread)); diff --git a/packages/db/prisma/schema.prisma b/packages/db/prisma/schema.prisma index d7da8ee5..a3b30afa 100644 --- a/packages/db/prisma/schema.prisma +++ b/packages/db/prisma/schema.prisma @@ -449,7 +449,7 @@ model AgentTask { personaId String? // When it, or the last of its sub-tasks at any depth, was started or had a request made for it: // one never finished is taken as left open 20 minutes after, until the next request takes it up - // again (@proxy/utils/tasks). Never before any of its sub-tasks', so a parent is never left open + // again, which it can for 20 minutes more (@proxy/utils/tasks). Never before any of its sub-tasks', so a parent is never left open // while one is under way. Set with createdAt, and on it and every task above it when a sub-task // starts, a request is made or it is finished. lastActiveAt DateTime @default(now()) diff --git a/packages/db/src/harness.test.ts b/packages/db/src/harness.test.ts index e54883d6..f62c9691 100644 --- a/packages/db/src/harness.test.ts +++ b/packages/db/src/harness.test.ts @@ -687,5 +687,42 @@ describe.skipIf(!databaseUrl)('the harness, against the database', () => { ), ), ).toBe(false); + + // One that stopped more than 20 minutes ago keeps its task finished as it was, and takes a + // message up with a new task for the same work. + const anHourAgo = new Date(Date.now() - 60 * 60 * 1000); + await db.agentTask.update({ + where: {id: sub.taskId ?? ''}, + data: {finishedAt: anHourAgo, lastActiveAt: anHourAgo}, + }); + const longAfter = unwrap(await claimJob('worker-o', 60_000)); + if (!longAfter || longAfter.threadId !== thread.id) { + throw new Error(`expected the parent’s turn, got ${JSON.stringify(longAfter)}`); + } + unwrap( + await completeInference( + longAfter.id, + 'worker-o', + thread.id, + parentTurn([{id: 'm4', name: 'send_to_subagent', arguments: '{}'}]), + { + plans: { + m4: {kind: 'send_to_subagent', subAgentId: sub.id, entry: say('Any from July?'), sent}, + }, + }, + readyForInference, + ), + ); + const [renewed] = unwrap(await listSubAgents(thread.id)); + if (!renewed) { + throw new Error('expected the sub-agent'); + } + expect(renewed.state).toEqual({kind: 'working'}); + expect(renewed.taskId).not.toBe(sub.taskId); + expect(await taskOf(sub)).toMatchObject({ + outcome: 'The sub-agent failed: boom', + finishedAt: anHourAgo, + }); + expect(await taskOf(renewed)).toMatchObject({...task, finishedAt: null, outcome: null}); }); }); diff --git a/packages/db/src/harness.ts b/packages/db/src/harness.ts index 9ae331fa..467998ce 100644 --- a/packages/db/src/harness.ts +++ b/packages/db/src/harness.ts @@ -25,6 +25,7 @@ import { threadState, type ToolCall, } from '@proxy/utils/harness'; +import {taskEndedLongAgo} from '@proxy/utils/tasks'; import {db} from '.'; import {signedInAgentSelect, type SignedInAgent, toSignedInAgent} from './signed-in-agent'; @@ -1171,7 +1172,7 @@ async function createSubAgent( } // Lands a message from the agent that started the sub-agent in the sub-agent's log, and wakes the -// sub-agent: one that had stopped takes it up, its task under way again; one at work reads it once +// sub-agent: one that had stopped takes it up, its task under way again (takeUpSubAgentTask); one at work reads it once // its current call is answered; and one asleep is woken now, its sleep's job due now, which answers // the sleep as any wake-up does, the message after it. The sub-agent as it was before the message; // null when the thread has no such sub-agent. With the conversation locked. @@ -1192,10 +1193,7 @@ async function messageSubAgent( const {jobs, ...thread} = found; const now = new Date(); if (thread.taskId !== null) { - await tx.agentTask.updateMany({ - where: {id: thread.taskId}, - data: {finishedAt: null, outcome: null, lastActiveAt: now}, - }); + await takeUpSubAgentTask(tx, {...thread, taskId: thread.taskId}, now); } // Only a sleep's job is due later; one a worker has claimed is no longer queued, and answers as it // would. @@ -1207,6 +1205,51 @@ async function messageSubAgent( return {...thread, state: threadState(jobs, now)}; } +// Puts the sub-agent's task under way again from `now`, as a message wakes it. One that ended, +// finished or left open, more than 20 minutes before stays as it was, and a new one for the same +// work takes its place, so a task long over isn't drawn as under way again (@proxy/utils/tasks). +// With the conversation locked. +async function takeUpSubAgentTask( + tx: Tx, + thread: ThreadRow & {taskId: string}, + now: Date, +): Promise { + const task = await tx.agentTask.findUnique({ + where: {id: thread.taskId}, + select: { + orgId: true, + agentId: true, + ownerUserId: true, + personaId: true, + title: true, + goal: true, + via: true, + finishedAt: true, + lastActiveAt: true, + }, + }); + if (task === null) { + return; + } + if (!taskEndedLongAgo(task, now)) { + await tx.agentTask.updateMany({ + where: {id: thread.taskId}, + data: {finishedAt: null, outcome: null, lastActiveAt: now}, + }); + return; + } + const {finishedAt: _finishedAt, lastActiveAt: _lastActiveAt, ...work} = task; + const fresh = await tx.agentTask.create({ + data: {...work, parentId: null, createdAt: now, lastActiveAt: now}, + select: {id: true}, + }); + await tx.agentThread.update({ + where: {id: thread.id}, + data: {taskId: fresh.id}, + select: {id: true}, + }); +} + // What a call is answered with at once, by its plan, done. With the conversation locked. async function answerAtOnce( tx: Tx, diff --git a/packages/db/src/task.test.ts b/packages/db/src/task.test.ts index 23b2c220..2a441c71 100644 --- a/packages/db/src/task.test.ts +++ b/packages/db/src/task.test.ts @@ -20,7 +20,7 @@ import { // is skipped. Everything made here is one organization's, deleted at the end. const databaseUrl = process.env.DATABASE_URL; -const HOUR = 60 * 60 * 1000; +const MINUTE = 60 * 1000; function unwrap(result: Result): T { if (result.isErr()) { @@ -64,8 +64,12 @@ describe.skipIf(!databaseUrl)('tasks, against the database', () => { await db.$disconnect(); }); - // A task with a sub-task, both started and last worked on an hour ago, so both left open. - async function leftOpenLine(title: string): Promise<{parentId: string; childId: string}> { + // A task with a sub-task, both started and last worked on half an hour ago, so both left open ten + // minutes ago, or `ago` before now where given. + async function leftOpenLine( + title: string, + ago = 30 * MINUTE, + ): Promise<{parentId: string; childId: string}> { const base = { orgId, agentId, @@ -80,10 +84,10 @@ describe.skipIf(!databaseUrl)('tasks, against the database', () => { if (!parent || !child) { throw new Error('the tasks were not made'); } - const anHourAgo = new Date(Date.now() - HOUR); + const then = new Date(Date.now() - ago); await db.agentTask.updateMany({ where: {id: {in: [parent.id, child.id]}}, - data: {createdAt: anHourAgo, lastActiveAt: anHourAgo}, + data: {createdAt: then, lastActiveAt: then}, }); return {parentId: parent.id, childId: child.id}; } @@ -130,6 +134,20 @@ describe.skipIf(!databaseUrl)('tasks, against the database', () => { expect(finished?.lastActiveAt).toEqual(before?.lastActiveAt); }); + test('a task left open more than 20 minutes ago is not taken up again, nor finished', async () => { + const {parentId, childId} = await leftOpenLine('Sort the photos', 41 * MINUTE); + const before = await read(childId); + + expect(unwrap(await isAgentTaskOpen(orgId, agentId, childId))).toBe(false); + expect(unwrap(await touchAgentTask(orgId, agentId, before))).toBe(false); + const finished = unwrap(await finishAgentTask(orgId, agentId, parentId, 'Sorted them')); + + expect(finished?.finishedAt).toBeNull(); + expect(finished?.outcome).toBeNull(); + expect((await read(childId)).lastActiveAt).toEqual(before.lastActiveAt); + expect((await read(parentId)).lastActiveAt).toEqual(before.lastActiveAt); + }); + test('a task finished is not taken up again', async () => { const {parentId} = await leftOpenLine('Reply to Dana'); unwrap(await finishAgentTask(orgId, agentId, parentId, null)); diff --git a/packages/db/src/task.ts b/packages/db/src/task.ts index ba714446..f8e9b621 100644 --- a/packages/db/src/task.ts +++ b/packages/db/src/task.ts @@ -2,7 +2,7 @@ import {type AuditVia, Prisma} from '@prisma/client'; import {Ok, type Result} from 'ts-results-es'; import {type ApiError, wrapDb} from '@proxy/utils'; -import {TASK_LEFT_OPEN_AFTER_MS} from '@proxy/utils/tasks'; +import {TASK_LEFT_OPEN_AFTER_MS, taskEndedLongAgo} from '@proxy/utils/tasks'; import {db} from '.'; import {type Viewer, visibleTo} from './viewer'; @@ -24,7 +24,8 @@ export type AgentTaskRow = { // What was done or found, as the agent said when it finished; null where it said nothing. outcome: string | null; // When it, or one of its sub-tasks at any depth, was last started or had a request made for it: - // one never finished is taken as left open 20 minutes after (@proxy/utils/tasks). + // one never finished is taken as left open 20 minutes after, and can't be taken up again 20 + // minutes after that (@proxy/utils/tasks). lastActiveAt: Date; }; @@ -65,7 +66,7 @@ export type NewAgentTask = { * the sub-task is made, so none can be finished in between and leave the sub-task open, and each * takes the sub-task's start as the last thing done for it (`lastActiveAt`), so none is left open * before it, and one left open is taken up again; `Ok(null)` means the parent, or a task above it, - * is finished (@proxy/utils/tasks). `ownerUserId` is the member whose own the agent is, who alone + * is finished, or was left open too long ago to be taken up again (@proxy/utils/tasks). `ownerUserId` is the member whose own the agent is, who alone * sees the task; null for a shared agent's. */ export async function createAgentTask( @@ -99,7 +100,7 @@ export async function createAgentTask( * Takes now as the last thing done for the agent's task and every task above it, as a request made * for it is (`lastActiveAt`), so none is left open while the agent is still at work on it, and one * left open is taken up again (@proxy/utils/tasks). `Ok(false)` means it, or a task above it, is - * finished, or the agent has no such task. One read unfinished and done for within the last + * finished, or was left open too long ago to be taken up again, or the agent has no such task. One read unfinished and done for within the last * minute isn't written again, nor the tasks above it read: none of them can have been finished * since without finishing it too (finishSubTasks). */ @@ -120,8 +121,9 @@ export async function touchAgentTask( } /** - * Whether the agent's task and every task above it are not finished, read without taking any of - * them up; `Ok(false)` where the agent has no such task. + * Whether the agent's task and every task above it are not finished, nor left open too long ago to + * be taken up again, read without taking any of them up; `Ok(false)` where the agent has no such + * task. */ export async function isAgentTaskOpen( orgId: string, @@ -130,11 +132,11 @@ export async function isAgentTaskOpen( ): Promise> { return wrapDb(async () => { const line = await lineOf(db, {orgId, agentId, taskId}, {lock: false}); - return isOpen(line, taskId); + return isOpen(line, taskId, new Date()); }); } -type LineRow = {id: string; finishedAt: Date | null}; +type LineRow = {id: string; finishedAt: Date | null; lastActiveAt: Date}; // The task and every task above it, from the top down; with `lock`, locked for the rest of the // transaction in that order, the one finishing a task locks its sub-tasks in (finishSubTasks), so @@ -153,20 +155,25 @@ function lineOf( FROM "AgentTask" AS "task" JOIN "line" ON "task"."id" = "line"."parentId" ) - SELECT "task"."id", "task"."finishedAt" + SELECT "task"."id", "task"."finishedAt", "task"."lastActiveAt" FROM "AgentTask" AS "task" JOIN "line" ON "task"."id" = "line"."id" ORDER BY "line"."depth" DESC ${lock ? Prisma.sql`FOR UPDATE OF "task"` : Prisma.empty}`; } -// Whether the line of `taskId` holds it, with none of it finished. -function isOpen(line: LineRow[], taskId: string): boolean { - return line.some((row) => row.id === taskId) && line.every((row) => row.finishedAt === null); +// Whether the line of `taskId` holds it, with none of it finished, nor left open more than 20 +// minutes before `now` (@proxy/utils/tasks). +function isOpen(line: LineRow[], taskId: string, now: Date): boolean { + return ( + line.some((row) => row.id === taskId) && + line.every((row) => row.finishedAt === null && !taskEndedLongAgo(row, now)) + ); } // Locks the task and every task above it, and takes `now` as the last thing done for each; false, -// with none changed, where any of them is finished. So a task taken up again takes up every task +// with none changed, where any of them is finished, or was left open too long ago to be taken up +// again. So a task taken up again takes up every task // above it too, none left open before it; and one left open when its parent was finished, which // it was not finished with (finishSubTasks), stays as it is. async function touchLine( @@ -175,7 +182,7 @@ async function touchLine( now: Date, ): Promise { const line = await lineOf(tx, {orgId, agentId, taskId}, {lock: true}); - if (!isOpen(line, taskId)) { + if (!isOpen(line, taskId, now)) { return false; } // Where another instance's clock is ahead, one may have set a later time already. @@ -201,7 +208,8 @@ export async function getAgentTask( * Marks the agent's task finished now, with what was done where the agent said, and its sub-tasks * still under way finished with it, in one transaction. Finishing is something done for it and * every task above it, as a request is (touchLine), so one left open (@proxy/utils/tasks) is taken - * up again to be finished. One already finished keeps the time it was, and takes the outcome only + * up again to be finished, unless it was left open too long ago, when it stays as it is. One + * already finished keeps the time it was, and takes the outcome only * where it has none, as a sub-task finished with its parent does; one left open under a task * finished without it stays as it is, ended when it was left open. `Ok(null)` means the agent has * no such task. diff --git a/packages/docs/content/agents/mcp.mdx b/packages/docs/content/agents/mcp.mdx index 2c1d553d..10812910 100644 --- a/packages/docs/content/agents/mcp.mdx +++ b/packages/docs/content/agents/mcp.mdx @@ -88,8 +88,9 @@ in a sentence or two. Every call it makes after that names the task, and a call a task that is finished, is refused. When the agent has done what you asked, or given it up, it finishes the task, saying what it did, and starts a new one for the next thing you ask for. Agents don't always say so: a task the agent never finishes is taken as left open once 20 minutes pass -with nothing done for it or its sub-tasks. If the agent comes back to it, say after you stepped -away, its next call picks the task up again where it was. An agent can have any number of tasks +with nothing done for it or its sub-tasks. If the agent comes back to it within 20 minutes more, +its next call picks the task up again where it was; after that the task stays ended, and the agent +starts a new one for what it does next. An agent can have any number of tasks under way at once, each with its own requests. An agent may split what you asked for among sub-agents of its own, which sign in as it does and diff --git a/packages/utils/src/tasks.test.ts b/packages/utils/src/tasks.test.ts index d9435c79..c2d4dd9e 100644 --- a/packages/utils/src/tasks.test.ts +++ b/packages/utils/src/tasks.test.ts @@ -1,6 +1,6 @@ import {describe, expect, test} from 'bun:test'; -import {taskEndedAt} from './tasks'; +import {taskEndedAt, taskEndedLongAgo} from './tasks'; const lastActiveAt = new Date('2026-10-16T09:00:00Z'); @@ -27,3 +27,33 @@ describe('taskEndedAt', () => { ); }); }); + +describe('taskEndedLongAgo', () => { + test('is false while the task is under way', () => { + expect( + taskEndedLongAgo({lastActiveAt, finishedAt: null}, new Date('2026-10-16T09:19:59Z')), + ).toBe(false); + }); + + test('is false for a task left open 20 minutes ago or less', () => { + expect( + taskEndedLongAgo({lastActiveAt, finishedAt: null}, new Date('2026-10-16T09:40:00Z')), + ).toBe(false); + }); + + test('is true for a task left open more than 20 minutes ago', () => { + expect( + taskEndedLongAgo({lastActiveAt, finishedAt: null}, new Date('2026-10-16T09:40:01Z')), + ).toBe(true); + }); + + test('counts from when a finished task was finished', () => { + const finishedAt = new Date('2026-10-16T10:00:00Z'); + expect(taskEndedLongAgo({lastActiveAt, finishedAt}, new Date('2026-10-16T10:20:00Z'))).toBe( + false, + ); + expect(taskEndedLongAgo({lastActiveAt, finishedAt}, new Date('2026-10-16T10:20:01Z'))).toBe( + true, + ); + }); +}); diff --git a/packages/utils/src/tasks.ts b/packages/utils/src/tasks.ts index 4c0cc19a..baa3c58d 100644 --- a/packages/utils/src/tasks.ts +++ b/packages/utils/src/tasks.ts @@ -4,10 +4,15 @@ // still at work on part of it, and a parent is never left open before any of its sub-tasks. The web // app shows one left open as having ended when anything was last done for it. It is taken up again, // with every task above it, by the next call made for it, a sub-task started under it or the agent -// finishing it, unless a task it is part of was finished meanwhile. +// finishing it, unless a task it is part of was finished meanwhile, or it was left open more than +// 20 minutes before (`taskEndedLongAgo`): one ended that long stays ended, and the agent starts a +// new task for what it does next, so a task long over isn't drawn as under way again. export const TASK_LEFT_OPEN_AFTER_MS = 20 * 60 * 1000; +// How long after a task ended it can still be taken up again. +export const TASK_REOPENABLE_FOR_MS = 20 * 60 * 1000; + /** * When a task ended: when it was finished, or, where it never was and that is past, 20 minutes * after anything was last done for it or its sub-tasks (`lastActiveAt`); null while it is under way. @@ -22,3 +27,18 @@ export function taskEndedAt( const leftOpenAt = new Date(task.lastActiveAt.getTime() + TASK_LEFT_OPEN_AFTER_MS); return leftOpenAt <= now ? leftOpenAt : null; } + +/** + * Whether the task ended, finished or left open, more than 20 minutes before `now`, so it can't be + * taken up again. + */ +export function taskEndedLongAgo( + task: {lastActiveAt: Date; finishedAt: Date | null}, + now: Date, +): boolean { + const ended = taskEndedAt(task, now); + if (ended === null) { + return false; + } + return now.getTime() - ended.getTime() > TASK_REOPENABLE_FOR_MS; +}