Q49: a task that spends its run budget shows as stalled; edits start a fresh budget
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
090c37abb8
commit
7659adbbdd
11 files changed
+94
-20
No files matched your search
@@ -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'));
|
||||
`;
|
||||
@@ -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;
|
||||
|
||||
@@ -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<string, 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, 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<string>();
|
||||
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();
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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<string> {
|
||||
|
||||
@@ -30,6 +30,7 @@ export const zh: Record<string, string> = {
|
||||
"Document": "文档",
|
||||
"Writing": "写作方式",
|
||||
"superseded": "已被取代",
|
||||
"stalled": "已停滞",
|
||||
"Why it did not run": "未运行的原因",
|
||||
"Done tasks do not run": "已完成的任务不能再运行",
|
||||
"A session is already queued or running": "已有会话在排队或运行中",
|
||||
|
||||
@@ -337,6 +337,7 @@ function TaskCard({
|
||||
</div>
|
||||
<div className="row" style={{ flexWrap: "wrap", gap: 6, minWidth: 0 }}>
|
||||
{project ? <Pill>{project.name}</Pill> : <Pill>{capitalize(task.risk)}</Pill>}
|
||||
{task.stalledReason && task.status !== "done" && <Pill tone="red">{t("stalled")}</Pill>}
|
||||
{task.status !== "done" && latestRun && (
|
||||
<span style={{ flexShrink: 0, display: "inline-flex" }}>
|
||||
<Pill tone={runTone(latestRun.state)}>{prettyRunState(latestRun.state)}</Pill>
|
||||
|
||||
@@ -142,6 +142,7 @@ function OverviewTab({ task, assignee, project, goal, workspace, provider, runs
|
||||
}) {
|
||||
return (
|
||||
<>
|
||||
{task.stalledReason && <div className="form-error" role="alert" style={{ marginBottom: 12 }}>{task.stalledReason}</div>}
|
||||
{task.description && (
|
||||
<p style={{ margin: "0 0 14px", color: "var(--text-2)", fontSize: 14, lineHeight: 1.55, whiteSpace: "pre-wrap" }}>{task.description}</p>
|
||||
)}
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
@@ -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();
|
||||
});
|
||||
|
||||
|
||||
@@ -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);
|
||||
|
||||
Reference in new issue
Block a user