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.
This commit is contained in:
1 parent
28cdf7c56f
commit
2dd62e5338
4 files changed
+71
-15
No files matched your search
@@ -3,6 +3,8 @@ import type { AppDatabase } from "../database";
|
|||||||
import type { AgentRecord, ConnectorRecord } from "@shared/types";
|
import type { AgentRecord, ConnectorRecord } from "@shared/types";
|
||||||
import type { OrchestrationDependencies } from "./types";
|
import type { OrchestrationDependencies } from "./types";
|
||||||
import type { RunDispatchParams } from "./turn-context";
|
import type { RunDispatchParams } from "./turn-context";
|
||||||
|
import type { ResolveWorkspaceFn } from "./run-dispatch";
|
||||||
|
import { WorkspaceConfigurationError } from "./resolve-workspace";
|
||||||
import { pickModelForAgent } from "../model-dispatch";
|
import { pickModelForAgent } from "../model-dispatch";
|
||||||
import { HUMAN_SENDER_ID } from "../messaging/dispatch";
|
import { HUMAN_SENDER_ID } from "../messaging/dispatch";
|
||||||
|
|
||||||
@@ -28,7 +30,8 @@ export type HeartbeatSkipReason =
|
|||||||
| "awaiting_user_review"
|
| "awaiting_user_review"
|
||||||
| "pending_deliverable_review"
|
| "pending_deliverable_review"
|
||||||
| "checkout_race_lost"
|
| "checkout_race_lost"
|
||||||
| "no_actionable_tasks";
|
| "no_actionable_tasks"
|
||||||
|
| "workspace_unresolved";
|
||||||
|
|
||||||
export type HeartbeatStartResult =
|
export type HeartbeatStartResult =
|
||||||
| { status: "started"; runId: string }
|
| { status: "started"; runId: string }
|
||||||
@@ -41,7 +44,7 @@ const SKIPPED = (reason: HeartbeatSkipReason): HeartbeatStartResult => ({
|
|||||||
|
|
||||||
export function createHeartbeat(
|
export function createHeartbeat(
|
||||||
deps: OrchestrationDependencies,
|
deps: OrchestrationDependencies,
|
||||||
resolveWorkspace: (task: { workspaceId: string | null; projectId?: string | null } | null, agent: { workspaceId: string | null }, companyId: string) => { workspace: ReturnType<AppDatabase["workspacesMod"]["getWorkspace"]>; lockMode: "exclusive" | "shared" | "none"; source: string },
|
resolveWorkspace: ResolveWorkspaceFn,
|
||||||
getRunEligibilityFailure: (agent: AgentRecord, connector: ConnectorRecord, projectId?: string | null) => string | null,
|
getRunEligibilityFailure: (agent: AgentRecord, connector: ConnectorRecord, projectId?: string | null) => string | null,
|
||||||
dispatchRunToWorker: (params: RunDispatchParams) => void,
|
dispatchRunToWorker: (params: RunDispatchParams) => void,
|
||||||
hasActiveRun: (filter: { taskId?: string; agentId?: string; companyId: string }) => boolean,
|
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}`);
|
logger.info(`[heartbeat] Starting taskless heartbeat run for ${agent.name}`);
|
||||||
}
|
}
|
||||||
|
|
||||||
const workspaceResolution = resolveWorkspace(task, agent, companyId);
|
let workspaceResolution: ReturnType<typeof resolveWorkspace>;
|
||||||
|
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;
|
const workspace = workspaceResolution.workspace;
|
||||||
logger.info(
|
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"}`,
|
`[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"}`,
|
||||||
|
|||||||
@@ -1,10 +1,25 @@
|
|||||||
import { getConnectorExecutionReadinessIssue } from "@shared/connector-policy";
|
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 type { OrchestrationDependencies } from "./types";
|
||||||
import { buildRunPayload, type RunDispatchParams } from "./turn-context";
|
import { buildRunPayload, type RunDispatchParams } from "./turn-context";
|
||||||
import { validateConnectorModelSelection } from "../connector-model-policy";
|
import { validateConnectorModelSelection } from "../connector-model-policy";
|
||||||
import { resolveWorkspaceWithLockMode } from "./resolve-workspace";
|
import { resolveWorkspaceWithLockMode } from "./resolve-workspace";
|
||||||
|
|
||||||
|
type ResolveWorkspaceTaskInput =
|
||||||
|
| (Pick<TaskRecord, "workspaceId" | "projectId"> & { requiresFilesystem?: boolean })
|
||||||
|
| null;
|
||||||
|
|
||||||
|
type ResolveWorkspaceAgentInput = Pick<AgentRecord, "workspaceId"> & {
|
||||||
|
role?: string | null;
|
||||||
|
department?: DepartmentId | null;
|
||||||
|
};
|
||||||
|
|
||||||
|
export type ResolveWorkspaceFn = (
|
||||||
|
task: ResolveWorkspaceTaskInput,
|
||||||
|
agent: ResolveWorkspaceAgentInput,
|
||||||
|
companyId: string,
|
||||||
|
) => ReturnType<typeof resolveWorkspaceWithLockMode>;
|
||||||
|
|
||||||
export function createRunDispatch(deps: OrchestrationDependencies) {
|
export function createRunDispatch(deps: OrchestrationDependencies) {
|
||||||
const {
|
const {
|
||||||
db,
|
db,
|
||||||
@@ -20,13 +35,8 @@ export function createRunDispatch(deps: OrchestrationDependencies) {
|
|||||||
evaluatePreRunSessionRotation,
|
evaluatePreRunSessionRotation,
|
||||||
} = deps;
|
} = deps;
|
||||||
|
|
||||||
function resolveWorkspace(
|
const resolveWorkspace: ResolveWorkspaceFn = (task, agent, companyId) =>
|
||||||
task: { workspaceId: string | null; projectId?: string | null } | null,
|
resolveWorkspaceWithLockMode(db, task, agent, companyId);
|
||||||
agent: { workspaceId: string | null },
|
|
||||||
companyId: string,
|
|
||||||
) {
|
|
||||||
return resolveWorkspaceWithLockMode(db, task, agent, companyId);
|
|
||||||
}
|
|
||||||
|
|
||||||
function getRunEligibilityFailure(
|
function getRunEligibilityFailure(
|
||||||
agent: { status: string; budgetMonthlyUsd: number; spentMonthlyUsd: number; name: string; connectorId: string; id: string; companyId: string; model?: string | null },
|
agent: { status: string; budgetMonthlyUsd: number; spentMonthlyUsd: number; name: string; connectorId: string; id: string; companyId: string; model?: string | null },
|
||||||
|
|||||||
@@ -1,14 +1,15 @@
|
|||||||
import { randomUUID } from "node:crypto";
|
import { randomUUID } from "node:crypto";
|
||||||
import type { AppDatabase } from "../database";
|
|
||||||
import type { AgentRecord, ConnectorRecord } from "@shared/types";
|
import type { AgentRecord, ConnectorRecord } from "@shared/types";
|
||||||
import type { DesktopResult } from "@shared/contracts";
|
import type { DesktopResult } from "@shared/contracts";
|
||||||
import type { OrchestrationDependencies } from "./types";
|
import type { OrchestrationDependencies } from "./types";
|
||||||
import type { RunDispatchParams } from "./turn-context";
|
import type { RunDispatchParams } from "./turn-context";
|
||||||
|
import type { ResolveWorkspaceFn } from "./run-dispatch";
|
||||||
|
import { WorkspaceConfigurationError } from "./resolve-workspace";
|
||||||
import { pickModelForAgent } from "../model-dispatch";
|
import { pickModelForAgent } from "../model-dispatch";
|
||||||
|
|
||||||
export function createTaskRunner(
|
export function createTaskRunner(
|
||||||
deps: OrchestrationDependencies,
|
deps: OrchestrationDependencies,
|
||||||
resolveWorkspace: (task: { workspaceId: string | null; projectId?: string | null } | null, agent: { workspaceId: string | null }, companyId: string) => { workspace: ReturnType<AppDatabase["workspacesMod"]["getWorkspace"]>; lockMode: "exclusive" | "shared" | "none"; source: string },
|
resolveWorkspace: ResolveWorkspaceFn,
|
||||||
getRunEligibilityFailure: (agent: AgentRecord, connector: ConnectorRecord, projectId?: string | null) => string | null,
|
getRunEligibilityFailure: (agent: AgentRecord, connector: ConnectorRecord, projectId?: string | null) => string | null,
|
||||||
dispatchRunToWorker: (params: RunDispatchParams) => void,
|
dispatchRunToWorker: (params: RunDispatchParams) => void,
|
||||||
hasActiveRun: (filter: { taskId?: string; agentId?: string; companyId: string }) => boolean,
|
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.");
|
return fail("CHECKOUT_RACE_LOST", "Another run claimed this task. Refresh and try again.");
|
||||||
}
|
}
|
||||||
|
|
||||||
const workspaceResolution = resolveWorkspace(task, agent, task.companyId);
|
let workspaceResolution: ReturnType<typeof resolveWorkspace>;
|
||||||
|
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;
|
const workspace = workspaceResolution.workspace;
|
||||||
if (workspace && workspace.companyId !== task.companyId) {
|
if (workspace && workspace.companyId !== task.companyId) {
|
||||||
return fail("WORKSPACE_COMPANY_MISMATCH", "The selected workspace does not belong to the task company.");
|
return fail("WORKSPACE_COMPANY_MISMATCH", "The selected workspace does not belong to the task company.");
|
||||||
|
|||||||
@@ -1,5 +1,8 @@
|
|||||||
import type { WorkerServiceDependencies } from "./types";
|
import type { WorkerServiceDependencies } from "./types";
|
||||||
import { resolveWorkspaceWithLockMode } from "../../orchestration/resolve-workspace";
|
import {
|
||||||
|
resolveWorkspaceWithLockMode,
|
||||||
|
WorkspaceConfigurationError,
|
||||||
|
} from "../../orchestration/resolve-workspace";
|
||||||
|
|
||||||
export function handleWorkerExit(
|
export function handleWorkerExit(
|
||||||
code: number | null,
|
code: number | null,
|
||||||
|
|||||||
Reference in new issue
Block a user