From 8a424822a5b6225128ac962ca1faf423102e8e8c 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, 26 Mar 2026 19:13:09 -0400 Subject: [PATCH] feat(04-02): wire CAS checkout and stale recovery into orchestration-service - Add randomUUID import for pre-generated runId in CAS checkout flow - Replace task-level hasActiveRun checks with db.checkoutTask CAS in startAgentHeartbeatRun and queueTaskRun - Pass pre-generated runId to createRun (database.ts accepts optional id field) - Log execution.checkout_rejected activity on CAS failure (D-04) - Return fail("CHECKOUT_RACE_LOST") in queueTaskRun on CAS failure - Rewrite wakeAgentIfPossible with proper lifecycle tracking via enqueueWakeupRequest - Add stale lock recovery sweep at startup (resumeQueuedRuns) and each heartbeat tick - Log execution.lock_recovered activity entries for recovered stale locks --- src/main/database.ts | 3 +- src/main/orchestration-service.ts | 81 ++++++++++++++++++++++++++++--- 2 files changed, 76 insertions(+), 8 deletions(-) diff --git a/src/main/database.ts b/src/main/database.ts index 7153887d..232f0f72 100644 --- a/src/main/database.ts +++ b/src/main/database.ts @@ -1196,13 +1196,14 @@ export class AppDatabase { } createRun(input: { + id?: string; companyId: string; taskId: string; agentId: string; workspaceId: string | null; connectorId: ConnectorId; }) { - const id = randomUUID(); + const id = input.id ?? randomUUID(); const ts = nowIso(); const logPath = join(this.logsDir, `${id}.log`); this.transaction(() => { diff --git a/src/main/orchestration-service.ts b/src/main/orchestration-service.ts index 96f49a32..c8426cee 100644 --- a/src/main/orchestration-service.ts +++ b/src/main/orchestration-service.ts @@ -1,3 +1,4 @@ +import { randomUUID } from "node:crypto"; import { buildTaskPrompt } from "./prompt-builder"; import type { AppDatabase } from "./database"; import { getConnectorExecutionReadinessIssue } from "@shared/connector-policy"; @@ -418,8 +419,20 @@ export function createOrchestrationService({ logger.info(`[heartbeat] Created bootstrap task for ${agent.name}: ${taskTitle}`); } - if (task.activeRunId || hasActiveRun({ taskId: task.id, companyId })) { - logger.info(`[heartbeat] Skipping ${agent.name} because task ${task.title} already has an active run.`); + // Generate runId upfront for CAS checkout (EXEC-01, EXEC-02) + const runId = randomUUID(); + const checkoutSuccess = db.checkoutTask(task.id, runId); + if (!checkoutSuccess) { + // D-04: Activity log entry only, no toast/popup + db.addActivity({ + companyId, + actor: "system", + action: "execution.checkout_rejected", + entityType: "task", + entityId: task.id, + detail: `Agent ${agent.name} lost checkout race on task "${task.title}"`, + }); + logger.info(`[heartbeat] Agent ${agent.name} lost checkout race on task ${task.title}`); return null; } @@ -430,6 +443,7 @@ export function createOrchestrationService({ ); const run = db.createRun({ + id: runId, companyId, taskId: task.id, agentId: agent.id, @@ -483,9 +497,16 @@ export function createOrchestrationService({ } const agentBusy = hasActiveRun({ agentId, companyId }); + if (agentBusy) { + // Agent is busy — enqueue wakeup with lifecycle tracking (D-05) + db.enqueueWakeupRequest(companyId, agentId, trigger, `Wake for agent: ${trigger}`); + return null; + } + const runId = startAgentHeartbeatRun(agentId, companyId, trigger); - if (!runId && agentBusy) { - db.enqueueWakeupRequest(companyId, agentId, trigger, trigger); + if (!runId) { + // Run failed to start but agent not busy — still enqueue for retry + db.enqueueWakeupRequest(companyId, agentId, trigger, `Wake for agent (deferred): ${trigger}`); } return runId; } @@ -503,9 +524,6 @@ export function createOrchestrationService({ if (companyStatus !== "active") { return fail("COMPANY_NOT_ACTIVE", "Runs can only start while the company is active."); } - if (task.activeRunId || hasActiveRun({ taskId: task.id, companyId: task.companyId })) { - return fail("TASK_ALREADY_RUNNING", "This task already has an active run."); - } if (!task.assigneeAgentId) return fail("TASK_UNASSIGNED", "Assign an agent before starting a run."); const agent = db.getAgent(task.assigneeAgentId); if (agent.companyId !== task.companyId) { @@ -536,6 +554,13 @@ export function createOrchestrationService({ ); } + // CAS checkout — atomic claim (EXEC-01, EXEC-02) + const runId = randomUUID(); + const checkoutSuccess = db.checkoutTask(task.id, runId); + if (!checkoutSuccess) { + return fail("CHECKOUT_RACE_LOST", "Another run claimed this task. Refresh and try again."); + } + const workspaceResolution = resolveWorkspace(task, agent, task.companyId); const workspace = workspaceResolution.workspace; if (workspace && workspace.companyId !== task.companyId) { @@ -543,6 +568,7 @@ export function createOrchestrationService({ } const run = db.createRun({ + id: runId, companyId: task.companyId, taskId: task.id, agentId: agent.id, @@ -577,6 +603,27 @@ export function createOrchestrationService({ } function resumeQueuedRuns() { + // D-02: Stale lock recovery at app startup (EXEC-03) + try { + const recovered = db.recoverStaleLocks(); + for (const task of recovered) { + db.addActivity({ + companyId: task.companyId, + actor: "system", + action: "execution.lock_recovered", + entityType: "task", + entityId: task.taskId, + detail: `Recovered stale lock on task "${task.title}" after 15min timeout (startup sweep)`, + }); + logger.info(`[init] Recovered stale lock on task ${task.taskId}: "${task.title}"`); + } + if (recovered.length > 0) { + logger.info(`[init] Recovered ${recovered.length} stale lock(s) at startup`); + } + } catch (err) { + logger.error(`[init] Stale lock recovery at startup failed: ${err instanceof Error ? err.message : String(err)}`); + } + if (isQuitting() || !hasWorkerReady()) { return; } @@ -632,6 +679,26 @@ export function createOrchestrationService({ heartbeatTimer = setInterval(() => { try { if (isQuitting() || !hasWorkerReady()) return; + + // D-02: Stale lock recovery piggybacked on heartbeat cycle (EXEC-03) + try { + const recovered = db.recoverStaleLocks(); + for (const task of recovered) { + // D-04: Activity log entry only + db.addActivity({ + companyId: task.companyId, + actor: "system", + action: "execution.lock_recovered", + entityType: "task", + entityId: task.taskId, + detail: `Recovered stale lock on task "${task.title}" after 15min timeout`, + }); + logger.info(`[heartbeat] Recovered stale lock on task ${task.taskId}: "${task.title}"`); + } + } catch (err) { + logger.error(`[heartbeat] Stale lock recovery failed: ${err instanceof Error ? err.message : String(err)}`); + } + const activeCompanies = db.listActiveCompanyIds(); const now = Date.now();