fix(runtime): fail runs on context input errors
This commit is contained in:
1 parent
43583ef30b
commit
33c6ede381
5 files changed
+317
-100
No files matched your search
@@ -4,6 +4,7 @@ import type { OrchestrationDependencies } from "./types";
|
||||
import { buildRunPayload, type RunDispatchParams } from "./turn-context";
|
||||
import { validateConnectorModelSelection } from "../connector-model-policy";
|
||||
import { resolveWorkspaceWithLockMode } from "./resolve-workspace";
|
||||
import { AppError } from "../errors/index.js";
|
||||
|
||||
type ResolveWorkspaceTaskInput =
|
||||
| (Pick<TaskRecord, "workspaceId" | "projectId"> & { requiresFilesystem?: boolean })
|
||||
@@ -24,6 +25,7 @@ export function createRunDispatch(deps: OrchestrationDependencies) {
|
||||
const {
|
||||
db,
|
||||
logger,
|
||||
emitEvent,
|
||||
isQuitting,
|
||||
hasWorkerReady,
|
||||
getApiPort,
|
||||
@@ -38,6 +40,47 @@ export function createRunDispatch(deps: OrchestrationDependencies) {
|
||||
const resolveWorkspace: ResolveWorkspaceFn = (task, agent, companyId) =>
|
||||
resolveWorkspaceWithLockMode(db, task, agent, companyId);
|
||||
|
||||
function formatDispatchError(error: unknown): string {
|
||||
if (error instanceof AppError) {
|
||||
return `[${error.code}] ${error.message}`;
|
||||
}
|
||||
if (error instanceof Error) {
|
||||
return error.message;
|
||||
}
|
||||
return String(error);
|
||||
}
|
||||
|
||||
function failRunBeforeWorkerEnqueue(
|
||||
params: RunDispatchParams,
|
||||
error: unknown,
|
||||
workspaceLockClaimed: boolean,
|
||||
): void {
|
||||
const errorMessage = formatDispatchError(error);
|
||||
logger.error(`[run] Dispatch failed before worker enqueue for ${params.runId}: ${errorMessage}`);
|
||||
db.runs.finishRun({
|
||||
runId: params.runId,
|
||||
status: "failed",
|
||||
summary: "Run failed before worker enqueue.",
|
||||
errorMessage,
|
||||
exitCode: null,
|
||||
signal: null,
|
||||
model: params.resolvedModel ?? null,
|
||||
unattributedCost: true,
|
||||
});
|
||||
if (workspaceLockClaimed && params.workspaceMode === "exclusive" && params.workspace) {
|
||||
db.workspacesMod.releaseWorkspaceLock(params.workspace.id, params.runId);
|
||||
}
|
||||
db.activityMod.addActivity({
|
||||
companyId: params.companyId,
|
||||
actor: params.agent.id,
|
||||
action: "run.failed",
|
||||
entityType: "run",
|
||||
entityId: params.runId,
|
||||
detail: errorMessage,
|
||||
});
|
||||
emitEvent({ type: "run-updated", runId: params.runId, action: "completed" });
|
||||
}
|
||||
|
||||
function getRunEligibilityFailure(
|
||||
agent: { status: string; budgetMonthlyUsd: number; spentMonthlyUsd: number; name: string; connectorId: string; id: string; companyId: string; model?: string | null },
|
||||
connector: Pick<ConnectorRecord, "id" | "label" | "status" | "capabilityMatrix">,
|
||||
@@ -74,57 +117,62 @@ export function createRunDispatch(deps: OrchestrationDependencies) {
|
||||
}
|
||||
|
||||
function dispatchRunToWorker(params: RunDispatchParams) {
|
||||
if (isQuitting()) {
|
||||
throw new Error("Application is shutting down — refusing to dispatch new run.");
|
||||
}
|
||||
if (!hasWorkerReady()) {
|
||||
throw new Error("Orchestration worker is not ready.");
|
||||
}
|
||||
const connectorIssue = getConnectorExecutionReadinessIssue(params.connector);
|
||||
if (connectorIssue) {
|
||||
throw new Error(connectorIssue);
|
||||
}
|
||||
let workspaceLockClaimed = false;
|
||||
try {
|
||||
if (isQuitting()) {
|
||||
throw new Error("Application is shutting down — refusing to dispatch new run.");
|
||||
}
|
||||
if (!hasWorkerReady()) {
|
||||
throw new Error("Orchestration worker is not ready.");
|
||||
}
|
||||
const connectorIssue = getConnectorExecutionReadinessIssue(params.connector);
|
||||
if (connectorIssue) {
|
||||
throw new Error(connectorIssue);
|
||||
}
|
||||
|
||||
// D-02: Claim the workspace lock synchronously before enqueuing the run.
|
||||
// Previously the worker claimed the lock only after it emitted `running`,
|
||||
// so two runs could race between enqueue and startup and both read the
|
||||
// same workspace mid-write. Main is the authoritative coordinator.
|
||||
//
|
||||
// On contention we mark the run `cancelled`, not `failed`. The Dashboard
|
||||
// "Needs attention" card only surfaces failed/timed_out/interrupted runs
|
||||
// (see src/renderer/components/dashboard/Dashboard.tsx), so the user is
|
||||
// not alarmed by a transient backoff. The task row is rolled back to
|
||||
// `todo` via the `finishRun` never-started branch, so heartbeat picks
|
||||
// it up on the next cycle with no manual intervention.
|
||||
if (
|
||||
params.workspaceMode === "exclusive" &&
|
||||
params.workspace &&
|
||||
!db.workspacesMod.claimWorkspaceLock(params.workspace.id, params.runId)
|
||||
) {
|
||||
db.runs.finishRun({
|
||||
runId: params.runId,
|
||||
status: "cancelled",
|
||||
summary: null,
|
||||
errorMessage: `Workspace ${params.workspace.id} is in use by another run; will retry on the next heartbeat.`,
|
||||
exitCode: null,
|
||||
signal: null,
|
||||
model: null,
|
||||
unattributedCost: true,
|
||||
// D-02: Claim the workspace lock synchronously before enqueuing the run.
|
||||
// Previously the worker claimed the lock only after it emitted `running`,
|
||||
// so two runs could race between enqueue and startup and both read the
|
||||
// same workspace mid-write. Main is the authoritative coordinator.
|
||||
//
|
||||
// On contention we mark the run `cancelled`, not `failed`. The Dashboard
|
||||
// "Needs attention" card only surfaces failed/timed_out/interrupted runs
|
||||
// (see src/renderer/components/dashboard/Dashboard.tsx), so the user is
|
||||
// not alarmed by a transient backoff. The task row is rolled back to
|
||||
// `todo` via the `finishRun` never-started branch, so heartbeat picks
|
||||
// it up on the next cycle with no manual intervention.
|
||||
if (params.workspaceMode === "exclusive" && params.workspace) {
|
||||
if (!db.workspacesMod.claimWorkspaceLock(params.workspace.id, params.runId)) {
|
||||
db.runs.finishRun({
|
||||
runId: params.runId,
|
||||
status: "cancelled",
|
||||
summary: null,
|
||||
errorMessage: `Workspace ${params.workspace.id} is in use by another run; will retry on the next heartbeat.`,
|
||||
exitCode: null,
|
||||
signal: null,
|
||||
model: null,
|
||||
unattributedCost: true,
|
||||
});
|
||||
emitEvent({ type: "run-updated", runId: params.runId, action: "completed" });
|
||||
return;
|
||||
}
|
||||
workspaceLockClaimed = true;
|
||||
}
|
||||
|
||||
const payload = buildRunPayload(params, {
|
||||
db,
|
||||
logger,
|
||||
getApiPort,
|
||||
createRunToken,
|
||||
listRuntimeSkillEntries,
|
||||
getSessionHandoff,
|
||||
evaluatePreRunSessionRotation,
|
||||
});
|
||||
return;
|
||||
|
||||
enqueueWorkerRun(payload);
|
||||
} catch (error) {
|
||||
failRunBeforeWorkerEnqueue(params, error, workspaceLockClaimed);
|
||||
}
|
||||
|
||||
const payload = buildRunPayload(params, {
|
||||
db,
|
||||
logger,
|
||||
getApiPort,
|
||||
createRunToken,
|
||||
listRuntimeSkillEntries,
|
||||
getSessionHandoff,
|
||||
evaluatePreRunSessionRotation,
|
||||
});
|
||||
|
||||
enqueueWorkerRun(payload);
|
||||
}
|
||||
|
||||
return { resolveWorkspace, getRunEligibilityFailure, dispatchRunToWorker };
|
||||
|
||||
@@ -127,15 +127,18 @@ export function createTaskRunner(
|
||||
trigger,
|
||||
resolvedModel: assignment.modelId,
|
||||
});
|
||||
db.activityMod.addActivity({
|
||||
companyId: task.companyId,
|
||||
actor,
|
||||
action: "run.queued",
|
||||
entityType: "run",
|
||||
entityId: run.id,
|
||||
detail: task.title,
|
||||
});
|
||||
emitEvent({ type: "run-updated", runId: run.id, action: "created" });
|
||||
const postDispatchRun = db.runs.getRun(run.id);
|
||||
if (postDispatchRun.status === "queued" || postDispatchRun.status === "running") {
|
||||
db.activityMod.addActivity({
|
||||
companyId: task.companyId,
|
||||
actor,
|
||||
action: "run.queued",
|
||||
entityType: "run",
|
||||
entityId: run.id,
|
||||
detail: task.title,
|
||||
});
|
||||
emitEvent({ type: "run-updated", runId: run.id, action: "created" });
|
||||
}
|
||||
return ok(run.id);
|
||||
}
|
||||
|
||||
|
||||
@@ -54,7 +54,7 @@ export interface RunDispatchParams {
|
||||
|
||||
export interface TurnContextDeps {
|
||||
db: AppDatabase;
|
||||
logger: { info: (...args: unknown[]) => void; warn: (...args: unknown[]) => void };
|
||||
logger: { info: (...args: unknown[]) => void; warn: (...args: unknown[]) => void; error?: (...args: unknown[]) => void };
|
||||
getApiPort: () => number;
|
||||
createRunToken: (agentId: string, companyId: string, runId: string) => string;
|
||||
listRuntimeSkillEntries: (companyId: string, agentId: string, connectorId: string) => ProjectedSkill[];
|
||||
@@ -123,13 +123,13 @@ function resolveSessionHandoff(
|
||||
try {
|
||||
return deps.getSessionHandoff(agentId);
|
||||
} catch (err) {
|
||||
logAppError(deps.logger, err, {
|
||||
kind: "recoverable",
|
||||
const appError = logAppError(deps.logger, err, {
|
||||
kind: "upstream",
|
||||
code: "run.session_handoff.load_failed",
|
||||
message: `Failed to load session handoff for agent ${agentId}; continuing without handoff`,
|
||||
message: `Failed to load session handoff for agent ${agentId}.`,
|
||||
context: { agentId },
|
||||
}, "warn");
|
||||
return null;
|
||||
});
|
||||
throw appError;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -143,13 +143,13 @@ function resolveSkills(
|
||||
try {
|
||||
return deps.listRuntimeSkillEntries(companyId, agentId, connectorId);
|
||||
} catch (err) {
|
||||
logAppError(deps.logger, err, {
|
||||
kind: "recoverable",
|
||||
const appError = logAppError(deps.logger, err, {
|
||||
kind: "upstream",
|
||||
code: "run.skills.fetch_failed",
|
||||
message: `Failed to fetch managed skills for agent ${agentName}; proceeding with empty skill set`,
|
||||
message: `Failed to fetch managed skills for agent ${agentName}.`,
|
||||
context: { agentId, agentName, connectorId, companyId },
|
||||
}, "warn");
|
||||
return [];
|
||||
});
|
||||
throw appError;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in new issue
Block a user