diff --git a/src/main/services/heartbeat-scheduler.ts b/src/main/services/heartbeat-scheduler.ts index 7a4f99bf..af5872fc 100644 --- a/src/main/services/heartbeat-scheduler.ts +++ b/src/main/services/heartbeat-scheduler.ts @@ -49,7 +49,7 @@ export class HeartbeatScheduler { `).all(window.start) as Array<{ task_id: string; org_id: string; provider_id: string }>; 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 = ?"); + const attemptMessages = this.db.prepare("select error_message from task_runs where task_id = ? and state <> 'superseded'"); const created: string[] = []; for (const row of rows) { if (cooling.has(row.provider_id)) continue; diff --git a/src/main/services/task-execution-service.ts b/src/main/services/task-execution-service.ts index 9f19204d..a69746b1 100644 --- a/src/main/services/task-execution-service.ts +++ b/src/main/services/task-execution-service.ts @@ -33,13 +33,17 @@ import { connectorDisplayName } from "./composio/connector-brands"; import { ConnectorGrantsService } from "./composio/connector-grants"; import { ProviderCooldowns } from "./provider-cooldowns"; import { HeartbeatScheduler } from "./heartbeat-scheduler"; -import { assertExecutionApproval, createIsolationContext, loadGoal, loadProject, loadWorkspace, requireAgentProvider, requireAssignedAgent, requireTask } from "./task-run-preparation"; +import { + assertExecutionApproval, createIsolationContext, executionApprovalProblem, loadGoal, loadProject, loadWorkspace, + requireAgentProvider, requireAssignedAgent, requireTask, sameWorkspaceAccess, workspaceAccess, +} from "./task-run-preparation"; const TASK_RUN_STATE = { QUEUED: "queued", RUNNING: "running", COMPLETED: "completed", FAILED: "failed", + SUPERSEDED: "superseded", } as const; @@ -94,11 +98,7 @@ export class TaskExecutionService { return this.executeAgentRun({ orgId: input.orgId, runId, task, agent, provider, workspace, isolation }); } - /** - * Executes an existing `running` task_run (claimed by the autonomous queue - * worker). System-initiated, so it does not re-gate on the manual execution - * approval — the operator configured the company to run autonomously. - */ + /** Executes a claimed run after re-checking that the task still matches it; a mismatch supersedes the run. */ async executeRun(orgId: string, runId: string): Promise { const run = this.db.prepare(` select task_id, agent_id, provider_id, execution_context_id, service_identity_id, workspace_mount_json, secret_scope_json @@ -108,12 +108,21 @@ export class TaskExecutionService { let prepared: Parameters[0]; try { const task = requireTask(this.db, orgId, run.task_id); + if (task.status === TASK_STATUS.DONE) return this.supersedeRun(orgId, runId, run, "the task was closed after this run was queued."); if (task.assignee_agent_id !== run.agent_id) { - throw new Error(`Task was reassigned after this run was queued (run agent ${run.agent_id}, assignee ${task.assignee_agent_id ?? "none"}).`); + return this.supersedeRun(orgId, runId, run, `the task was reassigned after this run was queued (run agent ${run.agent_id}, assignee ${task.assignee_agent_id ?? "none"}).`); } const agent = requireAssignedAgent(this.db, orgId, task); const provider = requireAgentProvider(this.db, orgId, agent); + if (provider.id !== run.provider_id) { + return this.supersedeRun(orgId, runId, run, `the assignee's provider changed after this run was queued (${run.provider_id ?? "none"} -> ${provider.id}).`); + } const workspace = loadWorkspace(this.db, task.workspace_id); + if (!sameWorkspaceAccess(run.workspace_mount_json, workspaceAccess(task, workspace))) { + return this.supersedeRun(orgId, runId, run, "the task's workspace access changed after this run was queued."); + } + const approvalProblem = executionApprovalProblem(this.db, orgId, task); + if (approvalProblem) return this.supersedeRun(orgId, runId, run, approvalProblem); const isolation: TaskRunIsolation = { executionContextId: run.execution_context_id, serviceIdentityId: run.service_identity_id, @@ -389,6 +398,25 @@ export class TaskExecutionService { this.organization.updateTaskStatus(orgId, run.task_id, TASK_STATUS.DONE); } + /** Closes a queued run that no longer matches its task; it is not a failure and spends no retry budget. */ + private supersedeRun(orgId: string, runId: string, run: RunRow, reason: string): string { + const message = `Superseded: ${reason}`; + console.info(`[task-run] run ${runId} ${message}`); + this.db.prepare("update task_runs set state = ?, completed_at = ?, error_message = ? where id = ? and org_id = ?") + .run(TASK_RUN_STATE.SUPERSEDED, new Date().toISOString(), message, runId, orgId); + this.closeChatWakeTask(orgId, runId); + this.audit.record({ + orgId, + entityType: "task_run", + entityId: runId, + actorType: "system", + actorId: null, + action: "superseded", + payload: { taskId: run.task_id, agentId: run.agent_id, providerId: run.provider_id, reason }, + }); + return runId; + } + private failRun(orgId: string, runId: string, providerRequestId: string | null, errorMessage: string): void { this.db.prepare(` update task_runs diff --git a/src/main/services/task-run-preparation.ts b/src/main/services/task-run-preparation.ts index ff181349..38ab9d15 100644 --- a/src/main/services/task-run-preparation.ts +++ b/src/main/services/task-run-preparation.ts @@ -34,7 +34,13 @@ export function requireAssignedAgent(db: DatabaseSync, orgId: string, task: Task } export function assertExecutionApproval(db: DatabaseSync, orgId: string, task: TaskExecutionTaskRow): void { - if (task.risk !== RISK_LEVEL.HIGH && task.requires_filesystem !== 1) return; + const problem = executionApprovalProblem(db, orgId, task); + if (problem) throw new Error(problem); +} + +/** Why the task may not execute without approval, or null when it may. */ +export function executionApprovalProblem(db: DatabaseSync, orgId: string, task: TaskExecutionTaskRow): string | null { + if (task.risk !== RISK_LEVEL.HIGH && task.requires_filesystem !== 1) return null; const approval = db.prepare(` select id, state from approvals @@ -45,9 +51,20 @@ export function assertExecutionApproval(db: DatabaseSync, orgId: string, task: T order by created_at desc limit 1 `).get(orgId, task.id) as TaskApprovalRow | undefined; - if (!approval || approval.state !== APPROVAL_STATE.APPROVED) { - throw new Error("Approved task execution approval is required."); - } + if (!approval || approval.state !== APPROVAL_STATE.APPROVED) return "Approved task execution approval is required."; + return null; +} + +/** The workspace a run of this task may touch, and how. */ +export function workspaceAccess(task: TaskExecutionTaskRow, workspace: WorkspaceRow | null): Array<{ workspaceId: string; mode: string }> { + return workspace ? [{ workspaceId: workspace.id, mode: task.requires_filesystem === 1 ? "read_write" : "read_only" }] : []; +} + +/** True when a run's recorded mounts grant exactly the access the task needs now. */ +export function sameWorkspaceAccess(mountJson: string, current: Array<{ workspaceId: string; mode: string }>): boolean { + const recorded = (JSON.parse(mountJson) as Array<{ workspaceId: string; mode: string }>) + .map((mount) => ({ workspaceId: mount.workspaceId, mode: mount.mode })); + return JSON.stringify(recorded) === JSON.stringify(current); } export function requireAgentProvider(db: DatabaseSync, orgId: string, agent: TaskExecutionAgentRow): TaskExecutionProviderRow { @@ -88,11 +105,10 @@ export function createIsolationContext( const serviceIdentityId = requireServiceIdentity(db, orgId, agent.id); const executionContextId = createId("ctx"); const workspaceMounts = workspace ? [{ - workspaceId: workspace.id, + ...workspaceAccess(task, workspace)[0], name: workspace.name, localPath: workspace.local_path, repoRef: workspace.repo_ref, - mode: task.requires_filesystem === 1 ? "read_write" : "read_only", }] : []; return { executionContextId, diff --git a/src/renderer/src/i18n/zh.ts b/src/renderer/src/i18n/zh.ts index c5dce819..41bbd1e9 100644 --- a/src/renderer/src/i18n/zh.ts +++ b/src/renderer/src/i18n/zh.ts @@ -29,6 +29,8 @@ export const zh: Record = { "Integrations": "集成密钥", "Document": "文档", "Writing": "写作方式", + "superseded": "已被取代", + "Why it did not run": "未运行的原因", "Done tasks do not run": "已完成的任务不能再运行", "A session is already queued or running": "已有会话在排队或运行中", "Conversational (house style)": "日常沟通（公司文风）", diff --git a/src/renderer/src/pages/HomePage.tsx b/src/renderer/src/pages/HomePage.tsx index 90e20688..47fcb106 100644 --- a/src/renderer/src/pages/HomePage.tsx +++ b/src/renderer/src/pages/HomePage.tsx @@ -344,5 +344,6 @@ function runStateLabel(state: string): string { if (state === "running") return t("Running"); if (state === "completed") return t("Done"); if (state === "failed") return t("Failed"); + if (state === "superseded") return t("Superseded"); return state; } diff --git a/src/renderer/src/pages/RunsPage.tsx b/src/renderer/src/pages/RunsPage.tsx index f0f5e869..9419ba02 100644 --- a/src/renderer/src/pages/RunsPage.tsx +++ b/src/renderer/src/pages/RunsPage.tsx @@ -138,6 +138,12 @@ function RunDetail({ run, tasks, agents, providers, providerRequests }: { {t("Context")}
{run.executionContextId}
+ {run.state === "superseded" && run.errorMessage && ( +
+
{t("Why it did not run")}
+
{run.errorMessage}
+
+ )} {run.state === "failed" && run.errorMessage && (
{t("Error")}
@@ -197,6 +203,7 @@ function runLabel(state: string): string { if (state === "running") return t("Running"); if (state === "completed") return t("Done"); if (state === "failed") return t("Failed"); + if (state === "superseded") return t("Superseded"); return state; } diff --git a/src/renderer/src/pages/work/task-helpers.ts b/src/renderer/src/pages/work/task-helpers.ts index bb06f1e7..0b1454e3 100644 --- a/src/renderer/src/pages/work/task-helpers.ts +++ b/src/renderer/src/pages/work/task-helpers.ts @@ -27,6 +27,7 @@ export function prettyRunState(state: string): string { if (state === "running") return t("running"); if (state === "completed") return t("completed"); if (state === "failed") return t("failed"); + if (state === "superseded") return t("superseded"); return state.replaceAll("_", " "); } diff --git a/tests/integration/run-memory.test.ts b/tests/integration/run-memory.test.ts index e2c65925..8fc4f821 100644 --- a/tests/integration/run-memory.test.ts +++ b/tests/integration/run-memory.test.ts @@ -80,10 +80,10 @@ describe("run memory", () => { test("a run that fails before it starts is remembered too", async () => { const h = await company(() => ({ text: "unused", toolCalls: [], rawResponseJson: "{}" })); const kickoff = claimNextQueuedRun(h.connection)!; - h.connection.db.prepare("update tasks set assignee_agent_id = null where id = (select task_id from task_runs where id = ?)").run(kickoff.id); - await expect(h.execution.executeRun(h.orgId, kickoff.id)).rejects.toThrow("Task was reassigned after this run was queued"); + h.connection.db.prepare("delete from providers where id = (select provider_id from task_runs where id = ?)").run(kickoff.id); + await expect(h.execution.executeRun(h.orgId, kickoff.id)).rejects.toThrow("Task assignee provider was not found."); expect(h.memories().map((entry) => entry.category)).toEqual(["failure_cause"]); - expect(h.memories()[0].content).toContain("Task was reassigned after this run was queued"); + expect(h.memories()[0].content).toContain("Task assignee provider was not found."); h.connection.close(); }); diff --git a/tests/integration/stale-queued-runs.test.ts b/tests/integration/stale-queued-runs.test.ts new file mode 100644 index 00000000..ddd7e230 --- /dev/null +++ b/tests/integration/stale-queued-runs.test.ts @@ -0,0 +1,131 @@ +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 { claimNextQueuedRun } from "../../src/main/app/task-queue-worker"; +import { ChatService } from "../../src/main/services/chat-service"; +import { ensureCeoDirectSpace } from "../../src/main/services/ceo-direct-space"; +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"), +}; + +function company() { + let calls = 0; + const turn = async (): Promise => { + calls += 1; + return { text: "done", toolCalls: [], rawResponseJson: "{}" }; + }; + const connection = new DatabaseConnection(join(tmpdir(), `ac-stale-${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; + }); + const claim = () => { + const run = claimNextQueuedRun(connection)!; + const taskId = (connection.db.prepare("select task_id from task_runs where id = ?").get(run.id) as { task_id: string }).task_id; + return { runId: run.id, taskId }; + }; + const runRow = (runId: string) => connection.db.prepare("select state, error_message from task_runs where id = ?").get(runId) as { state: string; error_message: string }; + const memories = () => (connection.db.prepare("select count(*) as n from knowledge_entries where org_id = ?").get(orgId) as { n: number }).n; + return { connection, execution, providers, orgId, ceoId, claim, runRow, memories, calls: () => calls }; +} + +describe("stale queued runs", () => { + test("a run whose task closed after it was queued is superseded, not failed or remembered", async () => { + const h = company(); + const { runId, taskId } = h.claim(); + h.connection.db.prepare("update tasks set status = 'done' where id = ?").run(taskId); + await expect(h.execution.executeRun(h.orgId, runId)).resolves.toBe(runId); + expect(h.runRow(runId)).toEqual({ state: "superseded", error_message: "Superseded: the task was closed after this run was queued." }); + expect(h.memories()).toBe(0); + expect(h.calls()).toBe(0); + h.connection.close(); + }); + + test("a provider switch after queueing supersedes the run and the next heartbeat queues a fresh one", async () => { + const h = company(); + const { runId, taskId } = h.claim(); + const newProvider = (h.connection.db.prepare("select id from providers where org_id = ? and provider_key = 'anthropic'").get(h.orgId) as { id: string }).id; + h.providers.saveCredential({ orgId: h.orgId, providerId: newProvider, secret: "sk-ant", model: "claude-sonnet-5-5" }); + h.connection.db.prepare("update agents set model_provider_id = ? where id = ?").run(newProvider, h.ceoId); + await h.execution.executeRun(h.orgId, runId); + expect(h.runRow(runId).state).toBe("superseded"); + expect(h.runRow(runId).error_message).toContain("the assignee's provider changed"); + + const requeued = h.connection.transaction(() => h.execution.enqueueHeartbeatRuns()); + expect(requeued).toHaveLength(1); + expect(h.connection.db.prepare("select task_id, provider_id from task_runs where id = ?").get(requeued[0])).toEqual({ task_id: taskId, provider_id: newProvider }); + h.connection.close(); + }); + + test("workspace scope and approval changes after queueing supersede the run", async () => { + const h = company(); + const first = h.claim(); + h.connection.db.prepare("update tasks set risk = 'high' where id = ?").run(first.taskId); + await h.execution.executeRun(h.orgId, first.runId); + expect(h.runRow(first.runId)).toEqual({ state: "superseded", error_message: "Superseded: Approved task execution approval is required." }); + + h.connection.db.prepare("update tasks set risk = 'low', requires_filesystem = 0 where id = ?").run(first.taskId); + h.connection.db.prepare("update task_runs set state = 'queued', error_message = null, completed_at = null where id = ?").run(first.runId); + h.connection.db.prepare("update task_runs set workspace_mount_json = ? where id = ?") + .run(JSON.stringify([{ workspaceId: "wsp_elsewhere", mode: "read_only" }]), first.runId); + const second = h.claim(); + await h.execution.executeRun(h.orgId, second.runId); + expect(h.runRow(second.runId).error_message).toBe("Superseded: the task's workspace access changed after this run was queued."); + expect(h.calls()).toBe(0); + h.connection.close(); + }); + + test("a superseded chat wake closes its reply task instead of being re-queued as background work", async () => { + const h = company(); + const kickoff = h.claim(); + await h.execution.executeRun(h.orgId, kickoff.runId); + const founder = (h.connection.db.prepare("select id from employees where org_id = ?").get(h.orgId) as { id: string }).id; + new ChatService(h.connection.db).sendMessage({ + orgId: h.orgId, spaceId: ensureCeoDirectSpace(h.connection.db, h.orgId)!, threadId: null, + senderType: "employee", senderId: founder, bodyMd: "Status?", attachmentFileIds: [], + }); + const wake = h.connection.db.prepare("select id, task_id from task_runs where agent_id = ? and trigger = 'chat_mention' and state = 'queued'").get(h.ceoId) as { id: string; task_id: string }; + h.connection.db.prepare("update task_runs set state = 'running' where id = ?").run(wake.id); + h.connection.db.prepare("update tasks set assignee_agent_id = null where id = ?").run(wake.task_id); + await h.execution.executeRun(h.orgId, wake.id); + expect(h.runRow(wake.id).state).toBe("superseded"); + expect(h.connection.db.prepare("select status from tasks where id = ?").get(wake.task_id)).toEqual({ status: "done" }); + h.connection.close(); + }); + + test("a reassigned task supersedes its queued run; superseded runs never spend the retry budget", async () => { + const h = company(); + const { runId, taskId } = h.claim(); + h.connection.db.prepare("update tasks set assignee_agent_id = null where id = ?").run(taskId); + await h.execution.executeRun(h.orgId, runId); + expect(h.runRow(runId).state).toBe("superseded"); + expect(h.runRow(runId).error_message).toContain("reassigned"); + h.connection.db.prepare("update tasks set assignee_agent_id = ? where id = ?").run(h.ceoId, taskId); + for (let index = 0; index < 4; index += 1) { + h.connection.db.prepare("insert into task_runs(id, org_id, task_id, agent_id, provider_id, trigger, state, created_at, error_message) values (?, ?, ?, ?, null, 'heartbeat', 'superseded', ?, 'Superseded: x')") + .run(`run_s${index}`, h.orgId, taskId, h.ceoId, new Date().toISOString()); + } + expect(h.connection.transaction(() => h.execution.enqueueHeartbeatRuns())).toHaveLength(1); + h.connection.close(); + }); +}); diff --git a/tests/integration/task-context-scope.test.ts b/tests/integration/task-context-scope.test.ts index f1b39f4c..267c1a6c 100644 --- a/tests/integration/task-context-scope.test.ts +++ b/tests/integration/task-context-scope.test.ts @@ -115,8 +115,9 @@ describe("task prompt scope", () => { const taskId = (connection.db.prepare("select task_id from task_runs where id = ?").get(kickoff.id) as { task_id: string }).task_id; connection.db.prepare("update tasks set assignee_agent_id = ? where id = ?").run(otherId, taskId); - await expect(execution.executeRun(orgId, kickoff.id)).rejects.toThrow("Task was reassigned after this run was queued"); - expect(connection.db.prepare("select state from task_runs where id = ?").get(kickoff.id)).toEqual({ state: "failed" }); + await execution.executeRun(orgId, kickoff.id); + expect(connection.db.prepare("select state, error_message like '%reassigned%' as reassigned from task_runs where id = ?").get(kickoff.id)) + .toEqual({ state: "superseded", reassigned: 1 }); expect(calls).toBe(0); connection.close(); });