diff --git a/src/main/database/migrations/093-task-enqueue-error.ts b/src/main/database/migrations/093-task-enqueue-error.ts new file mode 100644 index 00000000..fe47ee61 --- /dev/null +++ b/src/main/database/migrations/093-task-enqueue-error.ts @@ -0,0 +1,7 @@ +// Why the heartbeat last failed to queue an open task; cleared when it next queues it. +export const taskEnqueueErrorSchema = ` + alter table tasks add column enqueue_error text; + + insert into schema_migrations(version, name, applied_at) + values (93, 'task_enqueue_error', datetime('now')); +`; diff --git a/src/main/database/schema.ts b/src/main/database/schema.ts index ed80c84d..3e771e49 100644 --- a/src/main/database/schema.ts +++ b/src/main/database/schema.ts @@ -91,6 +91,7 @@ import { fileVersionsSchema } from "./migrations/089-file-versions"; import { erasureReceiptsSchema } from "./migrations/090-erasure-receipts"; import { requestPurposeSchema } from "./migrations/091-request-purpose"; import { runActionTargetsSchema } from "./migrations/092-run-action-targets"; +import { taskEnqueueErrorSchema } from "./migrations/093-task-enqueue-error"; interface Migration { version: number; @@ -191,6 +192,7 @@ const MIGRATIONS: Migration[] = [ { version: 90, name: "erasure_receipts", sql: erasureReceiptsSchema }, { version: 91, name: "request_purpose", sql: requestPurposeSchema }, { version: 92, name: "run_action_targets", sql: runActionTargetsSchema }, + { version: 93, name: "task_enqueue_error", sql: taskEnqueueErrorSchema }, ]; 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 5e1b6ded..90fd45bf 100644 --- a/src/main/services/heartbeat-scheduler.ts +++ b/src/main/services/heartbeat-scheduler.ts @@ -85,13 +85,23 @@ export class HeartbeatScheduler { try { created.push(this.enqueueRun(row.org_id, row.task_id)); pacedAgents.add(row.agent_id); - } catch { - // Skip tasks that cannot be prepared yet (e.g. provider not fully configured). + this.db.prepare("update tasks set enqueue_error = null where id = ? and enqueue_error is not null").run(row.task_id); + } catch (error) { + // A task that cannot be prepared yet (approval pending, setup missing) stays open; its task shows why until it queues. + this.recordEnqueueError(row.task_id, error instanceof Error ? error.message : String(error)); } } return created; } + /** Records on the task why it could not be queued, logging and notifying only when the reason changes. */ + private recordEnqueueError(taskId: string, message: string): void { + const changed = this.db.prepare("update tasks set enqueue_error = ? where id = ? and enqueue_error is not ?").run(message, taskId, message); + if (changed.changes === 0) return; + console.warn(`[task-queue] task ${taskId} could not be queued: ${message}`); + this.notifyDataChanged(); + } + /** Records once, on the task itself, that it spent its run budget without reaching done. */ 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.`; diff --git a/src/main/services/snapshot-service.ts b/src/main/services/snapshot-service.ts index d95aecec..b54a8474 100644 --- a/src/main/services/snapshot-service.ts +++ b/src/main/services/snapshot-service.ts @@ -4,6 +4,7 @@ import { PROVIDER_CATALOG } from "../../shared/provider-catalog"; import { IntegrationKeyStore } from "./integration-key-store"; import { currentProviderUsageWindow } from "../../shared/provider-runtime-domain"; import { buildUsageStats } from "./usage-stats"; +import { taskHolds } from "./task-holds"; type Row = Record; type ProviderUsageWindowRow = Pick< @@ -49,8 +50,9 @@ export class SnapshotService { const activeOrgId = this.resolveActiveOrgId(organizations.map((org) => org.id)); const agents = list(this.db, "agents", "created_at asc") .map((agent) => ({ ...agent, heartbeatEnabled: toBool(agent.heartbeatEnabled) })); + const holds = taskHolds(this.db, new Date()); const tasks = list(this.db, "tasks", "created_at asc") - .map((task) => ({ ...task, requiresFilesystem: toBool(task.requiresFilesystem) })); + .map((task) => ({ ...task, requiresFilesystem: toBool(task.requiresFilesystem), hold: holds.get(task.id) ?? null })); const workspaces = list(this.db, "workspaces", "created_at asc") .map((workspace) => ({ ...workspace, isPrimary: toBool(workspace.isPrimary) })); const providers = list(this.db, "providers", "label asc") diff --git a/src/main/services/spend.ts b/src/main/services/spend.ts index 7ce2ca97..64701779 100644 --- a/src/main/services/spend.ts +++ b/src/main/services/spend.ts @@ -69,7 +69,8 @@ export function budgetBlock(db: DatabaseSync, agentId: string, paid: boolean, no const agentSpend = (db.prepare("select coalesce(sum(cost_cents), 0) as cents from provider_requests where agent_id = ? and created_at >= ?") .get(agentId, since) as { cents: number }).cents; if (agentSpend >= agent.budget_cents) { - return `${agent.name} has spent its monthly budget (${dollars(agentSpend)} of ${dollars(agent.budget_cents)}). Raise the budget or wait for next month.`; + const next = agent.budget_cents === 0 ? "Raise the budget to let it run." : "Raise the budget or wait for next month."; + return `${agent.name} has spent its monthly budget (${dollars(agentSpend)} of ${dollars(agent.budget_cents)}). ${next}`; } const department = db.prepare("select monthly_budget_limit_cents from org_departments where org_id = ? and name = ?") .get(agent.org_id, agent.department) as { monthly_budget_limit_cents: number } | undefined; diff --git a/src/main/services/task-holds.ts b/src/main/services/task-holds.ts new file mode 100644 index 00000000..07662c19 --- /dev/null +++ b/src/main/services/task-holds.ts @@ -0,0 +1,76 @@ +import type { DatabaseSync } from "node:sqlite"; +import type { TaskHold } from "../../shared/domain"; +import { currentProviderUsageWindow } from "../../shared/provider-runtime-domain"; +import { budgetBlock, isPriced, providerPrices } from "./spend"; + +interface OpenTaskRow { + id: string; + stalled_reason: string | null; + enqueue_error: string | null; + agent_id: string | null; + provider_id: string | null; + provider_label: string | null; + request_limit: number; + request_count: number; + cooldown_until_ms: number | null; + cooldown_label: string | null; +} + +/** Why each open task without a run in flight is not moving, when it resumes on its own, and who must act otherwise. */ +export function taskHolds(db: DatabaseSync, now: Date): Map { + const window = currentProviderUsageWindow(now); + const rows = db.prepare(` + select t.id, t.stalled_reason, t.enqueue_error, a.id as agent_id, p.id as provider_id, p.label as provider_label, + p.monthly_request_limit as request_limit, coalesce(u.request_count, 0) as request_count, + c.until_ms as cooldown_until_ms, c.label as cooldown_label + from tasks t + left join agents a on a.id = t.assignee_agent_id + left join providers p on p.id = a.model_provider_id + left join provider_usage_windows u on u.org_id = p.org_id and u.provider_id = p.id and u.window_start = ? + left join provider_cooldowns c on c.provider_id = p.id + where t.status in ('backlog', 'in_progress') + and not exists (select 1 from task_runs r where r.task_id = t.id and r.state = 'running') + `).all(window.start) as unknown as OpenTaskRow[]; + const budgets = new Map(); + // A zero cap stays spent after the monthly reset, so only raising it lets the work resume. + const zeroCap = db.prepare(` + select a.budget_cents = 0 or coalesce(d.monthly_budget_limit_cents, 1) = 0 as zero + from agents a left join org_departments d on d.org_id = a.org_id and d.name = a.department where a.id = ? + `); + const holds = new Map(); + for (const row of rows) { + const hold = holdOf(row, now, window.end, (agentId, providerId) => { + let block = budgets.get(agentId); + if (block === undefined) { + const reason = budgetBlock(db, agentId, isPriced(providerPrices(db, providerId)), now); + block = reason ? { reason, renews: (zeroCap.get(agentId) as { zero: number }).zero === 0 } : null; + budgets.set(agentId, block); + } + return block; + }); + if (hold) holds.set(row.id, hold); + } + return holds; +} + +interface MoneyBlock { + reason: string; + /** False when a zero cap means the next month starts spent too. */ + renews: boolean; +} + +function holdOf(row: OpenTaskRow, now: Date, windowEnd: string, budget: (agentId: string, providerId: string) => MoneyBlock | null): TaskHold | null { + if (row.stalled_reason) return { reason: row.stalled_reason, resumesAt: null, waitingOn: "operator" }; + if (row.enqueue_error) return { reason: `Could not start: ${row.enqueue_error}`, resumesAt: null, waitingOn: "operator" }; + if (!row.agent_id || !row.provider_id) return null; + if (row.request_count >= row.request_limit) { + if (row.request_limit === 0) return { reason: `${row.provider_label} has a monthly request limit of 0. Raise the limit to let its work run.`, resumesAt: null, waitingOn: "operator" }; + return { reason: `${row.provider_label} used its monthly request limit (${row.request_limit}). Raise the limit or wait for the next window.`, resumesAt: windowEnd, waitingOn: "operator" }; + } + const spent = budget(row.agent_id, row.provider_id); + if (spent) return { reason: spent.reason, resumesAt: spent.renews ? windowEnd : null, waitingOn: "operator" }; + if (row.cooldown_until_ms !== null && row.cooldown_until_ms > now.getTime()) { + return { reason: `${row.cooldown_label} reported a passing failure; its work waits out a cooldown.`, resumesAt: new Date(row.cooldown_until_ms).toISOString(), waitingOn: "provider" }; + } + return null; +} diff --git a/src/main/services/task-mutations.ts b/src/main/services/task-mutations.ts index 26121d80..c0a622e1 100644 --- a/src/main/services/task-mutations.ts +++ b/src/main/services/task-mutations.ts @@ -50,7 +50,7 @@ export class TaskMutations { } const result = this.db.prepare(` 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) + stalled_reason = null, enqueue_error = 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."); diff --git a/src/renderer/src/i18n/zh.ts b/src/renderer/src/i18n/zh.ts index ed352a22..c33877e8 100644 --- a/src/renderer/src/i18n/zh.ts +++ b/src/renderer/src/i18n/zh.ts @@ -31,6 +31,11 @@ export const zh: Record = { "Writing": "写作方式", "superseded": "已被取代", "stalled": "已停滞", + "needs you": "待你处理", + "waiting": "等待中", + "Resumes on its own at {time}.": "将于 {time} 自动恢复。", + "Waiting on you, or resumes on its own at {time}.": "等待你处理，或于 {time} 自动恢复。", + "Waiting on you.": "等待你处理。", "Resume": "恢复", "Missing this run": "本次运行缺少的能力", "Disconnect": "断开连接", diff --git a/src/renderer/src/pages/WorkPage.tsx b/src/renderer/src/pages/WorkPage.tsx index 05d61bf8..96b24070 100644 --- a/src/renderer/src/pages/WorkPage.tsx +++ b/src/renderer/src/pages/WorkPage.tsx @@ -338,6 +338,11 @@ function TaskCard({
{project ? {project.name} : {capitalize(task.risk)}} {task.stalledReason && task.status !== "done" && {t("stalled")}} + {task.hold && !task.stalledReason && ( + + {task.hold.waitingOn === "operator" ? t("needs you") : t("waiting")} + + )} {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 66cb4322..2dd482df 100644 --- a/src/renderer/src/pages/work/TaskDetailDialog.tsx +++ b/src/renderer/src/pages/work/TaskDetailDialog.tsx @@ -1,3 +1,4 @@ +import { TaskHoldNotice } from "./TaskHoldNotice"; import { useState, type FormEvent } from "react"; import { Avatar, Button, Field, Icon, IconButton, Pill, Segmented } from "../../shared/ui"; import type { @@ -142,7 +143,7 @@ function OverviewTab({ task, assignee, project, goal, workspace, provider, runs }) { return ( <> - {task.stalledReason &&
{task.stalledReason}
} + {task.hold && } {task.description && (

{task.description}

)} diff --git a/src/renderer/src/pages/work/TaskHoldNotice.tsx b/src/renderer/src/pages/work/TaskHoldNotice.tsx new file mode 100644 index 00000000..ffc860d1 --- /dev/null +++ b/src/renderer/src/pages/work/TaskHoldNotice.tsx @@ -0,0 +1,22 @@ +import type { TaskHold } from "../../../../shared/domain"; +import { t } from "../../i18n/i18n"; + +function when(iso: string): string { + return new Date(iso).toLocaleString([], { month: "short", day: "numeric", hour: "2-digit", minute: "2-digit", hour12: false }); +} + +/** Why an open task is not moving, when it resumes on its own, and whether it waits on the operator. */ +export function TaskHoldNotice({ hold }: { hold: TaskHold }) { + const next = hold.waitingOn === "provider" + ? t("Resumes on its own at {time}.", { time: when(hold.resumesAt) }) + : hold.resumesAt + ? t("Waiting on you, or resumes on its own at {time}.", { time: when(hold.resumesAt) }) + : t("Waiting on you."); + const color = hold.waitingOn === "operator" ? "var(--red)" : "var(--amber)"; + return ( +
+
{hold.reason}
+
{next}
+
+ ); +} diff --git a/src/shared/domain.ts b/src/shared/domain.ts index e64038c7..b60bfec4 100644 --- a/src/shared/domain.ts +++ b/src/shared/domain.ts @@ -87,6 +87,11 @@ export interface AgentRecord { createdAt: string; } +/** What stops an open task: the reason, when it resumes on its own (null: only when someone acts), and who that is. */ +export type TaskHold = + | { reason: string; waitingOn: "provider"; resumesAt: string } + | { reason: string; waitingOn: "operator"; resumesAt: string | null }; + export interface TaskRecord { id: string; orgId: string; @@ -104,6 +109,10 @@ export interface TaskRecord { writingMode: WritingMode; /** Set when the task spent its run budget without reaching done. */ stalledReason: string | null; + /** Why the heartbeat last failed to queue the task; cleared once it queues. */ + enqueueError: string | null; + /** Why the open task is not moving now, or null when nothing holds it. */ + hold: TaskHold | null; /** Runs at or below this task_runs rowid no longer count toward the run budget. */ runBudgetFloor: number; createdAt: string; diff --git a/tests/integration/task-holds.test.ts b/tests/integration/task-holds.test.ts new file mode 100644 index 00000000..86338bc3 --- /dev/null +++ b/tests/integration/task-holds.test.ts @@ -0,0 +1,108 @@ +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 { SnapshotService } from "../../src/main/services/snapshot-service"; +import { currentProviderUsageWindow } from "../../src/shared/provider-runtime-domain"; +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"), +}; + +/** A company whose kickoff task is open with no run in flight. */ +function company() { + const turn = async (): Promise => ({ text: "Done.", toolCalls: [], rawResponseJson: "{}" }); + const connection = new DatabaseConnection(join(tmpdir(), `ac-task-holds-${randomUUID()}.db`)); + const providers = new ProviderService(connection.db, providerAdapters, testVault, 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; + }); + const run = connection.db.prepare("select id, task_id, provider_id from task_runs").get() as { id: string; task_id: string; provider_id: string }; + connection.db.prepare("update task_runs set state = 'failed' where id = ?").run(run.id); + const hold = () => new SnapshotService(connection.db).load().tasks.find((task) => task.id === run.task_id)!.hold; + return { connection, providers, execution, orgId, ceoId, taskId: run.task_id, providerId: run.provider_id, hold }; +} + +describe("why an open task is not moving", () => { + test("an open task with nothing in its way has no hold", () => { + const h = company(); + expect(h.hold()).toBeNull(); + h.connection.close(); + }); + + test("a provider cooldown says when the task resumes on its own", () => { + const h = company(); + const until = Date.now() + 90_000; + h.connection.db.prepare("insert into provider_cooldowns(provider_id, label, until_ms, strikes) values (?, 'OpenAI', ?, 2)").run(h.providerId, until); + expect(h.hold()).toEqual({ reason: "OpenAI reported a passing failure; its work waits out a cooldown.", resumesAt: new Date(until).toISOString(), waitingOn: "provider" }); + h.connection.close(); + }); + + test("a spent monthly request limit waits on the operator or the next window, and a zero limit only on the operator", () => { + const h = company(); + const window = currentProviderUsageWindow(); + h.connection.db.prepare("update providers set monthly_request_limit = 2 where id = ?").run(h.providerId); + h.connection.db.prepare(` + insert into provider_usage_windows(id, org_id, provider_id, window_start, window_end, request_count, input_char_count, output_char_count, created_at, updated_at) + values ('puw_t', ?, ?, ?, ?, 2, 0, 0, ?, ?) + `).run(h.orgId, h.providerId, window.start, window.end, new Date().toISOString(), new Date().toISOString()); + expect(h.hold()).toEqual({ reason: "OpenAI used its monthly request limit (2). Raise the limit or wait for the next window.", resumesAt: window.end, waitingOn: "operator" }); + h.connection.db.prepare("update providers set monthly_request_limit = 0 where id = ?").run(h.providerId); + expect(h.hold()).toEqual({ reason: "OpenAI has a monthly request limit of 0. Raise the limit to let its work run.", resumesAt: null, waitingOn: "operator" }); + h.connection.close(); + }); + + test("a spent money budget waits on the operator or next month, and a zero budget only on the operator", () => { + const h = company(); + h.providers.setPrices({ orgId: h.orgId, providerId: h.providerId, inputDollarsPerMtok: 2.5, cachedInputDollarsPerMtok: null, outputDollarsPerMtok: 10 }); + h.connection.db.prepare("update agents set budget_cents = 100 where id = ?").run(h.ceoId); + h.connection.db.prepare(` + insert into provider_requests(id, org_id, provider_id, provider_key, agent_id, model, prompt_sha256, state, created_at, cost_cents) + values ('preq_t', ?, ?, 'openai', ?, 'gpt-5.4', 'x', 'succeeded', ?, 150) + `).run(h.orgId, h.providerId, h.ceoId, new Date().toISOString()); + expect(h.hold()).toEqual({ + reason: "CEO has spent its monthly budget ($1.50 of $1.00). Raise the budget or wait for next month.", + resumesAt: currentProviderUsageWindow().end, + waitingOn: "operator", + }); + h.connection.db.prepare("update agents set budget_cents = 0 where id = ?").run(h.ceoId); + expect(h.hold()).toEqual({ reason: "CEO has spent its monthly budget ($1.50 of $0.00). Raise the budget to let it run.", resumesAt: null, waitingOn: "operator" }); + h.connection.close(); + }); + + test("a heartbeat that could not queue the task records why, until it can", () => { + const h = company(); + h.connection.db.prepare("update tasks set risk = 'high' where id = ?").run(h.taskId); + expect(h.connection.transaction(() => h.execution.enqueueHeartbeatRuns())).toEqual([]); + expect(h.hold()).toEqual({ reason: "Could not start: Approved task execution approval is required.", resumesAt: null, waitingOn: "operator" }); + h.connection.db.prepare("update tasks set risk = 'low' where id = ?").run(h.taskId); + expect(h.connection.transaction(() => h.execution.enqueueHeartbeatRuns())).toHaveLength(1); + expect(h.hold()).toBeNull(); + h.connection.close(); + }); + + test("a stalled task waits on the operator with no resume time", () => { + const h = company(); + h.connection.db.prepare("update tasks set stalled_reason = 'Stalled: 3 sessions made no progress.' where id = ?").run(h.taskId); + expect(h.hold()).toEqual({ reason: "Stalled: 3 sessions made no progress.", resumesAt: null, waitingOn: "operator" }); + h.connection.close(); + }); +}); diff --git a/tests/unit/domain-services.test.ts b/tests/unit/domain-services.test.ts index d14f73bc..d9adc4d5 100644 --- a/tests/unit/domain-services.test.ts +++ b/tests/unit/domain-services.test.ts @@ -35,7 +35,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, 67, 68, 69, 70, 71, 72, 73, 74, 75, 76, 77, 78, 79, 80, 81, 82, 83, 84, 85, 86, 87, 88, 89, 90, 91, 92]); + 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, 68, 69, 70, 71, 72, 73, 74, 75, 76, 77, 78, 79, 80, 81, 82, 83, 84, 85, 86, 87, 88, 89, 90, 91, 92, 93]); connection.close(); });