diff --git a/src/main/services/agent-run-loop.ts b/src/main/services/agent-run-loop.ts index 4fa5ec2f..d378201a 100644 --- a/src/main/services/agent-run-loop.ts +++ b/src/main/services/agent-run-loop.ts @@ -11,6 +11,7 @@ import { fitContextBudget } from "./context-budget"; import { throwIfRunStopped, untilStopped } from "./run-control"; const MAX_ITERATIONS = 12; +const STEP_LIMIT_TEXT = "I used this session's steps and will continue in the next one."; export interface AgentRunLoopInput { orgId: string; @@ -35,6 +36,8 @@ export interface AgentRunLoopResult { iterations: number; toolCallCount: number; lastRequestId: string; + /** The session was still working when it reached the step limit; the task continues next session. */ + hitStepLimit: boolean; } export interface AgentRunLoopDeps { @@ -60,6 +63,7 @@ export async function runAgentLoop(deps: AgentRunLoopDeps, input: AgentRunLoopIn const messages: AgentTurnMessage[] = [{ role: "user", text: input.initialUserTurn }]; let toolCallCount = 0; let lastRequestId = ""; + let lastText = ""; for (let iteration = 1; iteration <= MAX_ITERATIONS; iteration += 1) { const compaction = compactToolHistory(messages); @@ -96,9 +100,10 @@ export async function runAgentLoop(deps: AgentRunLoopDeps, input: AgentRunLoopIn messages.push({ role: "assistant", text: turn.text, toolCalls: [] }, { role: "user", text: lateMessages }); continue; } - return { finalText: turn.text, iterations: iteration, toolCallCount, lastRequestId }; + return { finalText: turn.text, iterations: iteration, toolCallCount, lastRequestId, hitStepLimit: false }; } + lastText = turn.text.trim() || lastText; messages.push({ role: "assistant", text: turn.text, toolCalls: turn.toolCalls }); const results: Array<{ toolCallId: string; toolName: string; output: string }> = []; for (const call of turn.toolCalls) { @@ -136,7 +141,9 @@ export async function runAgentLoop(deps: AgentRunLoopDeps, input: AgentRunLoopIn deps.onToolResults?.(); } - throw new Error(`Agent run did not converge within ${MAX_ITERATIONS} iterations.`); + // Still working at the limit: end as a checkpoint; the next session picks up from the run journal. + console.info(`[run-loop] reached the ${MAX_ITERATIONS}-step limit with work in progress; ending the session as a checkpoint`); + return { finalText: lastText || STEP_LIMIT_TEXT, iterations: MAX_ITERATIONS, toolCallCount, lastRequestId, hitStepLimit: true }; } function shouldAnnounceTaskAssignment(call: { name: string; arguments: Record }, toolCalls: Array<{ name: string; arguments: Record }>): boolean { diff --git a/src/main/services/heartbeat-scheduler.ts b/src/main/services/heartbeat-scheduler.ts index 5574ff24..35c218f4 100644 --- a/src/main/services/heartbeat-scheduler.ts +++ b/src/main/services/heartbeat-scheduler.ts @@ -5,7 +5,10 @@ import type { ProviderCooldowns } from "./provider-cooldowns"; import { assertExecutionApproval, createIsolationContext, loadWorkspace, requireAgentProvider, requireAssignedAgent, requireTask } from "./task-run-preparation"; import { AuditService } from "./audit-service"; +/** Sessions that made no progress before a task stalls. */ const TASK_RETRY_BUDGET = 3; +/** Sessions of any kind before a task stalls, so endless busywork still stops. */ +const TASK_SESSION_CEILING = 12; /** Enqueues follow-up runs for open tasks each worker tick. */ export class HeartbeatScheduler { @@ -19,22 +22,7 @@ export class HeartbeatScheduler { private readonly notifyDataChanged: () => void, ) {} - /** - * 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 - * certain failure), tasks on a provider inside a transient-failure cooldown - * wait it out, and transient failures (usage/rate limits, overload) never - * count toward the retry budget. Raising a limit — or the provider's own - * window resetting — therefore resumes work without a restart. - */ + /** Queues the next session of open work; stalls a task after sessions without progress or too many in all; transient provider failures never count. */ enqueueHeartbeatRuns(): string[] { const window = currentProviderUsageWindow(); const nowIso = new Date(this.now()).toISOString(); @@ -62,19 +50,26 @@ export class HeartbeatScheduler { `).all(window.start, nowIso) as Array<{ task_id: string; org_id: string; provider_id: string; agent_id: string; interval_sec: number }>; this.observeQuotaBlocks(window.start); const cooling = new Set(this.cooldowns.cooledProviderIds()); - const attemptMessages = this.db.prepare(` - select r.error_message from task_runs r join tasks t on t.id = r.task_id - where r.task_id = ? and r.state not in ('superseded', 'stopped') and r.rowid > t.run_budget_floor + const sessionsOf = this.db.prepare(` + select r.error_message, + exists (select 1 from run_actions a where a.run_id = r.id and a.result_excerpt not like 'ERROR%') as progressed + from task_runs r join tasks t on t.id = r.task_id + where r.task_id = ? and r.state not in ('superseded', 'stopped', 'queued', 'running') and r.rowid > t.run_budget_floor `); const created: string[] = []; const pacedAgents = new Set(); for (const row of rows) { if (cooling.has(row.provider_id)) continue; if (row.interval_sec > 0 && pacedAgents.has(row.agent_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) { - this.markStalled(row.org_id, row.task_id); + const sessions = (sessionsOf.all(row.task_id) as Array<{ error_message: string | null; progressed: number }>) + .filter((run) => !isTransientProviderFailure(run.error_message)); + const stuck = sessions.filter((run) => run.progressed === 0).length; + if (stuck >= TASK_RETRY_BUDGET) { + this.markStalled(row.org_id, row.task_id, `${TASK_RETRY_BUDGET} sessions made no progress`); + continue; + } + if (sessions.length >= TASK_SESSION_CEILING) { + this.markStalled(row.org_id, row.task_id, `${TASK_SESSION_CEILING} sessions ran without reaching done`); continue; } try { @@ -88,8 +83,8 @@ export class HeartbeatScheduler { } /** Records once, on the task itself, that it spent its run budget without reaching done. */ - private markStalled(orgId: string, taskId: string): void { - const reason = `Stalled: ${TASK_RETRY_BUDGET} sessions ran without reaching done. Edit, reassign or reopen the task to give it a fresh budget.`; + private markStalled(orgId: string, taskId: string, why: string): void { + const reason = `Stalled: ${why}. Edit, reassign or reopen the task to give it a fresh budget.`; const marked = this.db.prepare("update tasks set stalled_reason = ? where id = ? and stalled_reason is null").run(reason, taskId); if (marked.changes === 0) return; const task = this.db.prepare("select title from tasks where id = ?").get(taskId) as { title: string }; diff --git a/src/main/services/task-execution-service.ts b/src/main/services/task-execution-service.ts index 0d9497d8..bb79b0b0 100644 --- a/src/main/services/task-execution-service.ts +++ b/src/main/services/task-execution-service.ts @@ -246,7 +246,8 @@ export class TaskExecutionService { // in plain text instead of calling send_message, deliver that text into // the source thread BEFORE the run may complete — a delivery failure // fails the run rather than closing it on silence. - if (chatContext) { + // A checkpoint is unfinished work: no stand-in reply, and a chat wake stays open for its next session. + if (chatContext && !result.hitStepLimit) { const delivered = deliverChatReplyIfMissing(this.db, this.chat, { orgId, agentId: agent.id, @@ -257,7 +258,7 @@ export class TaskExecutionService { }); if (delivered) this.notifyDataChanged(); } - this.completeRun(orgId, runId, result.lastRequestId); + this.completeRun(orgId, runId, result.lastRequestId, result.hitStepLimit); this.cooldowns.clear(provider); if (task.origin === null) recordRunMemory(this.db, { orgId, runId, projectId: task.project_id, taskTitle: task.title, outcome: "completed", text: result.finalText }); this.audit.record({ @@ -274,6 +275,7 @@ export class TaskExecutionService { providerRequestId: result.lastRequestId, iterations: result.iterations, toolCalls: result.toolCallCount, + hitStepLimit: result.hitStepLimit, executionContextId: isolation.executionContextId, serviceIdentityId: isolation.serviceIdentityId, skillRuntimeSnapshotId: skillRuntimeSnapshot?.id ?? null, @@ -398,13 +400,13 @@ export class TaskExecutionService { return row?.language === "zh" ? "zh" : "en"; } - private completeRun(orgId: string, runId: string, providerRequestId: string): void { + private completeRun(orgId: string, runId: string, providerRequestId: string, checkpoint: boolean): void { this.db.prepare(` update task_runs set state = ?, provider_request_id = ?, completed_at = ?, error_message = null where id = ? and org_id = ? `).run(TASK_RUN_STATE.COMPLETED, providerRequestId, new Date().toISOString(), runId, orgId); - this.closeChatWakeTask(orgId, runId); + if (!checkpoint) this.closeChatWakeTask(orgId, runId); } /** @@ -414,9 +416,9 @@ export class TaskExecutionService { * enqueued again) and pile zombie "Reply to ..." rows onto the board. */ private closeChatWakeTask(orgId: string, runId: string): void { - const run = this.db.prepare("select task_id, trigger from task_runs where id = ? and org_id = ?") - .get(runId, orgId) as { task_id: string; trigger: string } | undefined; - if (!run || run.trigger !== "chat_mention") return; + const run = this.db.prepare("select r.task_id, t.origin from task_runs r join tasks t on t.id = r.task_id where r.id = ? and r.org_id = ?") + .get(runId, orgId) as { task_id: string; origin: string | null } | undefined; + if (!run || run.origin !== "chat_wake") return; const task = this.db.prepare("select status from tasks where id = ? and org_id = ?") .get(run.task_id, orgId) as { status: string } | undefined; if (!task || (task.status !== TASK_STATUS.BACKLOG && task.status !== TASK_STATUS.IN_PROGRESS)) return; diff --git a/tests/integration/run-checkpoint.test.ts b/tests/integration/run-checkpoint.test.ts new file mode 100644 index 00000000..1191fabd --- /dev/null +++ b/tests/integration/run-checkpoint.test.ts @@ -0,0 +1,67 @@ +import { randomUUID } from "node:crypto"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { describe, expect, test } from "vitest"; +import { DatabaseConnection } from "../../src/main/database/connection"; +import { OnboardingService } from "../../src/main/services/onboarding-service"; +import { ProviderService } from "../../src/main/services/provider-service"; +import { providerAdapters } from "../../src/main/services/provider-adapters"; +import { TaskExecutionService } from "../../src/main/services/task-execution-service"; +import { ChatService } from "../../src/main/services/chat-service"; +import { ensureCeoDirectSpace } from "../../src/main/services/ceo-direct-space"; +import { claimNextQueuedRun } from "../../src/main/app/task-queue-worker"; +import type { AgentTurnResult } from "../../src/main/services/provider-tool-turn"; + +const testVault = { + encryptSecret: (secret: string) => Buffer.from(secret, "utf8"), + decryptSecret: (ciphertext: Buffer) => ciphertext.toString("utf8"), +}; + +describe("session checkpoints", () => { + test("a chat reply still working at the step limit stays open and posts no stand-in reply", async () => { + let finishing = false; + const turn = async (): Promise => finishing + ? { text: "Audit done: all clean.", toolCalls: [], rawResponseJson: "{}" } + : { text: "Looking into it.", toolCalls: [{ id: `c_${randomUUID()}`, name: "read_messages", arguments: {} }], rawResponseJson: "{}" }; + const connection = new DatabaseConnection(join(tmpdir(), `ac-checkpoint-${randomUUID()}.db`)); + const providers = new ProviderService(connection.db, providerAdapters, testVault, undefined, turn); + const onboarding = new OnboardingService({ db: connection.db, saveProviderCredential: (input) => providers.saveCredential(input) }); + const execution = new TaskExecutionService(connection.db, providers); + let orgId = ""; + let ceoId = ""; + connection.transaction(() => { + const summary = onboarding.bootstrap({ + companyName: "FlowPilot", companyDescription: "CSM analytics.", employeeName: "Yang", workspacePath: "", + executionMode: "ai_ceo", autoApproveHires: true, reviewDeliverables: false, + goalTitle: "Ship the MVP.", goalDescription: "A working product.", + providerKey: "openai", providerSecret: "sk-test-key", providerModel: "gpt-5.4", + }); + orgId = summary.orgId; + ceoId = summary.ceoAgentId; + }); + connection.db.prepare("update task_runs set state = 'completed' where state = 'queued'").run(); + const founder = (connection.db.prepare("select id from employees where org_id = ?").get(orgId) as { id: string }).id; + new ChatService(connection.db).sendMessage({ + orgId, spaceId: ensureCeoDirectSpace(connection.db, orgId)!, threadId: null, + senderType: "employee", senderId: founder, bodyMd: "Audit every repo, please.", attachmentFileIds: [], + }); + const wake = claimNextQueuedRun(connection)!; + await execution.executeRun(orgId, wake.id); + const run = connection.db.prepare("select r.state, t.status from task_runs r join tasks t on t.id = r.task_id where r.id = ?").get(wake.id); + expect(run).toEqual({ state: "completed", status: "backlog" }); + expect(connection.db.prepare("select count(*) as n from chat_messages where sender_type = 'agent' and sender_id = ?").get(ceoId)).toEqual({ n: 0 }); + + // The heartbeat continuation answers and closes the conversation's task. + finishing = true; + const wakeTask = (connection.db.prepare("select task_id from task_runs where id = ?").get(wake.id) as { task_id: string }).task_id; + connection.db.prepare("update task_runs set created_at = '2000-01-01T00:00:00.000Z' where agent_id = ?").run(ceoId); + connection.db.prepare("update tasks set status = 'done' where origin is null").run(); + const queued = connection.transaction(() => execution.enqueueHeartbeatRuns()); + const continuation = queued.find((id) => (connection.db.prepare("select task_id from task_runs where id = ?").get(id) as { task_id: string }).task_id === wakeTask)!; + connection.db.prepare("update task_runs set state = 'running' where id = ?").run(continuation); + await execution.executeRun(orgId, continuation); + expect(connection.db.prepare("select status from tasks where id = ?").get(wakeTask)).toEqual({ status: "done" }); + expect(connection.db.prepare("select count(*) as n from chat_messages where sender_type = 'agent' and sender_id = ? and body_md = 'Audit done: all clean.'").get(ceoId)).toEqual({ n: 1 }); + connection.close(); + }); +}); diff --git a/tests/unit/agent-run-loop.test.ts b/tests/unit/agent-run-loop.test.ts index d25862a7..b124c62e 100644 --- a/tests/unit/agent-run-loop.test.ts +++ b/tests/unit/agent-run-loop.test.ts @@ -234,4 +234,28 @@ describe("agent run loop (real tool dispatch)", () => { )).rejects.toThrow(/empty turn/); connection.close(); }); + + test("a session still working at the step limit ends as a checkpoint instead of failing", async () => { + const connection = new DatabaseConnection(join(tmpdir(), `ac-loop4-${randomUUID()}.db`)); + const organization = new OrganizationService(connection.db); + let orgId = ""; + let ceoId = ""; + let providerId = ""; + connection.transaction(() => { + orgId = organization.createOrganization({ organizationName: "OrbitOps", employeeName: "Yang", mission: "m", goal: "g" }); + providerId = (connection.db.prepare("select id from providers where org_id = ? and provider_key = 'openai'").get(orgId) as { id: string }).id; + ceoId = organization.createAgent({ orgId, name: "CEO", role: "CEO", department: "Executive", modelProviderId: providerId, modelId: "gpt-5.4", budgetCents: 25000, managerAgentId: null }); + }); + const ctx: AgentToolContext = { + orgId, agent: { id: ceoId, modelProviderId: providerId, modelId: "gpt-5.4" }, + taskProjectId: null, taskGoalId: null, taskId: null, sourceThreadId: null, workspace: null, + }; + const busy = Array.from({ length: 40 }, (_, index) => ({ text: index === 11 ? "Still building." : "", toolCalls: [call("read_messages", {})] })); + const result = await runAgentLoop( + { providers: fakeProvider(busy, connection.db), tools: { db: connection.db, organization, chat: new ChatService(connection.db) } }, + { orgId, providerId, inputBudgetTokens: 112_000, workspaceId: null, instructions: "i", initialUserTurn: "u", toolContext: ctx }, + ); + expect(result).toMatchObject({ iterations: 12, toolCallCount: 12, hitStepLimit: true, finalText: "Still building." }); + connection.close(); + }); }); diff --git a/tests/unit/task-execution-service.test.ts b/tests/unit/task-execution-service.test.ts index 7b216cab..09ff8671 100644 --- a/tests/unit/task-execution-service.test.ts +++ b/tests/unit/task-execution-service.test.ts @@ -727,7 +727,7 @@ describe("task execution service", () => { spendRun(); spendRun(); spendRun(); expect(connection.transaction(() => execution.enqueueHeartbeatRuns())).toEqual([]); const stalled = connection.db.prepare("select status, stalled_reason from tasks where id = ?").get(taskId) as { status: string; stalled_reason: string }; - expect(stalled).toEqual({ status: "in_progress", stalled_reason: "Stalled: 3 sessions ran without reaching done. Edit, reassign or reopen the task to give it a fresh budget." }); + expect(stalled).toEqual({ status: "in_progress", stalled_reason: "Stalled: 3 sessions made no progress. Edit, reassign or reopen the task to give it a fresh budget." }); expect(connection.db.prepare("select count(*) as n from audit_events where entity_id = ? and action = 'stalled'").get(taskId)).toEqual({ n: 1 }); expect(notified).toBe(1); @@ -743,6 +743,42 @@ describe("task execution service", () => { connection.close(); }); + test("sessions that make progress do not spend the run budget, up to a ceiling of sessions", () => { + 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", employeeName: "Dana Lee", mission: "m", goal: "g" }); + 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: "Engineer", department: "Engineering", modelProviderId: providerId, modelId: "gpt-5.4", + budgetCents: 10000, managerAgentId: null, heartbeatEnabled: true, + }); + taskId = organization.createTask({ + orgId, title: "Big build", description: "d", status: TASK_STATUS.IN_PROGRESS, assigneeAgentId: agentId, + risk: RISK_LEVEL.LOW, projectId: null, goalId: null, workspaceId: null, priority: TASK_PRIORITY.NORMAL, requiresFilesystem: false, + }); + }); + const productiveSession = (): boolean => { + const [runId] = connection.transaction(() => execution.enqueueHeartbeatRuns()); + if (!runId) return false; + connection.db.prepare("update task_runs set state = 'completed' where id = ?").run(runId); + connection.db.prepare("insert into run_actions(id, org_id, run_id, task_id, tool_name, arguments_excerpt, result_excerpt, created_at) values (?, ?, ?, ?, 'write_deliverable', '{}', 'Wrote src/app.ts', ?)") + .run(`rac_${runId}`, orgId, runId, taskId, new Date().toISOString()); + return true; + }; + let sessions = 0; + while (productiveSession()) sessions += 1; + expect(sessions).toBe(12); + expect(connection.db.prepare("select stalled_reason from tasks where id = ?").get(taskId)) + .toEqual({ stalled_reason: "Stalled: 12 sessions ran without reaching done. Edit, reassign or reopen the task to give it a fresh budget." }); + 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);