diff --git a/src/main/services/task-execution-service.ts b/src/main/services/task-execution-service.ts index 4a0de7bf..94dca5e8 100644 --- a/src/main/services/task-execution-service.ts +++ b/src/main/services/task-execution-service.ts @@ -67,6 +67,8 @@ export class TaskExecutionService { private quotaBlockedProviders = new Map(); /** Providers parked after a transient failure (usage/rate limit, overload): id -> park state. */ private providerCooldowns = new Map(); + /** Tasks already logged as budget-spent, so the park is announced once, not every 5s tick. */ + private spentTasksLogged = new Set(); 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); diff --git a/src/main/services/task-prompts.ts b/src/main/services/task-prompts.ts index ade548df..72735fb6 100644 --- a/src/main/services/task-prompts.ts +++ b/src/main/services/task-prompts.ts @@ -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"); diff --git a/tests/integration/autonomous-run.test.ts b/tests/integration/autonomous-run.test.ts index 3e829aff..6f43d004 100644 --- a/tests/integration/autonomous-run.test.ts +++ b/tests/integration/autonomous-run.test.ts @@ -22,7 +22,7 @@ function call(name: string, args: Record): 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 { const script: Array<{ text: string; toolCalls: ToolCall[] }> = [ { text: "Staffing.", toolCalls: [ @@ -36,6 +36,7 @@ function scriptedCeo(): (input: ProviderToolTurnInput) => Promise Promise 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); diff --git a/tests/unit/task-execution-service.test.ts b/tests/unit/task-execution-service.test.ts index 6b09f192..b0584bef 100644 --- a/tests/unit/task-execution-service.test.ts +++ b/tests/unit/task-execution-service.test.ts @@ -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);