diff --git a/src/main/orchestration/run-dispatch.ts b/src/main/orchestration/run-dispatch.ts index fe558ca7..9baa9681 100644 --- a/src/main/orchestration/run-dispatch.ts +++ b/src/main/orchestration/run-dispatch.ts @@ -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 & { 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, @@ -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 }; diff --git a/src/main/orchestration/task-runner.ts b/src/main/orchestration/task-runner.ts index 9750f374..3863cdb2 100644 --- a/src/main/orchestration/task-runner.ts +++ b/src/main/orchestration/task-runner.ts @@ -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); } diff --git a/src/main/orchestration/turn-context.ts b/src/main/orchestration/turn-context.ts index d787c2ae..d7b27f76 100644 --- a/src/main/orchestration/turn-context.ts +++ b/src/main/orchestration/turn-context.ts @@ -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; } } diff --git a/src/main/service-graph/create-orchestration-service.ts b/src/main/service-graph/create-orchestration-service.ts index 728677f7..7438a980 100644 --- a/src/main/service-graph/create-orchestration-service.ts +++ b/src/main/service-graph/create-orchestration-service.ts @@ -73,41 +73,23 @@ export function buildOrchestrationService({ workerService.getWorker()?.postMessage({ type: "enqueue-run", payload }); }, isQuitting: () => isQuittingRef.current, - listRuntimeSkillEntries: (companyId, agentId, connectorId) => { - try { - return skillService.listRuntimeSkillEntries(companyId, agentId, connectorId); - } catch (error) { - logger.warn(`[service-graph] listRuntimeSkillEntries failed for agent ${agentId}:`, error); - return []; - } - }, + listRuntimeSkillEntries: (companyId, agentId, connectorId) => + skillService.listRuntimeSkillEntries(companyId, agentId, connectorId), budgetService, checkpointEngine, - getSessionHandoff: (agentId) => { - try { - return sessionService.getHandoffForAgent(agentId); - } catch (error) { - logger.warn(`[service-graph] getSessionHandoff failed for agent ${agentId}:`, error); - return null; - } - }, - evaluatePreRunSessionRotation: (agentId, connectorId) => { - try { - const agent = db.entityCrud.getAgent(agentId); - if (!agent.sessionId) return { rotated: false, handoff: null, sessionId: null }; - const session = db.tasks.getSessionById(agent.sessionId); - if (!session) return { rotated: false, handoff: null, sessionId: agent.sessionId }; - const result = sessionService.evaluateRotation(session); - if (result.shouldRotate) { - const newSessionId = sessionService.rotateSession(session); - const handoff = sessionService.getHandoffForAgent(agentId); - return { rotated: true, handoff, sessionId: newSessionId }; - } - return { rotated: false, handoff: null, sessionId: agent.sessionId }; - } catch (error) { - logger.warn(`[service-graph] evaluatePreRunSessionRotation failed for agent ${agentId}:`, error); - return { rotated: false, handoff: null, sessionId: null }; + getSessionHandoff: (agentId) => sessionService.getHandoffForAgent(agentId), + evaluatePreRunSessionRotation: (agentId, _connectorId) => { + const agent = db.entityCrud.getAgent(agentId); + if (!agent.sessionId) return { rotated: false, handoff: null, sessionId: null }; + const session = db.tasks.getSessionById(agent.sessionId); + if (!session) return { rotated: false, handoff: null, sessionId: agent.sessionId }; + const result = sessionService.evaluateRotation(session); + if (result.shouldRotate) { + const newSessionId = sessionService.rotateSession(session); + const handoff = sessionService.getHandoffForAgent(agentId); + return { rotated: true, handoff, sessionId: newSessionId }; } + return { rotated: false, handoff: null, sessionId: agent.sessionId }; }, }); } diff --git a/tests/unit/run-context-input-failures.test.ts b/tests/unit/run-context-input-failures.test.ts new file mode 100644 index 00000000..4692d26d --- /dev/null +++ b/tests/unit/run-context-input-failures.test.ts @@ -0,0 +1,184 @@ +import { join } from "node:path"; +import { tmpdir } from "node:os"; +import { mkdtemp, rm } from "node:fs/promises"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +vi.mock("electron", () => ({ + safeStorage: { + isEncryptionAvailable: () => false, + getSelectedStorageBackend: () => "basic_text", + encryptString: (value: string) => Buffer.from(value, "utf8"), + decryptString: (value: Buffer) => value.toString("utf8"), + }, +})); + +import { AppDatabase } from "../../src/main/database"; +import { createBudgetEnforcementService } from "../../src/main/budget-enforcement-service"; +import { createRunDispatch } from "../../src/main/orchestration/run-dispatch"; +import { createTaskRunner } from "../../src/main/orchestration/task-runner"; +import type { OrchestrationDependencies } from "../../src/main/orchestration/types"; + +function createLogger() { + return { + debug: vi.fn(), + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + }; +} + +function createCompanyRunFixture(db: AppDatabase, profileDir: string) { + db.connectorsCrud.updateConnectorHealth({ + id: "openai-codex", + status: "ready", + authState: "authenticated", + version: "test", + lastError: null, + }); + const companyId = db.company.saveCompany({ + name: "Context Integrity Co", + description: "Validates run context failures", + status: "active", + }); + const workspaceId = db.workspacesMod.saveWorkspace({ + companyId, + name: "Primary", + localPath: profileDir, + repoUrl: "", + repoRef: "", + isPrimary: true, + }); + const agentId = db.agents.saveAgent({ + companyId, + name: "Runtime Engineer", + role: "Engineer", + title: "Runtime Engineer", + status: "idle", + reportsTo: null, + connectorId: "openai-codex", + workspaceId, + model: "gpt-5.4", + capabilities: "Ships runtime fixes", + budgetMonthlyUsd: 250, + }); + const taskId = db.tasks.saveTask({ + companyId, + title: "Assemble safe run context", + description: "The worker must start only with a complete context.", + assigneeAgentId: agentId, + workspaceId, + priority: "high", + status: "todo", + requiresFilesystem: true, + }); + return { companyId, workspaceId, agentId, taskId }; +} + +function createHarness( + db: AppDatabase, + overrides: Partial> = {}, +) { + const logger = createLogger(); + const enqueueWorkerRun = vi.fn(); + const emitEvent = vi.fn(); + const budgetService = createBudgetEnforcementService({ + db, + logger, + notify: vi.fn(), + publishDomainChanged: vi.fn(), + }); + const deps: OrchestrationDependencies = { + db, + logger, + publishDomainChanged: vi.fn(), + emitEvent, + notify: vi.fn(), + createRunToken: () => "test-token", + getApiPort: () => 3210, + dispatchAutomation: vi.fn(), + hasWorkerReady: () => true, + enqueueWorkerRun, + isQuitting: () => false, + listRuntimeSkillEntries: () => [], + budgetService, + checkpointEngine: { shouldPauseHeartbeat: () => false }, + ...overrides, + }; + const runDispatch = createRunDispatch(deps); + const taskRunner = createTaskRunner( + deps, + runDispatch.resolveWorkspace, + runDispatch.getRunEligibilityFailure, + runDispatch.dispatchRunToWorker, + (filter) => db.runs.hasActiveRun(filter), + ); + return { enqueueWorkerRun, emitEvent, logger, taskRunner }; +} + +describe("run context input failures", () => { + let tempDir: string; + let db: AppDatabase; + + beforeEach(async () => { + tempDir = await mkdtemp(join(tmpdir(), "agentcompany-context-input-")); + db = new AppDatabase(tempDir); + db.init(); + }); + + afterEach(async () => { + db.close(); + await rm(tempDir, { recursive: true, force: true }); + }); + + it("fails the run before worker enqueue when managed skills cannot load", () => { + const { taskId } = createCompanyRunFixture(db, tempDir); + const { enqueueWorkerRun, emitEvent, taskRunner } = createHarness(db, { + listRuntimeSkillEntries: () => { + throw new Error("skill database unavailable"); + }, + }); + + const result = taskRunner.queueTaskRun(taskId, "manual", "unit-test"); + + expect(result.ok).toBe(true); + if (!result.ok) { + throw new Error("Expected the task runner to persist a failed run."); + } + expect(enqueueWorkerRun).not.toHaveBeenCalled(); + const run = db.runs.getRun(result.data); + const task = db.entityCrud.getTask(taskId); + const workspace = db.workspacesMod.getWorkspace(task.workspaceId); + expect(run.status).toBe("failed"); + expect(run.errorMessage).toContain("run.skills.fetch_failed"); + expect(task.status).toBe("todo"); + expect(task.activeRunId).toBeNull(); + expect(workspace?.state).toBe("ready"); + expect(emitEvent).toHaveBeenCalledWith({ type: "run-updated", runId: run.id, action: "completed" }); + expect(emitEvent).not.toHaveBeenCalledWith({ type: "run-updated", runId: run.id, action: "created" }); + }); + + it("fails the run before worker enqueue when session handoff cannot load", () => { + const { taskId } = createCompanyRunFixture(db, tempDir); + const { enqueueWorkerRun, emitEvent, taskRunner } = createHarness(db, { + getSessionHandoff: () => { + throw new Error("session handoff store unavailable"); + }, + }); + + const result = taskRunner.queueTaskRun(taskId, "manual", "unit-test"); + + expect(result.ok).toBe(true); + if (!result.ok) { + throw new Error("Expected the task runner to persist a failed run."); + } + expect(enqueueWorkerRun).not.toHaveBeenCalled(); + const run = db.runs.getRun(result.data); + const task = db.entityCrud.getTask(taskId); + expect(run.status).toBe("failed"); + expect(run.errorMessage).toContain("run.session_handoff.load_failed"); + expect(task.status).toBe("todo"); + expect(task.activeRunId).toBeNull(); + expect(emitEvent).toHaveBeenCalledWith({ type: "run-updated", runId: run.id, action: "completed" }); + expect(emitEvent).not.toHaveBeenCalledWith({ type: "run-updated", runId: run.id, action: "created" }); + }); +});