A task is finished when it reaches done, not when a run ends

The two real goal tasks from July 9 each got exactly one session: the run
completed, the agent never moved the task off backlog, and the heartbeat's
"has a completed run" exclusion locked them out forever. The agents could
not have closed them anyway — update_task_status requires a task_id the
prompt never carried.

The progress contract, made explicit end to end:
- The task prompt now carries "Task id: ..." and tells the agent exactly how
  to end a session: done via update_task_status when the work is complete,
  in_progress with a note on what remains when it is not.
- The heartbeat re-enqueues any open task with no run in flight — a
  completed run that left its task open earns a follow-up session, with the
  prompt stating how many prior sessions were spent so the agent continues
  instead of redoing work. The retry budget (3 non-transient runs) still
  bounds every task; a task that spends it parks with one log line naming it.
- Scripted agents in the integration suites now close their tasks the way
  the contract expects real ones to.
This commit is contained in:
ZacharyZhang-NY committed 2026-07-09 19:17:44 -04:00
1 parent 2526633bfb
commit 9c85b5ee3c
4 files changed
+147 -14

No files matched your search

+33 -6
View File
@@ -67,6 +67,8 @@ export class TaskExecutionService {
private quotaBlockedProviders = new Map<string, string>();
/** Providers parked after a transient failure (usage/rate limit, overload): id -> park state. */
private providerCooldowns = new Map<string, { until: number; label: string }>();
/** Tasks already logged as budget-spent, so the park is announced once, not every 5s tick. */
private spentTasksLogged = new Set<string>();
constructor(
private readonly db: DatabaseSync,
@@ -174,6 +176,7 @@ export class TaskExecutionService {
managerName: this.agentName(orgId, agent.manager_agent_id),
chatContext,
inboxContext,
priorCompletedRuns: this.completedRunCount(task.id, runId),
}),
toolContext,
externalTools,
@@ -309,10 +312,13 @@ export class TaskExecutionService {
}
/**
* Heartbeat: enqueues one run per actionable assigned task of a
* heartbeat-enabled agent with an enabled provider that has not run yet, so
* CEO-delegated work starts on its own. Each task is enqueued at most once
* (guarded by existing runs), preventing runaway re-execution.
* Heartbeat: keeps every open assigned task of a heartbeat-enabled agent
* moving. A task is enqueued whenever it has no run in flight and its
* status is still open — a completed run that left the task open gets a
* follow-up session (the prompt carries the prior-run count), because a
* task is finished when it reaches done, not when a run happens to end.
* The retry budget bounds this: at most TASK_RETRY_BUDGET non-transient
* runs per task, then it parks for a human or manager to intervene.
*
* Three rules keep the loop self-healing when a provider stops serving:
* tasks on a provider over its monthly window are parked (not enqueued into
@@ -335,7 +341,7 @@ export class TaskExecutionService {
and a.lifecycle = 'active'
and p.enabled = 1
and coalesce(u.request_count, 0) < p.monthly_request_limit
and not exists (select 1 from task_runs r where r.task_id = t.id and r.state in ('queued', 'running', 'completed'))
and not exists (select 1 from task_runs r where r.task_id = t.id and r.state in ('queued', 'running'))
`).all(window.start) as Array<{ task_id: string; org_id: string; provider_id: string }>;
this.observeQuotaBlocks(window.start);
const cooling = new Set(this.cooledProviderIds());
@@ -345,7 +351,10 @@ export class TaskExecutionService {
if (cooling.has(row.provider_id)) continue;
const attempts = (attemptMessages.all(row.task_id) as Array<{ error_message: string | null }>)
.filter((run) => !isTransientProviderFailure(run.error_message)).length;
if (attempts >= TASK_RETRY_BUDGET) continue;
if (attempts >= TASK_RETRY_BUDGET) {
this.observeSpentTask(row.task_id);
continue;
}
try {
created.push(this.enqueueRun(row.org_id, row.task_id));
} catch {
@@ -355,6 +364,18 @@ export class TaskExecutionService {
return created;
}
/**
* A task that spent its whole retry budget without reaching done parks
* silently otherwise — one log line makes the stall diagnosable from the
* main-process log instead of looking like an idle queue.
*/
private observeSpentTask(taskId: string): void {
if (this.spentTasksLogged.has(taskId)) return;
this.spentTasksLogged.add(taskId);
const task = this.db.prepare("select title from tasks where id = ?").get(taskId) as { title: string } | undefined;
console.warn(`[task-queue] task "${task?.title ?? taskId}" spent its ${TASK_RETRY_BUDGET}-run budget without reaching done; it stays parked until someone closes or reassigns it.`);
}
/**
* Parks a provider after it reported a transient failure (usage limit, rate
* limit, overload), so the queue stops feeding it doomed runs every tick.
@@ -495,6 +516,12 @@ export class TaskExecutionService {
};
}
private completedRunCount(taskId: string, excludingRunId: string): number {
const row = this.db.prepare("select count(*) as n from task_runs where task_id = ? and state = 'completed' and id <> ?")
.get(taskId, excludingRunId) as { n: number };
return row.n;
}
private agentHasReports(orgId: string, agentId: string): boolean {
const row = this.db.prepare("select 1 from agents where org_id = ? and manager_agent_id = ? and lifecycle = 'active' limit 1")
.get(orgId, agentId);
+11 -2
View File
@@ -165,11 +165,14 @@ export function buildTaskPrompt(input: {
managerName: string | null;
chatContext: ChatContextBlock | null;
inboxContext: InboxContextBlock | null;
/** Completed runs this task already spent without reaching done. */
priorCompletedRuns: number;
}): string {
const lines = [
"Execute this AgentCompany task.",
"",
`Task: ${input.task.title}`,
`Task id: ${input.task.id}`,
`Description: ${input.task.description}`,
`Status: ${input.task.status}`,
`Priority: ${input.task.priority}`,
@@ -223,17 +226,23 @@ export function buildTaskPrompt(input: {
"Respond to anyone above who needs your input before starting the task.",
);
}
if (input.priorCompletedRuns > 0) {
lines.push(
"",
`This task already spent ${input.priorCompletedRuns} completed run(s) without being closed. Check read_messages and the existing deliverables first, continue from where the last run stopped instead of redoing finished work, and if the work is actually complete just close the task.`,
);
}
if (input.isManager && !input.hasReports) {
lines.push(
"",
"Your team is currently empty. Hire 3-5 direct reports for the roles this goal needs (hire_agent), create and assign their first tasks (create_task), and write the initial operating plan into the workspace (write_deliverable). Then use send_message with /handoff @Name and attachment_file_ids to pass the operating plan to the relevant report. Brief EACH report in a direct message (send_message with to=\"Their Name\") about the work you assigned them — note the thread_id each returns so you can follow up in the same conversation. Post one company-wide kickoff to the #general channel. Then summarize.",
`Your team is currently empty. Hire 3-5 direct reports for the roles this goal needs (hire_agent), create and assign their first tasks (create_task), and write the initial operating plan into the workspace (write_deliverable). Then use send_message with /handoff @Name and attachment_file_ids to pass the operating plan to the relevant report. Brief EACH report in a direct message (send_message with to="Their Name") about the work you assigned them — note the thread_id each returns so you can follow up in the same conversation. Post one company-wide kickoff to the #general channel. Then close this kickoff with update_task_status(task_id="${input.task.id}", status="done") and summarize.`,
);
} else {
const manager = input.managerName ?? "your manager";
const managerReference = input.managerName ? `${manager} (${agentMention(input.managerName)})` : manager;
lines.push(
"",
`First, read_messages to check if anyone has said something you should respond to — reply to them like a colleague, not a report. Then do the work: write real output with write_deliverable, raise decisions and blockers with ${managerReference} in a direct message (send_message with to=), keep team-wide topics in channels, and when the work is genuinely done move the task to done. Hand files over with /handoff and attachment_file_ids. Keep conversation going — ask questions, share your thinking, push back if something seems wrong.`,
`First, read_messages to check if anyone has said something you should respond to — reply to them like a colleague, not a report. Then do the work: write real output with write_deliverable, raise decisions and blockers with ${managerReference} in a direct message (send_message with to=), and keep team-wide topics in channels. Hand files over with /handoff and attachment_file_ids. Keep conversation going — ask questions, share your thinking, push back if something seems wrong. Before you finish: if the work is genuinely done, call update_task_status(task_id="${input.task.id}", status="done"); if it needs another session, move it to "in_progress" and say in the thread what remains — an open task will be run again.`,
);
}
return lines.join("\n");
+38 -6
View File
@@ -22,7 +22,7 @@ function call(name: string, args: Record<string, unknown>): ToolCall {
return { id: `call_${randomUUID()}`, name, arguments: args };
}
/** Scripts the CEO: staff two reports, delegate work, hand off the file in chat, then stop. */
/** Scripts the CEO: staff two reports, delegate work, hand off the file in chat, close the kickoff, then stop. */
function scriptedCeo(): (input: ProviderToolTurnInput) => Promise<AgentTurnResult> {
const script: Array<{ text: string; toolCalls: ToolCall[] }> = [
{ text: "Staffing.", toolCalls: [
@@ -36,6 +36,7 @@ function scriptedCeo(): (input: ProviderToolTurnInput) => Promise<AgentTurnResul
call("write_deliverable", { path: "plan/operating-plan.md", content: "# OrbitOps Operating Plan\n\n- Validate demand\n- Launch booking site\n" }),
] },
{ text: "Handing off the plan.", toolCalls: [] },
{ text: "Closing the kickoff.", toolCalls: [] },
{ text: "Team staffed, work delegated, plan written.", toolCalls: [] },
];
let index = 0;
@@ -55,6 +56,17 @@ function scriptedCeo(): (input: ProviderToolTurnInput) => Promise<AgentTurnResul
rawResponseJson: JSON.stringify({ id: `resp_${index}` }),
};
}
if (index === 4) {
// The prompt carries the task id so the agent can actually close its task.
const firstUser = input.messages.find((message) => message.role === "user");
const taskId = firstUser ? /Task id: (\S+)/.exec(firstUser.text)?.[1] : undefined;
index += 1;
return {
text: "Closing the kickoff.",
toolCalls: taskId ? [call("update_task_status", { task_id: taskId, status: "done" })] : [],
rawResponseJson: JSON.stringify({ id: `resp_${index}` }),
};
}
const turn = script[index] ?? { text: "done", toolCalls: [] };
index += 1;
return { text: turn.text, toolCalls: turn.toolCalls, rawResponseJson: JSON.stringify({ id: `resp_${index}` }) };
@@ -150,6 +162,11 @@ describe("autonomous bootstrap run", () => {
const run = state.taskRuns.find((r) => r.id === claimed!.id);
expect(run?.state).toBe("completed");
// The CEO closed the kickoff task using the task id the prompt carries.
const kickoffTask = connection.db.prepare("select status from tasks where id = (select task_id from task_runs where id = ?)")
.get(claimed!.id) as { status: string };
expect(kickoffTask).toEqual({ status: "done" });
// Stage 4 heartbeat: the CEO-delegated task (assigned to a hired, heartbeat-enabled
// agent, with no run yet) is enqueued exactly once for autonomous execution.
const enqueued = taskExecution.enqueueHeartbeatRuns();
@@ -166,11 +183,26 @@ describe("autonomous bootstrap run", () => {
test("a goal created after onboarding reaches the CEO and runs autonomously", async () => {
const connection = new DatabaseConnection(join(tmpdir(), `ac-goal-${randomUUID()}.db`));
const providers = new ProviderService(connection.db, providerAdapters, testVault, undefined, async () => ({
text: "Reviewed the goal and posted the operating plan.",
toolCalls: [],
rawResponseJson: JSON.stringify({ id: "resp_goal" }),
}));
// A minimal well-behaved agent: does its session, closes its task, then stops.
const providers = new ProviderService(connection.db, providerAdapters, testVault, undefined, async (input: ProviderToolTurnInput) => {
const firstUser = input.messages.find((message) => message.role === "user");
const taskId = firstUser && firstUser.role === "user" ? /Task id: (\S+)/.exec(firstUser.text)?.[1] : undefined;
const alreadyClosed = input.messages.some(
(message) => message.role === "tool" && message.results.some((result) => result.toolName === "update_task_status"),
);
if (taskId && !alreadyClosed) {
return {
text: "Reviewed the goal; closing the kickoff.",
toolCalls: [call("update_task_status", { task_id: taskId, status: "done" })],
rawResponseJson: JSON.stringify({ id: "resp_goal_close" }),
};
}
return {
text: "Reviewed the goal and posted the operating plan.",
toolCalls: [],
rawResponseJson: JSON.stringify({ id: "resp_goal" }),
};
});
const onboarding = new OnboardingService({ db: connection.db, saveProviderCredential: (input) => providers.saveCredential(input) });
const organization = new OrganizationService(connection.db);
const taskExecution = new TaskExecutionService(connection.db, providers);
+65
View File
@@ -128,6 +128,8 @@ describe("task execution service", () => {
expect(skillRuntimeSnapshot?.compiled_prompt).toContain("Verify release readiness");
expect(skillRuntimeSnapshot?.source_revision_ids).toContain(revisionId);
expect(capturedPrompt).toContain("Task: Draft launch checklist");
expect(capturedPrompt).toContain(`Task id: ${taskId}`);
expect(capturedPrompt).toContain(`update_task_status(task_id="${taskId}", status="done")`);
expect(capturedPrompt).toContain("Assignee: Ada (Execution lead)");
expect(capturedPrompt).toContain("Workspace: Task Workspace");
expect(capturedPrompt).toContain("Execution context:");
@@ -610,6 +612,69 @@ describe("task execution service", () => {
connection.close();
});
test("an open task with a completed run keeps getting sessions until done or the budget is spent", () => {
const connection = createConnection();
const organization = new OrganizationService(connection.db);
const providers = new ProviderService(connection.db, providerAdapters, testVault);
const execution = new TaskExecutionService(connection.db, providers);
let orgId = "";
let taskId = "";
connection.transaction(() => {
orgId = organization.createOrganization({
organizationName: "Acme Operations",
employeeName: "Dana Lee",
mission: "Coordinate agents with auditable approvals.",
goal: "Ship a verified control plane.",
});
const providerId = openAiProviderId(connection, orgId);
providers.saveCredential({ orgId, providerId, secret: "sk-live-provider", model: "gpt-5.4" });
const agentId = organization.createAgent({
orgId,
name: "Ada",
role: "Execution lead",
department: "Engineering",
modelProviderId: providerId,
modelId: "gpt-5.4",
budgetCents: 10000,
managerAgentId: null,
heartbeatEnabled: true,
});
taskId = organization.createTask({
orgId,
title: "Ship the tenant module",
description: "Multi-session engineering work.",
status: TASK_STATUS.BACKLOG,
assigneeAgentId: agentId,
risk: RISK_LEVEL.MEDIUM,
projectId: null,
goalId: null,
workspaceId: null,
priority: TASK_PRIORITY.HIGH,
requiresFilesystem: false,
});
});
const completeQueuedRuns = (): void => {
connection.db.prepare("update task_runs set state = 'completed', completed_at = datetime('now') where task_id = ? and state = 'queued'").run(taskId);
};
// Session 1, then 2, then 3: a run completing does not finish a task — done does.
expect(connection.transaction(() => execution.enqueueHeartbeatRuns())).toHaveLength(1);
completeQueuedRuns();
expect(connection.transaction(() => execution.enqueueHeartbeatRuns())).toHaveLength(1);
completeQueuedRuns();
expect(connection.transaction(() => execution.enqueueHeartbeatRuns())).toHaveLength(1);
completeQueuedRuns();
// Budget spent without done: the task parks instead of looping forever.
expect(connection.transaction(() => execution.enqueueHeartbeatRuns())).toEqual([]);
// A closed task never gets another session, budget or not.
connection.transaction(() => organization.updateTaskStatus(orgId, taskId, TASK_STATUS.DONE));
expect(connection.transaction(() => execution.enqueueHeartbeatRuns())).toEqual([]);
connection.close();
});
test("a transient provider failure parks the provider and the cooldown expiry resumes it", async () => {
const connection = createConnection();
const organization = new OrganizationService(connection.db);