Q44: queued runs re-check their task and are superseded when it changed

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

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;

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<string> {
     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<TaskExecutionService["executeAgentRun"]>[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

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,