From 2dd62e53383291624aaf73c8bb7fbae2e5206edc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=9B=B7=E7=94=B5=E8=8A=BD=E8=A1=A3?= Date: Mon, 20 Apr 2026 15:32:55 -0400 Subject: [PATCH] Adapt heartbeat/task-runner/turn-context to lockMode=none first-class path - heartbeat.ts and task-runner.ts now share a single ResolveWorkspaceFn signature exported from run-dispatch.ts. Callers handle WorkspaceConfigurationError explicitly: * heartbeat: release the task lease and skip with new workspace_unresolved reason (task re-enters pool next heartbeat) * task-runner: release the lease and return WORKSPACE_UNRESOLVED to the caller No silent fallback to a random shared workspace. - crash-recovery.ts imports WorkspaceConfigurationError for clean catch-branch attribution; the outer try already rolls back the queued run on any resolve failure. - run-dispatch.ts exposes ResolveWorkspaceFn so downstream modules type their injected resolver against the same shape (single source of truth). - HeartbeatSkipReason gains workspace_unresolved to keep the closed enum honest about all heartbeat exit paths. --- src/main/orchestration/heartbeat.ts | 36 ++++++++++++++++++++++--- src/main/orchestration/run-dispatch.ts | 26 ++++++++++++------ src/main/orchestration/task-runner.ts | 19 ++++++++++--- src/main/worker/entry/crash-recovery.ts | 5 +++- 4 files changed, 71 insertions(+), 15 deletions(-) diff --git a/src/main/orchestration/heartbeat.ts b/src/main/orchestration/heartbeat.ts index c5192c9b..7d8017ba 100644 --- a/src/main/orchestration/heartbeat.ts +++ b/src/main/orchestration/heartbeat.ts @@ -3,6 +3,8 @@ import type { AppDatabase } from "../database"; import type { AgentRecord, ConnectorRecord } from "@shared/types"; import type { OrchestrationDependencies } from "./types"; import type { RunDispatchParams } from "./turn-context"; +import type { ResolveWorkspaceFn } from "./run-dispatch"; +import { WorkspaceConfigurationError } from "./resolve-workspace"; import { pickModelForAgent } from "../model-dispatch"; import { HUMAN_SENDER_ID } from "../messaging/dispatch"; @@ -28,7 +30,8 @@ export type HeartbeatSkipReason = | "awaiting_user_review" | "pending_deliverable_review" | "checkout_race_lost" - | "no_actionable_tasks"; + | "no_actionable_tasks" + | "workspace_unresolved"; export type HeartbeatStartResult = | { status: "started"; runId: string } @@ -41,7 +44,7 @@ const SKIPPED = (reason: HeartbeatSkipReason): HeartbeatStartResult => ({ export function createHeartbeat( deps: OrchestrationDependencies, - resolveWorkspace: (task: { workspaceId: string | null; projectId?: string | null } | null, agent: { workspaceId: string | null }, companyId: string) => { workspace: ReturnType; lockMode: "exclusive" | "shared" | "none"; source: string }, + resolveWorkspace: ResolveWorkspaceFn, getRunEligibilityFailure: (agent: AgentRecord, connector: ConnectorRecord, projectId?: string | null) => string | null, dispatchRunToWorker: (params: RunDispatchParams) => void, hasActiveRun: (filter: { taskId?: string; agentId?: string; companyId: string }) => boolean, @@ -201,7 +204,34 @@ export function createHeartbeat( logger.info(`[heartbeat] Starting taskless heartbeat run for ${agent.name}`); } - const workspaceResolution = resolveWorkspace(task, agent, companyId); + let workspaceResolution: ReturnType; + try { + workspaceResolution = resolveWorkspace(task, agent, companyId); + } catch (error) { + if (error instanceof WorkspaceConfigurationError) { + // A coding-role task landed here with no bound workspace. We + // refuse to silently pick a shared workspace (the old + // company_fallback silent-corruption path) — the operator + // needs to bind a workspace explicitly. Release the lease + // the earlier checkoutTask just claimed so the task re-enters + // the pool at the next heartbeat instead of stalling as + // `in_progress` with no run. + if (task) { + db.tasks.releaseTaskLock(task.id, runId); + } + db.activityMod.addActivity({ + companyId, + actor: "system", + action: "heartbeat.workspace_unresolved", + entityType: "agent", + entityId: agent.id, + detail: error.message, + }); + logger.warn(`[heartbeat] Workspace unresolved for ${agent.name}: ${error.message}`); + return SKIPPED("workspace_unresolved"); + } + throw error; + } const workspace = workspaceResolution.workspace; logger.info( `[heartbeat] Workspace resolution for ${agent.name}: task.workspaceId=${task?.workspaceId ?? "null"}, agent.workspaceId=${agent.workspaceId ?? "null"}, projectId=${task?.projectId ?? "null"}, source=${workspaceResolution.source}, mode=${workspaceResolution.lockMode}, resolved=${workspace?.name ?? "NONE"}, localPath=${workspace?.localPath ?? "NONE"}`, diff --git a/src/main/orchestration/run-dispatch.ts b/src/main/orchestration/run-dispatch.ts index 20a1fc3c..fe558ca7 100644 --- a/src/main/orchestration/run-dispatch.ts +++ b/src/main/orchestration/run-dispatch.ts @@ -1,10 +1,25 @@ import { getConnectorExecutionReadinessIssue } from "@shared/connector-policy"; -import type { ConnectorRecord } from "@shared/types"; +import type { AgentRecord, ConnectorRecord, DepartmentId, TaskRecord } from "@shared/types"; import type { OrchestrationDependencies } from "./types"; import { buildRunPayload, type RunDispatchParams } from "./turn-context"; import { validateConnectorModelSelection } from "../connector-model-policy"; import { resolveWorkspaceWithLockMode } from "./resolve-workspace"; +type ResolveWorkspaceTaskInput = + | (Pick & { requiresFilesystem?: boolean }) + | null; + +type ResolveWorkspaceAgentInput = Pick & { + role?: string | null; + department?: DepartmentId | null; +}; + +export type ResolveWorkspaceFn = ( + task: ResolveWorkspaceTaskInput, + agent: ResolveWorkspaceAgentInput, + companyId: string, +) => ReturnType; + export function createRunDispatch(deps: OrchestrationDependencies) { const { db, @@ -20,13 +35,8 @@ export function createRunDispatch(deps: OrchestrationDependencies) { evaluatePreRunSessionRotation, } = deps; - function resolveWorkspace( - task: { workspaceId: string | null; projectId?: string | null } | null, - agent: { workspaceId: string | null }, - companyId: string, - ) { - return resolveWorkspaceWithLockMode(db, task, agent, companyId); - } + const resolveWorkspace: ResolveWorkspaceFn = (task, agent, companyId) => + resolveWorkspaceWithLockMode(db, task, agent, companyId); function getRunEligibilityFailure( agent: { status: string; budgetMonthlyUsd: number; spentMonthlyUsd: number; name: string; connectorId: string; id: string; companyId: string; model?: string | null }, diff --git a/src/main/orchestration/task-runner.ts b/src/main/orchestration/task-runner.ts index 76fae0f8..cc733ee9 100644 --- a/src/main/orchestration/task-runner.ts +++ b/src/main/orchestration/task-runner.ts @@ -1,14 +1,15 @@ import { randomUUID } from "node:crypto"; -import type { AppDatabase } from "../database"; import type { AgentRecord, ConnectorRecord } from "@shared/types"; import type { DesktopResult } from "@shared/contracts"; import type { OrchestrationDependencies } from "./types"; import type { RunDispatchParams } from "./turn-context"; +import type { ResolveWorkspaceFn } from "./run-dispatch"; +import { WorkspaceConfigurationError } from "./resolve-workspace"; import { pickModelForAgent } from "../model-dispatch"; export function createTaskRunner( deps: OrchestrationDependencies, - resolveWorkspace: (task: { workspaceId: string | null; projectId?: string | null } | null, agent: { workspaceId: string | null }, companyId: string) => { workspace: ReturnType; lockMode: "exclusive" | "shared" | "none"; source: string }, + resolveWorkspace: ResolveWorkspaceFn, getRunEligibilityFailure: (agent: AgentRecord, connector: ConnectorRecord, projectId?: string | null) => string | null, dispatchRunToWorker: (params: RunDispatchParams) => void, hasActiveRun: (filter: { taskId?: string; agentId?: string; companyId: string }) => boolean, @@ -93,7 +94,19 @@ export function createTaskRunner( return fail("CHECKOUT_RACE_LOST", "Another run claimed this task. Refresh and try again."); } - const workspaceResolution = resolveWorkspace(task, agent, task.companyId); + let workspaceResolution: ReturnType; + try { + workspaceResolution = resolveWorkspace(task, agent, task.companyId); + } catch (error) { + if (error instanceof WorkspaceConfigurationError) { + // Resolver refused to guess a workspace. Release the lease + // we just claimed so the task row does not stall at + // `in_progress`; surface the error to the caller. + db.tasks.releaseTaskLock(task.id, runId); + return fail("WORKSPACE_UNRESOLVED", error.message); + } + throw error; + } const workspace = workspaceResolution.workspace; if (workspace && workspace.companyId !== task.companyId) { return fail("WORKSPACE_COMPANY_MISMATCH", "The selected workspace does not belong to the task company."); diff --git a/src/main/worker/entry/crash-recovery.ts b/src/main/worker/entry/crash-recovery.ts index 4e768def..3b2111fe 100644 --- a/src/main/worker/entry/crash-recovery.ts +++ b/src/main/worker/entry/crash-recovery.ts @@ -1,5 +1,8 @@ import type { WorkerServiceDependencies } from "./types"; -import { resolveWorkspaceWithLockMode } from "../../orchestration/resolve-workspace"; +import { + resolveWorkspaceWithLockMode, + WorkspaceConfigurationError, +} from "../../orchestration/resolve-workspace"; export function handleWorkerExit( code: number | null,