From 7659adbbddd3f2002a3dd8e3fe383f14023554d9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=9B=B7=E7=94=B5=E8=8A=BD=E8=A1=A3?= Date: Thu, 8 Oct 2026 08:38:02 -0400 Subject: [PATCH] Q49: a task that spends its run budget shows as stalled; edits start a fresh budget Co-Authored-By: Claude Opus 5.5 --- .../migrations/067-task-run-budget.ts | 7 +++ src/main/database/schema.ts | 2 + src/main/services/heartbeat-scheduler.ts | 36 ++++++++------- src/main/services/organization-service.ts | 13 ++++-- src/main/services/task-execution-service.ts | 2 +- src/renderer/src/i18n/zh.ts | 1 + src/renderer/src/pages/WorkPage.tsx | 1 + .../src/pages/work/TaskDetailDialog.tsx | 1 + src/shared/domain.ts | 4 ++ tests/unit/domain-services.test.ts | 2 +- tests/unit/task-execution-service.test.ts | 45 +++++++++++++++++++ 11 files changed, 94 insertions(+), 20 deletions(-) create mode 100644 src/main/database/migrations/067-task-run-budget.ts diff --git a/src/main/database/migrations/067-task-run-budget.ts b/src/main/database/migrations/067-task-run-budget.ts new file mode 100644 index 00000000..a0f0e485 --- /dev/null +++ b/src/main/database/migrations/067-task-run-budget.ts @@ -0,0 +1,7 @@ +export const taskRunBudgetSchema = ` + alter table tasks add column stalled_reason text; + alter table tasks add column run_budget_floor integer not null default 0; + + insert into schema_migrations(version, name, applied_at) + values (67, 'task_run_budget', datetime('now')); +`; diff --git a/src/main/database/schema.ts b/src/main/database/schema.ts index 721493c7..17518a25 100644 --- a/src/main/database/schema.ts +++ b/src/main/database/schema.ts @@ -65,6 +65,7 @@ import { knowledgeScopeSchema } from "./migrations/063-knowledge-scope"; import { knowledgeRevisionsSchema } from "./migrations/064-knowledge-revisions"; import { documentVersionsSchema } from "./migrations/065-document-versions"; import { taskWritingModeSchema } from "./migrations/066-task-writing-mode"; +import { taskRunBudgetSchema } from "./migrations/067-task-run-budget"; interface Migration { version: number; @@ -139,6 +140,7 @@ const MIGRATIONS: Migration[] = [ { version: 64, name: "knowledge_revisions", sql: knowledgeRevisionsSchema }, { version: 65, name: "document_versions", sql: documentVersionsSchema }, { version: 66, name: "task_writing_mode", sql: taskWritingModeSchema }, + { version: 67, name: "task_run_budget", sql: taskRunBudgetSchema }, ]; const SCHEMA_VERSION = MIGRATIONS[MIGRATIONS.length - 1].version; diff --git a/src/main/services/heartbeat-scheduler.ts b/src/main/services/heartbeat-scheduler.ts index 9b410bd7..7a049fd5 100644 --- a/src/main/services/heartbeat-scheduler.ts +++ b/src/main/services/heartbeat-scheduler.ts @@ -3,6 +3,7 @@ import { currentProviderUsageWindow, isTransientProviderFailure } from "../../sh import { createId } from "../../shared/ids"; import type { ProviderCooldowns } from "./provider-cooldowns"; import { assertExecutionApproval, createIsolationContext, loadWorkspace, requireAgentProvider, requireAssignedAgent, requireTask } from "./task-run-preparation"; +import { AuditService } from "./audit-service"; const TASK_RETRY_BUDGET = 3; @@ -10,10 +11,13 @@ const TASK_RETRY_BUDGET = 3; export class HeartbeatScheduler { /** Providers currently parked on their monthly quota, for transition logging. */ private quotaBlockedProviders = 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, private readonly cooldowns: ProviderCooldowns, private readonly now: () => number) {} + constructor( + private readonly db: DatabaseSync, + private readonly cooldowns: ProviderCooldowns, + private readonly now: () => number, + private readonly notifyDataChanged: () => void, + ) {} /** * Heartbeat: keeps every open assigned task of a heartbeat-enabled agent @@ -57,7 +61,10 @@ 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 error_message from task_runs where task_id = ? and state <> 'superseded'"); + 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 <> 'superseded' and r.rowid > t.run_budget_floor + `); const created: string[] = []; const pacedAgents = new Set(); for (const row of rows) { @@ -66,7 +73,7 @@ export class HeartbeatScheduler { 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.observeSpentTask(row.task_id); + this.markStalled(row.org_id, row.task_id); continue; } try { @@ -79,16 +86,15 @@ export class HeartbeatScheduler { 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.`); + /** 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.`; + 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 }; + console.warn(`[task-queue] task "${task.title}" ${reason}`); + new AuditService(this.db).record({ orgId, entityType: "task", entityId: taskId, actorType: "system", actorId: null, action: "stalled", payload: { reason } }); + this.notifyDataChanged(); } /** diff --git a/src/main/services/organization-service.ts b/src/main/services/organization-service.ts index 01bf0e7f..c99108fd 100644 --- a/src/main/services/organization-service.ts +++ b/src/main/services/organization-service.ts @@ -494,7 +494,8 @@ export class OrganizationService { if (input.assigneeAgentId) this.verifyOptionalOrgEntity("agents", input.assigneeAgentId, input.orgId, "Assignee agent"); if (input.parentTaskId) this.assertParentAllowed(input.orgId, input.taskId, input.parentTaskId, input.title.trim()); const result = this.db.prepare(` - update tasks set title = ?, description = ?, assignee_agent_id = ?, risk = ?, priority = ?, parent_task_id = ?, updated_at = ? + update tasks set title = ?, description = ?, assignee_agent_id = ?, risk = ?, priority = ?, parent_task_id = ?, updated_at = ?, + stalled_reason = null, run_budget_floor = (select coalesce(max(rowid), 0) from task_runs) where id = ? and org_id = ? `).run(input.title.trim(), input.description.trim(), input.assigneeAgentId, input.risk, input.priority, input.parentTaskId, now, input.taskId, input.orgId); if (result.changes === 0) throw new Error("Task was not found."); @@ -528,8 +529,14 @@ export class OrganizationService { updateTaskStatus(orgId: string, taskId: string, status: string): void { if (status === TASK_STATUS.DONE) this.assertNoOpenSubtasks(orgId, taskId); - const result = this.db.prepare("update tasks set status = ?, updated_at = ? where id = ? and org_id = ?") - .run(status, new Date().toISOString(), taskId, orgId); + const now = new Date().toISOString(); + // Reopening a done task gives it a fresh run budget. + const result = this.db.prepare(` + update tasks set status = ?, updated_at = ?, + stalled_reason = case when status = 'done' and ? <> 'done' then null else stalled_reason end, + run_budget_floor = case when status = 'done' and ? <> 'done' then (select coalesce(max(rowid), 0) from task_runs) else run_budget_floor end + where id = ? and org_id = ? + `).run(status, now, status, status, taskId, orgId); if (result.changes === 0) throw new Error("Task was not found."); this.audit.record({ orgId, diff --git a/src/main/services/task-execution-service.ts b/src/main/services/task-execution-service.ts index 799197c6..308749e6 100644 --- a/src/main/services/task-execution-service.ts +++ b/src/main/services/task-execution-service.ts @@ -78,7 +78,7 @@ export class TaskExecutionService { this.organization = new OrganizationService(db); this.chat = new ChatService(db); this.cooldowns = new ProviderCooldowns(now); - this.heartbeat = new HeartbeatScheduler(db, this.cooldowns, now); + this.heartbeat = new HeartbeatScheduler(db, this.cooldowns, now, notifyDataChanged); } async runTask(input: RunTaskInput): Promise { diff --git a/src/renderer/src/i18n/zh.ts b/src/renderer/src/i18n/zh.ts index 41bbd1e9..11bcd170 100644 --- a/src/renderer/src/i18n/zh.ts +++ b/src/renderer/src/i18n/zh.ts @@ -30,6 +30,7 @@ export const zh: Record = { "Document": "文档", "Writing": "写作方式", "superseded": "已被取代", + "stalled": "已停滞", "Why it did not run": "未运行的原因", "Done tasks do not run": "已完成的任务不能再运行", "A session is already queued or running": "已有会话在排队或运行中", diff --git a/src/renderer/src/pages/WorkPage.tsx b/src/renderer/src/pages/WorkPage.tsx index d50d9233..05d61bf8 100644 --- a/src/renderer/src/pages/WorkPage.tsx +++ b/src/renderer/src/pages/WorkPage.tsx @@ -337,6 +337,7 @@ function TaskCard({
{project ? {project.name} : {capitalize(task.risk)}} + {task.stalledReason && task.status !== "done" && {t("stalled")}} {task.status !== "done" && latestRun && ( {prettyRunState(latestRun.state)} diff --git a/src/renderer/src/pages/work/TaskDetailDialog.tsx b/src/renderer/src/pages/work/TaskDetailDialog.tsx index 7a2c4fb8..4f9b28c3 100644 --- a/src/renderer/src/pages/work/TaskDetailDialog.tsx +++ b/src/renderer/src/pages/work/TaskDetailDialog.tsx @@ -142,6 +142,7 @@ function OverviewTab({ task, assignee, project, goal, workspace, provider, runs }) { return ( <> + {task.stalledReason &&
{task.stalledReason}
} {task.description && (

{task.description}

)} diff --git a/src/shared/domain.ts b/src/shared/domain.ts index e344a9f2..3403811a 100644 --- a/src/shared/domain.ts +++ b/src/shared/domain.ts @@ -98,6 +98,10 @@ export interface TaskRecord { parentTaskId: string | null; requiresFilesystem: boolean; writingMode: WritingMode; + /** Set when the task spent its run budget without reaching done. */ + stalledReason: string | null; + /** Runs at or below this task_runs rowid no longer count toward the run budget. */ + runBudgetFloor: number; createdAt: string; updatedAt: string; } diff --git a/tests/unit/domain-services.test.ts b/tests/unit/domain-services.test.ts index 0aea20cc..d3f14cd0 100644 --- a/tests/unit/domain-services.test.ts +++ b/tests/unit/domain-services.test.ts @@ -24,7 +24,7 @@ describe("domain services", () => { const connection = createConnection(); const rows = connection.db.prepare("select version from schema_migrations order by version asc").all() as Array<{ version: number }>; - expect(rows.map((row) => row.version)).toEqual([1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42, 43, 44, 45, 46, 47, 48, 49, 50, 51, 52, 53, 54, 55, 56, 57, 58, 59, 60, 61, 62, 63, 64, 65, 66]); + expect(rows.map((row) => row.version)).toEqual([1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42, 43, 44, 45, 46, 47, 48, 49, 50, 51, 52, 53, 54, 55, 56, 57, 58, 59, 60, 61, 62, 63, 64, 65, 66, 67]); connection.close(); }); diff --git a/tests/unit/task-execution-service.test.ts b/tests/unit/task-execution-service.test.ts index 347094e2..c0f35cfc 100644 --- a/tests/unit/task-execution-service.test.ts +++ b/tests/unit/task-execution-service.test.ts @@ -698,6 +698,51 @@ describe("task execution service", () => { connection.close(); }); + test("a task that spends its run budget is marked stalled with a reason; editing or reassigning it starts a fresh budget", () => { + const connection = createConnection(); + const organization = new OrganizationService(connection.db); + const providers = new ProviderService(connection.db, providerAdapters, testVault); + let notified = 0; + const execution = new TaskExecutionService(connection.db, providers, undefined, () => { notified += 1; }); + let orgId = ""; + let taskId = ""; + let agentId = ""; + 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" }); + 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: "Ship it", 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 spendRun = () => { + connection.transaction(() => execution.enqueueHeartbeatRuns()); + connection.db.prepare("update task_runs set state = 'completed' where task_id = ? and state = 'queued'").run(taskId); + }; + 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(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); + + organization.updateTask({ orgId, taskId, title: "Ship it", description: "Narrower scope.", assigneeAgentId: agentId, risk: RISK_LEVEL.LOW, priority: TASK_PRIORITY.NORMAL, parentTaskId: null }); + expect(connection.db.prepare("select stalled_reason from tasks where id = ?").get(taskId)).toEqual({ stalled_reason: null }); + expect(connection.transaction(() => execution.enqueueHeartbeatRuns())).toHaveLength(1); + + connection.db.prepare("update task_runs set state = 'completed' where task_id = ?").run(taskId); + connection.db.prepare("update tasks set stalled_reason = 'Stalled: x', status = 'done' where id = ?").run(taskId); + organization.updateTaskStatus(orgId, taskId, TASK_STATUS.IN_PROGRESS); + const reopened = connection.db.prepare("select stalled_reason, run_budget_floor = (select max(rowid) from task_runs) as fresh from tasks where id = ?").get(taskId); + expect(reopened).toEqual({ stalled_reason: null, fresh: 1 }); + 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);