Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 3 additions & 2 deletions packages/api/src/agent-side/route.ts
Original file line number Diff line number Diff line change
Expand Up @@ -30,8 +30,9 @@ export function handleAgentListTasksRoute(request: AgentRequest): Promise<Result

/**
* `GET /api/agent/tasks/:taskId`: the task while it is not finished, as the agent pages open it
* from their address; unknown is 404, finished, or under a task finished, 409, as a run for it
* would be. Only read: one left open is taken up again by the next run made for it.
* from their address; unknown is 404, finished, under a task finished, or left open more than 20
* minutes ago, 409, as a run for it would be. Only read: one left open more recently is taken up
* again by the next run made for it.
*/
export function handleAgentGetTaskRoute(request: AgentRequest): Promise<Result<AgentTask, ApiError>> {
return getAgentOpenTask(request.agent, request.params.taskId ?? '');
Expand Down
60 changes: 56 additions & 4 deletions packages/api/src/agent-side/tasks.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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'});
Expand All @@ -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));
Expand All @@ -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'});
Expand Down Expand Up @@ -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', () => {
Expand Down Expand Up @@ -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');
Expand Down Expand Up @@ -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');
Expand All @@ -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();
});
});
29 changes: 21 additions & 8 deletions packages/api/src/agent-side/tasks.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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';
Expand All @@ -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<AgentTaskRow, 'id' | 'title'>) =>
`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<AgentTaskRow, 'id' | 'title'>) =>
`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 {
Expand Down Expand Up @@ -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<Result<AgentTask, ApiError>> {
return Do(async ($) => {
Expand All @@ -113,6 +120,9 @@ export function finishAgentTask(caller: AgentCaller, args: unknown): Promise<Res
return $(Err(ApiErr.validationError(`The outcome is ${outcome.length} characters long; keep it under ${MAX_OUTCOME}, a sentence or two.`)));
}
const task = $(requirePresent($(await markAgentTaskFinished(caller.agent.orgId, caller.agent.agentId, taskId, outcome)), ApiErr.notFound('task', taskId)));
if (task.finishedAt === null && taskEndedLongAgo(task, new Date())) {
return $(Err(ApiErr.conflict(leftOpenLongAgo(task))));
}
if (task.finishedAt === null) {
return $(Err(ApiErr.conflict(`Task ${taskId} (${task.title}) was left open, and a task it is part of was finished without it, so it can't be finished any more.`)));
}
Expand All @@ -125,7 +135,7 @@ export function finishAgentTask(caller: AgentCaller, args: unknown): Promise<Res
* finished already, or under a task finished, is refused as a conflict, saying to start a new one,
* so the agent pages can tell it from any other refusal of the call. The call counts as something
* done for the task and every task above it, so none is left open while it is worked on, and one
* left open is taken up again (@proxy/utils/tasks).
* left open is taken up again, unless that was more than 20 minutes ago (@proxy/utils/tasks).
*/
export function requireOpenTask(agent: SignedInAgent, taskId: string): Promise<Result<AgentTaskRow, ApiError>> {
return Do(async ($) => {
Expand All @@ -138,14 +148,17 @@ export function requireOpenTask(agent: SignedInAgent, taskId: string): Promise<R
});
}

// The agent's task `taskId`, read: unknown, or another agent's, is not found, and one finished a
// conflict.
// The agent's task `taskId`, read: unknown, or another agent's, is not found, and one finished, or
// left open more than 20 minutes ago, a conflict.
function readUnfinishedTask(agent: SignedInAgent, taskId: string): Promise<Result<AgentTaskRow, ApiError>> {
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;
});
}
14 changes: 12 additions & 2 deletions packages/api/src/harness/thread-task.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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',
Expand Down Expand Up @@ -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);
Expand Down
10 changes: 6 additions & 4 deletions packages/api/src/harness/thread-task.ts
Original file line number Diff line number Diff line change
@@ -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';
Expand All @@ -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
Expand All @@ -23,14 +25,14 @@ import type {HarnessStore} from './store';
/** What a sub-agent's task needs of the store. */
export type ThreadTaskStore = Pick<HarnessStore, 'getAgentTask' | 'createAgentTask' | 'setThreadTask'>;

/** 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<Result<string, ApiError>> {
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})));
Expand Down
3 changes: 2 additions & 1 deletion packages/api/src/harness/tool-job.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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));
Expand Down
2 changes: 1 addition & 1 deletion packages/db/prisma/schema.prisma
Original file line number Diff line number Diff line change
Expand Up @@ -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())
Expand Down
37 changes: 37 additions & 0 deletions packages/db/src/harness.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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});
});
});
Loading
Loading