diff --git a/landing/components/release-links.generated.ts b/landing/components/release-links.generated.ts index ded3eb90..5d7f1c3d 100644 --- a/landing/components/release-links.generated.ts +++ b/landing/components/release-links.generated.ts @@ -16,7 +16,7 @@ export type PlatformDownload = { assets: PlatformDownloadAsset[]; }; -export const releaseVersion = "0.4.180"; +export const releaseVersion = "0.4.181"; export const platformDownloads: PlatformDownload[] = [ { @@ -29,14 +29,14 @@ export const platformDownloads: PlatformDownload[] = [ architectures: ["arm64"], id: "macos-arm64-dmg", label: "Apple Silicon DMG", - href: "https://pub-a19a3a2c1d2e4fa3a9c120b184024448.r2.dev/AC/AgentCompany-0.4.180-arm64.dmg", + href: "https://pub-a19a3a2c1d2e4fa3a9c120b184024448.r2.dev/AC/AgentCompany-0.4.181-arm64.dmg", note: "arm64", }, { architectures: ["x64"], id: "macos-x64-dmg", label: "Intel Mac DMG", - href: "https://pub-a19a3a2c1d2e4fa3a9c120b184024448.r2.dev/AC/AgentCompany-0.4.180.dmg", + href: "https://pub-a19a3a2c1d2e4fa3a9c120b184024448.r2.dev/AC/AgentCompany-0.4.181.dmg", note: "x64", }, ], @@ -51,7 +51,7 @@ export const platformDownloads: PlatformDownload[] = [ architectures: ["arm64", "x64"], id: "windows-universal-installer", label: "Windows Universal Installer", - href: "https://pub-a19a3a2c1d2e4fa3a9c120b184024448.r2.dev/AC/AgentCompany.Setup.0.4.180.exe", + href: "https://pub-a19a3a2c1d2e4fa3a9c120b184024448.r2.dev/AC/AgentCompany.Setup.0.4.181.exe", note: "arm64 + x64", }, ], @@ -66,42 +66,42 @@ export const platformDownloads: PlatformDownload[] = [ architectures: ["x64"], id: "linux-x64-appimage", label: "Linux x64 AppImage", - href: "https://pub-a19a3a2c1d2e4fa3a9c120b184024448.r2.dev/AC/AgentCompany-0.4.180.AppImage", + href: "https://pub-a19a3a2c1d2e4fa3a9c120b184024448.r2.dev/AC/AgentCompany-0.4.181.AppImage", note: "x64", }, { architectures: ["x64"], id: "linux-x64-deb", label: "Linux x64 DEB", - href: "https://pub-a19a3a2c1d2e4fa3a9c120b184024448.r2.dev/AC/agent-company_0.4.180_amd64.deb", + href: "https://pub-a19a3a2c1d2e4fa3a9c120b184024448.r2.dev/AC/agent-company_0.4.181_amd64.deb", note: "x64", }, { architectures: ["x64"], id: "linux-x64-rpm", label: "Linux x64 RPM", - href: "https://pub-a19a3a2c1d2e4fa3a9c120b184024448.r2.dev/AC/agent-company-0.4.180.x86_64.rpm", + href: "https://pub-a19a3a2c1d2e4fa3a9c120b184024448.r2.dev/AC/agent-company-0.4.181.x86_64.rpm", note: "x64", }, { architectures: ["arm64"], id: "linux-arm64-appimage", label: "Linux ARM64 AppImage", - href: "https://pub-a19a3a2c1d2e4fa3a9c120b184024448.r2.dev/AC/AgentCompany-0.4.180-arm64.AppImage", + href: "https://pub-a19a3a2c1d2e4fa3a9c120b184024448.r2.dev/AC/AgentCompany-0.4.181-arm64.AppImage", note: "arm64", }, { architectures: ["arm64"], id: "linux-arm64-deb", label: "Linux ARM64 DEB", - href: "https://pub-a19a3a2c1d2e4fa3a9c120b184024448.r2.dev/AC/agent-company_0.4.180_arm64.deb", + href: "https://pub-a19a3a2c1d2e4fa3a9c120b184024448.r2.dev/AC/agent-company_0.4.181_arm64.deb", note: "arm64", }, { architectures: ["arm64"], id: "linux-arm64-rpm", label: "Linux ARM64 RPM", - href: "https://pub-a19a3a2c1d2e4fa3a9c120b184024448.r2.dev/AC/agent-company-0.4.180.aarch64.rpm", + href: "https://pub-a19a3a2c1d2e4fa3a9c120b184024448.r2.dev/AC/agent-company-0.4.181.aarch64.rpm", note: "arm64", }, ], diff --git a/package.json b/package.json index 1d837390..8e66d19e 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "agent-company", - "version": "0.4.180", + "version": "0.4.181", "private": true, "type": "module", "description": "Cross-platform Electron desktop control plane for local-first AI agent companies.", diff --git a/src/main/automation/automation-approval-handler.ts b/src/main/automation/automation-approval-handler.ts index 0506ea7f..b779ec69 100644 --- a/src/main/automation/automation-approval-handler.ts +++ b/src/main/automation/automation-approval-handler.ts @@ -83,6 +83,13 @@ export function handleApprovalResolved( if (deps.isQuitting()) return; const { db, logger, publishDomainChanged, notify, wakeAgentIfPossible, queueBrowserActionExecution } = deps; const { dispatchAutomation, handleTaskStatusChange, runWorkflowPipeline, trackTimer } = callbacks; + const explicitlyWokenAgentIds = new Set(); + + function wakeAgent(agentId: string | null | undefined, trigger: string): void { + if (!agentId) return; + wakeAgentIfPossible(agentId, companyId, trigger); + explicitlyWokenAgentIds.add(agentId); + } const snapshot = db.projection.snapshot.listSnapshot(); const approval = snapshot.approvals.find((entry) => entry.id === approvalId); @@ -134,11 +141,11 @@ export function handleApprovalResolved( companyId, agentId: agent.id, }); - wakeAgentIfPossible(agent.id, companyId, "assignment"); + wakeAgent(agent.id, "assignment"); const ceo = db.core.agents.getCeoForCompany(companyId); if (ceo && ceo.id !== agent.id) { - wakeAgentIfPossible(ceo.id, companyId, "assignment"); + wakeAgent(ceo.id, "assignment"); } publishDomainChanged(); @@ -186,20 +193,15 @@ export function handleApprovalResolved( db.core.tasks.advanceTaskStatus(task.id, "done", "Board approved deliverable."); handleTaskStatusChange(task.id, companyId, task.status, "done"); db.core.activity.addActivity({ companyId, actor: "board", action: "deliverable.approved", entityType: "task", entityId: task.id, detail: approval.decisionNote || "Deliverable approved by board." }); - } else if (decision === "rejected") { + } else if (decision === "rejected" || decision === "revision_requested") { + const previousStatus = task.status; + const statusMessage = decision === "rejected" + ? `Board rejected deliverable: ${approval.decisionNote || "No reason given."}` + : `Board requested revisions: ${approval.decisionNote || "No reason given."}`; recordReviewFeedbackKnowledge(db, approval, decision); - db.core.tasks.advanceTaskStatus(task.id, "in_progress", `Board rejected deliverable: ${approval.decisionNote || "No reason given."}`); - handleTaskStatusChange(task.id, companyId, task.status, "in_progress"); - if (task.assigneeAgentId) { - wakeAgentIfPossible(task.assigneeAgentId, companyId, "revision_needed"); - } - } else if (decision === "revision_requested") { - db.core.tasks.advanceTaskStatus(task.id, "in_progress", `Board requested revisions: ${approval.decisionNote || "No reason given."}`); - handleTaskStatusChange(task.id, companyId, task.status, "in_progress"); - recordReviewFeedbackKnowledge(db, approval, decision); - if (task.assigneeAgentId) { - wakeAgentIfPossible(task.assigneeAgentId, companyId, "revision_needed"); - } + db.core.tasks.advanceTaskStatus(task.id, "todo", statusMessage); + handleTaskStatusChange(task.id, companyId, previousStatus, "todo"); + wakeAgent(task.assigneeAgentId, "revision_needed"); } } else if (approval.requestedByAgentId) { const agent = db.core.entity.getAgent(approval.requestedByAgentId); @@ -208,7 +210,7 @@ export function handleApprovalResolved( logger.info(`[deliverable-review] Resumed heartbeat for ${agent.name} after user ${decision}.`); } if (decision !== "approved") { - wakeAgentIfPossible(approval.requestedByAgentId, companyId, "revision_needed"); + wakeAgent(approval.requestedByAgentId, "revision_needed"); } } @@ -366,6 +368,8 @@ export function handleApprovalResolved( } if (approval.requestedByAgentId) { - wakeAgentIfPossible(approval.requestedByAgentId, companyId, "approval_resolved"); + if (!explicitlyWokenAgentIds.has(approval.requestedByAgentId)) { + wakeAgentIfPossible(approval.requestedByAgentId, companyId, "approval_resolved"); + } } } diff --git a/src/main/database/entity-crud.ts b/src/main/database/entity-crud.ts index 7451831d..9320a93b 100644 --- a/src/main/database/entity-crud.ts +++ b/src/main/database/entity-crud.ts @@ -1,7 +1,7 @@ import type { DatabaseSync } from "node:sqlite"; import type { AgentRecord, ConnectorId, TaskRecord } from "@shared/types"; import type { ReadAll, Transaction } from "./helpers"; -import { ensureId, normalizeOptionalId, nowIso } from "./helpers"; +import { activeRunIdSql, ensureId, normalizeOptionalId, nowIso } from "./helpers"; /** * Goal row shape returned by listGoals / getGoalForUpdate. Mirrors the @@ -53,7 +53,7 @@ export function buildEntityCrudMethods( getTask(id: string): TaskRecord { const row = getDb() .prepare( - "select id, company_id as companyId, project_id as projectId, goal_id as goalId, parent_id as parentId, title, description, assignee_agent_id as assigneeAgentId, workspace_id as workspaceId, status, priority, (select l.run_id from task_leases l where l.task_id = tasks.id) as activeRunId, cost_usd as costUsd, task_type as taskType, requires_user_review as requiresUserReview, requires_filesystem as requiresFilesystem, started_at as startedAt, completed_at as completedAt, cancelled_at as cancelledAt, created_at as createdAt, updated_at as updatedAt from tasks where id = ?", + `select id, company_id as companyId, project_id as projectId, goal_id as goalId, parent_id as parentId, title, description, assignee_agent_id as assigneeAgentId, workspace_id as workspaceId, status, priority, ${activeRunIdSql("tasks")} as activeRunId, cost_usd as costUsd, task_type as taskType, requires_user_review as requiresUserReview, requires_filesystem as requiresFilesystem, started_at as startedAt, completed_at as completedAt, cancelled_at as cancelledAt, created_at as createdAt, updated_at as updatedAt from tasks where id = ?`, ) .get(id) as TaskRecord | undefined; if (!row) throw new Error(`Task not found: ${id}`); diff --git a/src/main/database/helpers.ts b/src/main/database/helpers.ts index 3a4ddfc4..80f4c21c 100644 --- a/src/main/database/helpers.ts +++ b/src/main/database/helpers.ts @@ -42,3 +42,12 @@ export function taskPriorityOrderSql(columnName: string): string { else 0 end`; } + +export function activeRunIdSql(taskTableAlias: string): string { + return `(select l.run_id + from task_leases l + join runs r on r.id = l.run_id + where l.task_id = ${taskTableAlias}.id + and r.status in ('queued', 'running') + limit 1)`; +} diff --git a/src/main/database/runs.ts b/src/main/database/runs.ts index 45e444b8..d84f362c 100644 --- a/src/main/database/runs.ts +++ b/src/main/database/runs.ts @@ -166,20 +166,20 @@ export function buildRunMethods( model: string | null; unattributedCost: boolean; usageStatus?: RunUsageStatus; + recoverTaskForRetry?: boolean; }) { const run = getDb().prepare("select task_id as taskId, agent_id as agentId, company_id as companyId, connector_id as connectorId, started_at as startedAt from runs where id = ?").get(input.runId) as | { taskId: string | null; agentId: string; companyId: string; connectorId: ConnectorId; startedAt: string | null } | undefined; if (!run) return { budgetHardStopped: false }; - // A run that never transitioned to 'running' (startedAt === null) and - // is now terminal with status 'failed' / 'cancelled' must roll the - // task back to 'todo'. Otherwise the task is stuck in 'in_progress' - // with no owner — checkoutTask refuses (WHERE status IN todo/backlog), - // and the scheduler loops forever logging "lost checkout race". The - // common trigger is workspace-lock contention: `dispatchRunToWorker` - // bails before work starts and calls finishRun('failed'). + // Recoverable terminal paths must roll the task back to 'todo'. + // Otherwise the task is stuck in 'in_progress' with no owner: + // checkoutTask refuses (WHERE status IN todo/backlog), and the + // scheduler loops forever logging "lost checkout race". const runNeverStarted = run.startedAt === null && (input.status === "failed" || input.status === "cancelled"); + const shouldReturnTaskToTodo = + run.taskId !== null && (runNeverStarted || input.recoverTaskForRetry === true); const ts = nowIso(); const usageStatus = input.usageStatus ?? deriveLegacyRunUsageStatus({ runStatus: input.status, @@ -220,7 +220,7 @@ export function buildRunMethods( getDb() .prepare("delete from task_leases where task_id = ? and run_id = ?") .run(run.taskId, input.runId); - if (runNeverStarted) { + if (shouldReturnTaskToTodo) { // Roll the task back to `todo` when no *other* live run claims // it. Invariant: `task.status = 'in_progress'` ⇒ a queued or // running run owns it through `task_leases`. diff --git a/src/main/database/snapshot.ts b/src/main/database/snapshot.ts index 7d2d3810..6dccf599 100644 --- a/src/main/database/snapshot.ts +++ b/src/main/database/snapshot.ts @@ -15,7 +15,7 @@ import type { TaskRecord, WorkspaceRecord, } from "@shared/types"; -import { BOARD_READER_ID, type ReadAll, type SqliteRow, toBoolean } from "./helpers"; +import { activeRunIdSql, BOARD_READER_ID, type ReadAll, type SqliteRow, toBoolean } from "./helpers"; export function buildSnapshotMethods( getDb: () => DatabaseSync, @@ -83,7 +83,7 @@ export function buildSnapshotMethods( ...companyArgs, ), tasks: readAll( - `select id, company_id as companyId, project_id as projectId, goal_id as goalId, parent_id as parentId, title, description, assignee_agent_id as assigneeAgentId, workspace_id as workspaceId, status, priority, (select l.run_id from task_leases l where l.task_id = tasks.id) as activeRunId, cost_usd as costUsd, task_type as taskType, requires_user_review as requiresUserReview, requires_filesystem as requiresFilesystem, started_at as startedAt, completed_at as completedAt, cancelled_at as cancelledAt, created_at as createdAt, updated_at as updatedAt from tasks${companyWhere} order by created_at desc limit 2000`, + `select id, company_id as companyId, project_id as projectId, goal_id as goalId, parent_id as parentId, title, description, assignee_agent_id as assigneeAgentId, workspace_id as workspaceId, status, priority, ${activeRunIdSql("tasks")} as activeRunId, cost_usd as costUsd, task_type as taskType, requires_user_review as requiresUserReview, requires_filesystem as requiresFilesystem, started_at as startedAt, completed_at as completedAt, cancelled_at as cancelledAt, created_at as createdAt, updated_at as updatedAt from tasks${companyWhere} order by created_at desc limit 2000`, ...companyArgs, ).map((record) => ({ ...record, requiresUserReview: toBoolean(record.requiresUserReview), requiresFilesystem: toBoolean(record.requiresFilesystem), startedAt: record.startedAt ?? null, completedAt: record.completedAt ?? null, cancelledAt: record.cancelledAt ?? null })) as TaskRecord[], approvals: readAll( diff --git a/src/main/database/task-queries.ts b/src/main/database/task-queries.ts index 52f6aeab..a1a6838c 100644 --- a/src/main/database/task-queries.ts +++ b/src/main/database/task-queries.ts @@ -1,6 +1,6 @@ import type { DatabaseSync } from "node:sqlite"; import type { TaskRecord } from "@shared/types"; -import { type ReadAll, type Transaction, taskPriorityOrderSql } from "./helpers"; +import { activeRunIdSql, type ReadAll, type Transaction, taskPriorityOrderSql } from "./helpers"; import type { TaskListByStatusesRow } from "./tasks-crud"; /** @@ -25,7 +25,7 @@ export function buildTaskQueryMethods( ): TaskRecord[] { const placeholders = statuses.map(() => "?").join(","); return readAll( - `select id, company_id as companyId, project_id as projectId, goal_id as goalId, parent_id as parentId, title, description, assignee_agent_id as assigneeAgentId, workspace_id as workspaceId, status, priority, (select l.run_id from task_leases l where l.task_id = tasks.id) as activeRunId, cost_usd as costUsd, task_type as taskType, created_at as createdAt, updated_at as updatedAt + `select id, company_id as companyId, project_id as projectId, goal_id as goalId, parent_id as parentId, title, description, assignee_agent_id as assigneeAgentId, workspace_id as workspaceId, status, priority, ${activeRunIdSql("tasks")} as activeRunId, cost_usd as costUsd, task_type as taskType, created_at as createdAt, updated_at as updatedAt from tasks where company_id = ? and assignee_agent_id = ? and status in (${placeholders}) order by @@ -91,7 +91,7 @@ export function buildTaskQueryMethods( ): TaskListByStatusesRow[] { const placeholders = statuses.map(() => "?").join(", "); return readAll( - `select id, company_id as companyId, project_id as projectId, goal_id as goalId, parent_id as parentId, title, description, assignee_agent_id as assigneeAgentId, workspace_id as workspaceId, status, priority, (select l.run_id from task_leases l where l.task_id = tasks.id) as activeRunId, cost_usd as costUsd, requires_user_review as requiresUserReview, started_at as startedAt, completed_at as completedAt, cancelled_at as cancelledAt, created_at as createdAt, updated_at as updatedAt from tasks where company_id = ? and status in (${placeholders}) order by created_at desc`, + `select id, company_id as companyId, project_id as projectId, goal_id as goalId, parent_id as parentId, title, description, assignee_agent_id as assigneeAgentId, workspace_id as workspaceId, status, priority, ${activeRunIdSql("tasks")} as activeRunId, cost_usd as costUsd, requires_user_review as requiresUserReview, started_at as startedAt, completed_at as completedAt, cancelled_at as cancelledAt, created_at as createdAt, updated_at as updatedAt from tasks where company_id = ? and status in (${placeholders}) order by created_at desc`, companyId, ...statuses, ).map((t) => ({ diff --git a/src/main/database/tasks-status.ts b/src/main/database/tasks-status.ts index 468c8ad8..66855d21 100644 --- a/src/main/database/tasks-status.ts +++ b/src/main/database/tasks-status.ts @@ -3,6 +3,7 @@ import type { TaskRecord } from "@shared/types"; import { type ReadAll, type Transaction, + activeRunIdSql, nowIso, taskPriorityOrderSql, } from "./helpers"; @@ -82,7 +83,7 @@ export function buildTaskStatusMethods( t.parent_id as parentId, t.title, t.description, t.assignee_agent_id as assigneeAgentId, t.workspace_id as workspaceId, t.status, t.priority, - (select l.run_id from task_leases l where l.task_id = t.id) as activeRunId, + ${activeRunIdSql("t")} as activeRunId, t.cost_usd as costUsd, t.created_at as createdAt, t.updated_at as updatedAt from tasks t diff --git a/src/main/orchestration/scheduler.ts b/src/main/orchestration/scheduler.ts index 458828b5..d10a9a4c 100644 --- a/src/main/orchestration/scheduler.ts +++ b/src/main/orchestration/scheduler.ts @@ -223,11 +223,12 @@ export function createScheduler( if (hasActiveRun({ agentId: ceo.id, companyId: company.id })) return; const lastHeartbeat = ceo.lastHeartbeatAt ? new Date(ceo.lastHeartbeatAt).getTime() : 0; - const lastAttempt = bootstrapAttemptByAgent.get(ceo.id) ?? 0; + const attemptKey = agentAttemptKey(company.id, ceo.id); + const lastAttempt = bootstrapAttemptByAgent.get(attemptKey) ?? 0; const bootstrapCooldown = 120_000; if (now - Math.max(lastHeartbeat, lastAttempt) <= bootstrapCooldown) return; - bootstrapAttemptByAgent.set(ceo.id, now); + bootstrapAttemptByAgent.set(attemptKey, now); logger.info(`[scheduler] Company ${company.name} has ${goalCount} goal(s) and no advancing tasks; bootstrapping CEO ${ceo.name}`); wakeAgentIfPossible(ceo.id, company.id, "bootstrap"); } @@ -243,6 +244,8 @@ export function createScheduler( const wokenAgentIds = new Set(); const pendingWakeupAgents = db.automation.wakeup.listPendingWakeupAgents(companyId); for (const { agentId } of pendingWakeupAgents) { + // Queued wakeups are explicit durable events; they bypass the periodic + // heartbeat_enabled scan and let wakeAgentIfPossible own re-enabling. if (hasActiveRun({ agentId, companyId })) continue; if (deps.checkpointEngine?.shouldPauseHeartbeat(companyId, agentId)) continue; @@ -264,21 +267,22 @@ export function createScheduler( } function runAssignedTaskWakeCadence(companyId: string, now: number, wokenAgentIds: Set) { - const activeAgents = db.core.agents.listActiveAgentsForCompany(companyId); + const activeAgents = db.core.agents.listHeartbeatEligibleAgents(companyId); for (const activeAgent of activeAgents) { if (wokenAgentIds.has(activeAgent.id)) continue; if (hasActiveRun({ agentId: activeAgent.id, companyId })) continue; if (deps.checkpointEngine?.shouldPauseHeartbeat(companyId, activeAgent.id)) continue; const lastHeartbeat = activeAgent.lastHeartbeatAt ? new Date(activeAgent.lastHeartbeatAt).getTime() : 0; - const lastAttempt = taskWakeAttemptByAgent.get(activeAgent.id) ?? 0; - const intervalMs = (activeAgent.heartbeatIntervalSec || 120) * 1000; + const attemptKey = agentAttemptKey(companyId, activeAgent.id); + const lastAttempt = taskWakeAttemptByAgent.get(attemptKey) ?? 0; + const intervalMs = (activeAgent.heartbeatIntervalSec ?? 120) * 1000; if (now - Math.max(lastHeartbeat, lastAttempt) < intervalMs) continue; const actionableTasks = db.core.tasks.listAgentTasks(activeAgent.id, companyId, ["todo"]); if (!actionableTasks.some((task) => !task.activeRunId)) continue; - taskWakeAttemptByAgent.set(activeAgent.id, now); + taskWakeAttemptByAgent.set(attemptKey, now); wakeAgentIfPossible(activeAgent.id, companyId, "timer"); } } diff --git a/src/main/orchestration/task-runner.ts b/src/main/orchestration/task-runner.ts index 8601ff75..e9861966 100644 --- a/src/main/orchestration/task-runner.ts +++ b/src/main/orchestration/task-runner.ts @@ -214,6 +214,31 @@ export function createTaskRunner( } const snapshot = db.projection.snapshot.listSnapshot(); const companiesById = new Map(snapshot.companies.map((company) => [company.id, company])); + const reservedAgentIds = new Set(); + const reservedTaskIds = new Set(); + + function finishQueuedRecoveryRun( + run: ReturnType[number], + status: "cancelled" | "interrupted", + summary: string, + errorMessage: string | null, + ): void { + db.core.runs.finishRun({ + runId: run.id, + status, + summary, + errorMessage, + exitCode: null, + signal: null, + model: null, + unattributedCost: true, + recoverTaskForRetry: run.taskId !== null, + }); + if (run.workspaceId) { + db.core.workspaces.releaseWorkspaceLock(run.workspaceId, run.id); + } + emitEvent({ type: "run-updated", runId: run.id, action: "completed" }); + } for (const run of db.core.runs.listQueuedRuns()) { const company = companiesById.get(run.companyId); @@ -221,13 +246,23 @@ export function createTaskRunner( continue; } - const hasBlockingSiblingRun = snapshot.runs.some((entry) => + const hasRunningSiblingRun = snapshot.runs.some((entry) => entry.id !== run.id && entry.companyId === run.companyId && - ["queued", "running"].includes(entry.status) && + entry.status === "running" && (entry.taskId === run.taskId || entry.agentId === run.agentId), ); - if (hasBlockingSiblingRun) { + if (hasRunningSiblingRun) { + continue; + } + + if (reservedAgentIds.has(run.agentId) || (run.taskId !== null && reservedTaskIds.has(run.taskId))) { + finishQueuedRecoveryRun( + run, + "cancelled", + "Queued run superseded during recovery.", + null, + ); continue; } @@ -257,6 +292,21 @@ export function createTaskRunner( detail: `Recovered queued run blocked by connector/model policy for task "${task?.title ?? "(heartbeat)"}"`, }); emitEvent({ type: "run-updated", runId: run.id, action: "completed" }); + } else { + finishQueuedRecoveryRun( + run, + "cancelled", + "Queued run recovery deferred by eligibility gate.", + eligibilityFailure, + ); + db.core.activity.addActivity({ + companyId: run.companyId, + actor: "system", + action: "run.recovery_deferred", + entityType: "run", + entityId: run.id, + detail: `Recovered queued run deferred by ${eligibilityFailure} for task "${task?.title ?? "(heartbeat)"}"`, + }); } continue; } @@ -275,8 +325,19 @@ export function createTaskRunner( snapshot, trigger: "recovery", }); + const currentRun = db.core.runs.getRun(run.id); + if (currentRun.status === "queued" || currentRun.status === "running") { + reservedAgentIds.add(run.agentId); + if (run.taskId !== null) reservedTaskIds.add(run.taskId); + } } catch (error) { logger.warn(`[run] Skipped queued run recovery for ${run.id}: ${error instanceof Error ? error.message : String(error)}`); + finishQueuedRecoveryRun( + run, + "interrupted", + "Queued run recovery failed.", + error instanceof Error ? error.message : String(error), + ); } } } diff --git a/src/main/worker-service.ts b/src/main/worker-service.ts index 700d355a..18f1ba5f 100644 --- a/src/main/worker-service.ts +++ b/src/main/worker-service.ts @@ -1,6 +1,6 @@ import { app, utilityProcess } from "electron"; import { workerMessageSchema, type WorkerServiceDependencies } from "./worker/entry/types"; -import { handleWorkerExit } from "./worker/entry/crash-recovery"; +import { handleWorkerExit, requeueQueuedRunsAfterWorkerRestart } from "./worker/entry/crash-recovery"; import { handleWorkerMessage } from "./worker/entry/message-handler"; import { DEFAULT_MAX_CONSECUTIVE_SELF_WAKES } from "./run-attempt-policy"; @@ -44,7 +44,7 @@ export function createWorkerService({ const HEALTHY_UPTIME_MS = 60_000; const crashLoopState = { consecutive: 0, halted: false, lastCreatedAt: 0 }; - function scheduleWorkerRespawn() { + function scheduleWorkerRespawn(afterRespawn?: () => void) { if (crashLoopState.halted || destroyed || isQuitting()) return; const uptimeMs = Date.now() - crashLoopState.lastCreatedAt; if (uptimeMs >= HEALTHY_UPTIME_MS) { @@ -70,6 +70,7 @@ export function createWorkerService({ setTimeout(() => { if (destroyed || isQuitting() || crashLoopState.halted) return; createWorker(); + afterRespawn?.(); }, delay); } @@ -91,29 +92,36 @@ export function createWorkerService({ }, }); - worker.stdout?.on("data", (chunk) => { + const currentWorker = worker; + + currentWorker.stdout?.on("data", (chunk) => { const text = String(chunk).trim(); if (text) logger.debug(`[worker:stdout] ${text}`); }); - worker.stderr?.on("data", (chunk) => { + currentWorker.stderr?.on("data", (chunk) => { const text = String(chunk).trim(); if (text) logger.warn(`[worker:stderr] ${text}`); }); - worker.on("exit", (code) => { + currentWorker.on("exit", (code) => { + if (worker === currentWorker) worker = null; if (isQuitting() || destroyed) return; - handleWorkerExit( + const requeueCandidates = handleWorkerExit( code, - { db, logger, isQuitting, publishDomainChanged, emitEvent, wakeAgentIfPossible, dispatchRunToWorker }, - scheduleWorkerRespawn, - getWorker, + { db, logger, isQuitting, publishDomainChanged, emitEvent, wakeAgentIfPossible }, + ); + scheduleWorkerRespawn( + () => requeueQueuedRunsAfterWorkerRestart( + requeueCandidates, + { db, logger, dispatchRunToWorker }, + ), ); }); - worker.on("message", async (message) => { + currentWorker.on("message", async (message) => { if (isShutdownReadyMessage(message)) { try { - worker?.postMessage({ type: "shutdown-exit" }); + currentWorker.postMessage({ type: "shutdown-exit" }); } catch (error) { logger.debug( `[worker-service] shutdown-exit postMessage failed: ${error instanceof Error ? error.message : String(error)}`, @@ -123,7 +131,7 @@ export function createWorkerService({ } await handleWorkerMessage( message, - worker, + currentWorker, { db, logger, diff --git a/src/main/worker/entry/crash-recovery.ts b/src/main/worker/entry/crash-recovery.ts index 7dcb1e80..5c0cae21 100644 --- a/src/main/worker/entry/crash-recovery.ts +++ b/src/main/worker/entry/crash-recovery.ts @@ -1,6 +1,15 @@ import type { WorkerServiceDependencies } from "./types"; import { resolveWorkspaceWithLockMode } from "../../orchestration/resolve-workspace"; +export interface QueuedRunRecoveryCandidate { + id: string; + taskId: string | null; + agentId: string; + companyId: string; + workspaceId: string | null; + connectorId: string; +} + export function handleWorkerExit( code: number | null, { @@ -10,26 +19,16 @@ export function handleWorkerExit( publishDomainChanged, emitEvent: _emitEvent, wakeAgentIfPossible, - dispatchRunToWorker, - }: Pick, - createWorker: () => void, - getWorker: () => ReturnType | null, -) { - if (isQuitting()) return; + }: Pick, +): QueuedRunRecoveryCandidate[] { + if (isQuitting()) return []; logger.warn(`[worker] Worker exited with code ${code}, recovering...`); const snapshot = db.projection.snapshot.listSnapshot(); const affectedAgentIds = new Set(); const affectedCompanyIds = new Map(); - const requeueCandidates: Array<{ - id: string; - taskId: string | null; - agentId: string; - companyId: string; - workspaceId: string | null; - connectorId: string; - }> = []; + const requeueCandidates: QueuedRunRecoveryCandidate[] = []; for (const run of snapshot.runs) { if (run.status === "running") { @@ -42,14 +41,11 @@ export function handleWorkerExit( signal: null, model: null, unattributedCost: true, + recoverTaskForRetry: run.taskId !== null, }); if (run.workspaceId) { db.core.workspaces.releaseWorkspaceLock(run.workspaceId, run.id); } - // EXEC-02: Release execution lock on crash recovery - if (run.taskId) { - db.core.tasks.releaseTaskLock(run.taskId, run.id); - } affectedAgentIds.add(run.agentId); affectedCompanyIds.set(run.agentId, run.companyId); @@ -77,61 +73,70 @@ export function handleWorkerExit( logger.info(`[worker] Promoted ${promoted} deferred wake(s) for agent ${agentId} after crash recovery`); } const pendingWakeTrigger = db.automation.wakeup.getPreferredWakeTrigger(agentId, cid); - if (pendingWakeTrigger) { - wakeAgentIfPossible(agentId, cid, pendingWakeTrigger); - } + wakeAgentIfPossible(agentId, cid, pendingWakeTrigger ?? "recovery"); } publishDomainChanged(); - createWorker(); + return requeueCandidates; +} - const worker = getWorker(); - if (requeueCandidates.length > 0 && worker) { - logger.info(`[worker] Re-enqueuing ${requeueCandidates.length} queued run(s) after worker restart.`); - const recoverySnapshot = db.projection.snapshot.listSnapshot(); - for (const candidate of requeueCandidates) { - try { - // The crashed run may still own a workspace lock acquired during the - // original dispatch; release it before re-dispatch so the replay - // does not self-collide with "Workspace is locked by another run". - if (candidate.workspaceId) { - db.core.workspaces.releaseWorkspaceLock(candidate.workspaceId, candidate.id); - } - const task = candidate.taskId ? db.core.entity.getTask(candidate.taskId) : null; - const agent = db.core.entity.getAgent(candidate.agentId); - const connector = db.core.connectors.getConnector(agent.connectorId); - const { workspace, lockMode } = resolveWorkspaceWithLockMode(db, task, agent, candidate.companyId); - const recoveryCompany = recoverySnapshot.companies.find((company) => company.id === candidate.companyId); +export function requeueQueuedRunsAfterWorkerRestart( + requeueCandidates: QueuedRunRecoveryCandidate[], + { + db, + logger, + dispatchRunToWorker, + }: Pick, +): void { + if (requeueCandidates.length === 0) return; + logger.info(`[worker] Re-enqueuing ${requeueCandidates.length} queued run(s) after worker restart.`); + const recoverySnapshot = db.projection.snapshot.listSnapshot(); + for (const candidate of requeueCandidates) { + try { + const currentRun = db.core.runs.getRun(candidate.id); + if (currentRun.status !== "queued") continue; + // The crashed run may still own a workspace lock acquired during the + // original dispatch; release it before re-dispatch so the replay + // does not self-collide with "Workspace is locked by another run". + if (candidate.workspaceId) { + db.core.workspaces.releaseWorkspaceLock(candidate.workspaceId, candidate.id); + } + const task = candidate.taskId ? db.core.entity.getTask(candidate.taskId) : null; + const agent = db.core.entity.getAgent(candidate.agentId); + const connector = db.core.connectors.getConnector(agent.connectorId); + const { workspace, lockMode, observationPath } = resolveWorkspaceWithLockMode(db, task, agent, candidate.companyId); + const recoveryCompany = recoverySnapshot.companies.find((company) => company.id === candidate.companyId); - dispatchRunToWorker({ - runId: candidate.id, - companyId: candidate.companyId, - agent, - task, - connector, - workspace, - workspaceMode: lockMode, - companyName: recoveryCompany?.name ?? "Company", - companyDescription: recoveryCompany?.description ?? "", - snapshot: recoverySnapshot, - trigger: "recovery", - }); - } catch (error) { - logger.warn(`[worker] Failed to re-enqueue run ${candidate.id}: ${error instanceof Error ? error.message : String(error)}`); - db.core.runs.finishRun({ - runId: candidate.id, - status: "interrupted", - summary: "Could not re-enqueue after worker restart.", - errorMessage: error instanceof Error ? error.message : String(error), - exitCode: null, - signal: null, - model: null, - unattributedCost: true, - }); - if (candidate.workspaceId) { - db.core.workspaces.releaseWorkspaceLock(candidate.workspaceId, candidate.id); - } + dispatchRunToWorker({ + runId: candidate.id, + companyId: candidate.companyId, + agent, + task, + connector, + workspace, + workspaceMode: lockMode, + observationPath, + companyName: recoveryCompany?.name ?? "Company", + companyDescription: recoveryCompany?.description ?? "", + snapshot: recoverySnapshot, + trigger: "recovery", + }); + } catch (error) { + logger.warn(`[worker] Failed to re-enqueue run ${candidate.id}: ${error instanceof Error ? error.message : String(error)}`); + db.core.runs.finishRun({ + runId: candidate.id, + status: "interrupted", + summary: "Could not re-enqueue after worker restart.", + errorMessage: error instanceof Error ? error.message : String(error), + exitCode: null, + signal: null, + model: null, + unattributedCost: true, + recoverTaskForRetry: candidate.taskId !== null, + }); + if (candidate.workspaceId) { + db.core.workspaces.releaseWorkspaceLock(candidate.workspaceId, candidate.id); } } } diff --git a/src/main/worker/entry/types.ts b/src/main/worker/entry/types.ts index cbc391ba..d7c87177 100644 --- a/src/main/worker/entry/types.ts +++ b/src/main/worker/entry/types.ts @@ -5,6 +5,7 @@ import type { NotificationOptions } from "../../notification-service"; import type { createBudgetEnforcementService } from "../../budget-enforcement-service"; import type { createSessionManagementService } from "../../session-management-service"; import type { RuntimeBridge, RuntimeMessageInjector } from "../../agent-runtime"; +import type { RunDispatchParams } from "../../orchestration/turn-context"; const runStatusSchema = z.enum(["queued", "running", "succeeded", "failed", "cancelled", "timed_out", "interrupted"]); const runUsageStatusSchema = z.enum(RUN_USAGE_STATUSES); @@ -110,19 +111,7 @@ export interface WorkerServiceDependencies { publishDomainChanged: () => void; emitEvent: (event: DesktopEvent) => void; wakeAgentIfPossible: (agentId: string | null | undefined, companyId: string, trigger: string) => string | null; - dispatchRunToWorker: (params: { - runId: string; - companyId: string; - agent: ReturnType; - task: ReturnType | null; - connector: ReturnType; - workspace: ReturnType; - workspaceMode: "exclusive" | "shared" | "none"; - companyName: string; - companyDescription: string; - snapshot: ReturnType; - trigger: string; - }) => void; + dispatchRunToWorker: (params: RunDispatchParams) => void; processRunCompletion: (run: RunRecord, event: z.infer & { type: "run-finished" }) => void; onRunStarted?: (runId: string, taskId: string | null, agentId: string, companyId: string) => void; onRunCompleted?: (runId: string, taskId: string | null, agentId: string, companyId: string, status: string, summary: string | null, errorMessage: string | null) => void; diff --git a/src/main/worker/handlers/run-finished-handler.ts b/src/main/worker/handlers/run-finished-handler.ts index 9ccd2707..e562e2c5 100644 --- a/src/main/worker/handlers/run-finished-handler.ts +++ b/src/main/worker/handlers/run-finished-handler.ts @@ -15,6 +15,10 @@ export interface RunFinishedHooks { | { ok: false; errorMessage: string }; } +function errorMessage(error: unknown): string { + return error instanceof Error ? error.message : String(error); +} + export function handleRunFinished( event: Extract, deps: HandlerDeps, @@ -37,7 +41,11 @@ export function handleRunFinished( } : event; if (boundaryResult && !boundaryResult.ok) { - deps.db.core.runs.markRunCompletionFailure(event.runId, boundaryResult.errorMessage); + try { + deps.db.core.runs.markRunCompletionFailure(event.runId, boundaryResult.errorMessage); + } catch (error) { + deps.logger.error(`[worker] markRunCompletionFailure failed for artifact boundary ${event.runId}: ${errorMessage(error)}`); + } } emitAndNotifyCompletion(effectiveEvent, deps, rfCtx); const { hadDeferredWakes } = runPostCompletionAutomation(effectiveEvent, deps, context, rfCtx); diff --git a/src/main/worker/handlers/run-finished-helpers.ts b/src/main/worker/handlers/run-finished-helpers.ts index 4b1e9cda..9717f8f8 100644 --- a/src/main/worker/handlers/run-finished-helpers.ts +++ b/src/main/worker/handlers/run-finished-helpers.ts @@ -16,6 +16,22 @@ export interface RunFinishedContext { runOutcome: ReturnType; } +function errorMessage(error: unknown): string { + return error instanceof Error ? error.message : String(error); +} + +function runCompletionSideEffect( + logger: HandlerDeps["logger"], + label: string, + operation: () => void, +): void { + try { + operation(); + } catch (error) { + logger.error(`[worker] ${label} failed after run completion: ${errorMessage(error)}`); + } +} + export function finalizeRun( event: RunFinishedEvent, deps: HandlerDeps, @@ -71,13 +87,15 @@ export function finalizeRun( leaseRenewalMap.delete(event.runId); const runOutcome = db.core.runs.finishRun({ ...event, unattributedCost }); - db.core.activity.addActivity({ - companyId: run.companyId, - actor: run.agentId, - action: `run.${event.status}`, - entityType: "run", - entityId: run.id, - detail: event.summary ?? event.errorMessage ?? "", + runCompletionSideEffect(logger, "run activity logging", () => { + db.core.activity.addActivity({ + companyId: run.companyId, + actor: run.agentId, + action: `run.${event.status}`, + entityType: "run", + entityId: run.id, + detail: event.summary ?? event.errorMessage ?? "", + }); }); return { run, task, runOutcome }; @@ -85,11 +103,13 @@ export function finalizeRun( export function releaseCompletedRunWorkspaceLock( rfCtx: RunFinishedContext | null, - deps: Pick, + deps: Pick, ): void { const workspaceId = rfCtx?.run.workspaceId; if (!workspaceId) return; - deps.db.core.workspaces.releaseWorkspaceLock(workspaceId, rfCtx.run.id); + runCompletionSideEffect(deps.logger, "workspace lock release", () => { + deps.db.core.workspaces.releaseWorkspaceLock(workspaceId, rfCtx.run.id); + }); } export function accumulateAndRotateSession( @@ -120,22 +140,26 @@ export function accumulateAndRotateSession( const session = db.core.tasks.getSessionById(agentRecord.sessionId); if (!session) return; - sessionService.accumulateTokenUsage( - session.id, - event.inputTokens ?? 0, - event.outputTokens ?? 0, - event.runId, - ); - // Re-read after accumulation: evaluateRotation must see the counts - // that include this run, otherwise a run that just crossed the - // threshold waits a full additional cycle to rotate — the exact - // failure mode that let transcripts balloon past the context window. - const updatedSession = db.core.tasks.getSessionById(agentRecord.sessionId); - if (!updatedSession) return; - const rotationResult = sessionService.evaluateRotation(updatedSession); - if (rotationResult.shouldRotate) { - sessionService.rotateSession(updatedSession); - logger.info(`[session] Rotated session for agent ${run.agentId}: ${rotationResult.reason}`); + try { + sessionService.accumulateTokenUsage( + session.id, + event.inputTokens ?? 0, + event.outputTokens ?? 0, + event.runId, + ); + // Re-read after accumulation: evaluateRotation must see the counts + // that include this run, otherwise a run that just crossed the + // threshold waits a full additional cycle to rotate — the exact + // failure mode that let transcripts balloon past the context window. + const updatedSession = db.core.tasks.getSessionById(agentRecord.sessionId); + if (!updatedSession) return; + const rotationResult = sessionService.evaluateRotation(updatedSession); + if (rotationResult.shouldRotate) { + sessionService.rotateSession(updatedSession); + logger.info(`[session] Rotated session for agent ${run.agentId}: ${rotationResult.reason}`); + } + } catch (error) { + logger.error(`[session] accumulation/rotation failed for run ${run.id}: ${errorMessage(error)}`); } } @@ -147,20 +171,26 @@ export function emitAndNotifyCompletion( const { db, emitEvent, notify, logger } = deps; const { run, task, runOutcome } = rfCtx; - emitEvent({ type: "run-updated", runId: run.id, action: "completed" }); - emitEvent({ - type: "run-status", - runId: run.id, - status: event.status, - message: event.summary ?? event.errorMessage ?? event.status, + runCompletionSideEffect(logger, "run-updated event emission", () => { + emitEvent({ type: "run-updated", runId: run.id, action: "completed" }); + }); + runCompletionSideEffect(logger, "run-status event emission", () => { + emitEvent({ + type: "run-status", + runId: run.id, + status: event.status, + message: event.summary ?? event.errorMessage ?? event.status, + }); }); if ((event.status === "failed" || event.status === "timed_out" || event.status === "interrupted") && task) { - notify({ - title: "Task Failed", - body: `${task.title} -- manual intervention needed`, - urgency: "informational", - navigation: { section: "tasks", entityId: task.id }, - batchKey: "failed_runs", + runCompletionSideEffect(logger, "task failure notification", () => { + notify({ + title: "Task Failed", + body: `${task.title} -- manual intervention needed`, + urgency: "informational", + navigation: { section: "tasks", entityId: task.id }, + batchKey: "failed_runs", + }); }); } @@ -175,22 +205,28 @@ export function emitAndNotifyCompletion( } if (updatedAgent && runOutcome.budgetHardStopped) { - db.core.activity.addActivity({ - companyId: updatedAgent.companyId, - actor: "system", - action: "agent.auto_paused", - entityType: "agent", - entityId: updatedAgent.id, - detail: `Budget exhausted at ${updatedAgent.spentMonthlyUsd.toFixed(2)} / ${updatedAgent.budgetMonthlyUsd.toFixed(2)} USD.`, + runCompletionSideEffect(logger, "budget auto-pause activity logging", () => { + db.core.activity.addActivity({ + companyId: updatedAgent.companyId, + actor: "system", + action: "agent.auto_paused", + entityType: "agent", + entityId: updatedAgent.id, + detail: `Budget exhausted at ${updatedAgent.spentMonthlyUsd.toFixed(2)} / ${updatedAgent.budgetMonthlyUsd.toFixed(2)} USD.`, + }); }); - notify({ - title: "Agent Paused", - body: `${updatedAgent.name} paused -- budget limit reached`, - urgency: "informational", - navigation: { section: "agents", entityId: updatedAgent.id }, - batchKey: "agent_paused", + runCompletionSideEffect(logger, "budget auto-pause notification", () => { + notify({ + title: "Agent Paused", + body: `${updatedAgent.name} paused -- budget limit reached`, + urgency: "informational", + navigation: { section: "agents", entityId: updatedAgent.id }, + batchKey: "agent_paused", + }); + }); + runCompletionSideEffect(logger, "agent-updated event emission", () => { + emitEvent({ type: "agent-updated", agentId: updatedAgent.id, action: "paused" }); }); - emitEvent({ type: "agent-updated", agentId: updatedAgent.id, action: "paused" }); } } @@ -203,58 +239,97 @@ export function runPostCompletionAutomation( const { db, logger, isQuitting, wakeAgentIfPossible, processRunCompletion, onRunCompleted, notify, budgetService } = deps; const { destroyed, MAX_RETRY_ATTEMPTS, RETRY_COOLDOWN_MS, pendingRetryTimers } = context; const { run, task } = rfCtx; + let housekeepingStatus = event.status; + let housekeepingSummary = event.summary ?? null; + let housekeepingErrorMessage = event.errorMessage ?? null; // Policy-based incident check after run completion. Always runs once the // run is finalized — cost is read from the cost_entries SUM inside // finishRun/checkAndCreateIncidents, not from the worker event. if (budgetService && event.status === "succeeded") { - budgetService.checkAndCreateIncidents(run.companyId, run.agentId, task?.projectId ?? null); + try { + budgetService.checkAndCreateIncidents(run.companyId, run.agentId, task?.projectId ?? null); + } catch (error) { + logger.error(`[worker] budget incident check failed for run ${run.id}: ${errorMessage(error)}`); + } } if ((event.status === "failed" || event.status === "timed_out") && !isQuitting() && !destroyed && run.taskId) { - handleFailureEscalation( - { - db, - wakeAgentIfPossible, - notify: (title, body) => - notify({ title, body, urgency: "critical", navigation: { section: "tasks", entityId: run.taskId ?? "" } }), - logger, - }, - { taskId: run.taskId, agentId: run.agentId, companyId: run.companyId }, - task, - event.status, - MAX_RETRY_ATTEMPTS, - RETRY_COOLDOWN_MS, - (fn, delayMs) => { - const retryTimer = setTimeout(() => { - pendingRetryTimers.delete(retryTimer); - if (isQuitting() || destroyed) return; - fn(); - }, delayMs); - pendingRetryTimers.add(retryTimer); - }, - ); + try { + handleFailureEscalation( + { + db, + wakeAgentIfPossible, + notify: (title, body) => + notify({ title, body, urgency: "critical", navigation: { section: "tasks", entityId: run.taskId ?? "" } }), + logger, + }, + { taskId: run.taskId, agentId: run.agentId, companyId: run.companyId }, + task, + event.status, + MAX_RETRY_ATTEMPTS, + RETRY_COOLDOWN_MS, + (fn, delayMs) => { + const retryTimer = setTimeout(() => { + pendingRetryTimers.delete(retryTimer); + if (isQuitting() || destroyed) return; + fn(); + }, delayMs); + pendingRetryTimers.add(retryTimer); + }, + ); + } catch (error) { + logger.error(`[worker] failure escalation failed for run ${run.id}: ${errorMessage(error)}`); + } } // EXEC-05: Promote deferred wakes after run completion let hadDeferredWakes = false; if (!isQuitting() && !destroyed) { - const promoted = db.automation.wakeup.promoteWakeupRequests(run.agentId, run.companyId); - const pendingWakeTrigger = db.automation.wakeup.getPreferredWakeTrigger(run.agentId, run.companyId); - if (promoted > 0) { - logger.info(`[worker] Promoted ${promoted} deferred wake(s) for agent ${run.agentId}`); - } - if (pendingWakeTrigger) { - hadDeferredWakes = true; - logger.info( - `[worker] Processing pending wake(s) for agent ${run.agentId} with priority trigger "${pendingWakeTrigger}"`, - ); - wakeAgentIfPossible(run.agentId, run.companyId, pendingWakeTrigger); + try { + const promoted = db.automation.wakeup.promoteWakeupRequests(run.agentId, run.companyId); + const pendingWakeTrigger = db.automation.wakeup.getPreferredWakeTrigger(run.agentId, run.companyId); + if (promoted > 0) { + logger.info(`[worker] Promoted ${promoted} deferred wake(s) for agent ${run.agentId}`); + } + if (pendingWakeTrigger) { + hadDeferredWakes = true; + logger.info( + `[worker] Processing pending wake(s) for agent ${run.agentId} with priority trigger "${pendingWakeTrigger}"`, + ); + wakeAgentIfPossible(run.agentId, run.companyId, pendingWakeTrigger); + } + } catch (error) { + logger.error(`[worker] deferred wake processing failed for run ${run.id}: ${errorMessage(error)}`); } } if (!isQuitting() && !destroyed) { - processRunCompletion(run, event as z.infer & { type: "run-finished" }); + try { + processRunCompletion(run, event as z.infer & { type: "run-finished" }); + } catch (error) { + const message = errorMessage(error); + logger.error(`[worker] processRunCompletion failed for run ${run.id}: ${message}`); + housekeepingStatus = "failed"; + housekeepingSummary = null; + housekeepingErrorMessage = message; + try { + db.core.runs.markRunCompletionFailure(run.id, message); + } catch (markError) { + logger.error(`[worker] markRunCompletionFailure failed for run ${run.id}: ${errorMessage(markError)}`); + } + if (task?.status === "in_progress") { + try { + db.core.tasks.advanceTaskStatus( + task.id, + "blocked", + `Run completion automation failed: ${message}`, + ); + } catch (statusError) { + logger.error(`[worker] failed to block task ${task.id} after completion automation failure: ${errorMessage(statusError)}`); + } + } + } } // Run housekeeping: thread events + knowledge candidates @@ -265,9 +340,9 @@ export function runPostCompletionAutomation( run.taskId, run.agentId, run.companyId, - event.status, - event.summary ?? null, - event.errorMessage ?? null, + housekeepingStatus, + housekeepingSummary, + housekeepingErrorMessage, ); } catch (err) { logger.error(`[worker] onRunCompleted housekeeping failed: ${err instanceof Error ? err.message : String(err)}`); @@ -275,7 +350,7 @@ export function runPostCompletionAutomation( } // Reset retry count on successful completion - if (event.status === "succeeded" && run.taskId) { + if (housekeepingStatus === "succeeded" && run.taskId) { handleSuccessReset(db, run.taskId, logger); } diff --git a/tests/unit/active-run-projection.test.ts b/tests/unit/active-run-projection.test.ts new file mode 100644 index 00000000..b2ce7759 --- /dev/null +++ b/tests/unit/active-run-projection.test.ts @@ -0,0 +1,117 @@ +import { mkdtemp, rm } from "node:fs/promises"; +import { join } from "node:path"; +import { tmpdir } from "node:os"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { AppDatabase } from "../../src/main/database"; + +vi.mock("electron", () => ({ + safeStorage: { + isEncryptionAvailable: () => false, + getSelectedStorageBackend: () => "basic_text", + encryptString: (value: string) => Buffer.from(value, "utf8"), + decryptString: (value: Buffer) => value.toString("utf8"), + }, +})); + +type RawDb = { + db: { + prepare: (sql: string) => { + run: (...args: unknown[]) => void; + }; + }; +}; + +function createCompany(db: AppDatabase): string { + return db.core.company.saveCompany({ + name: "Projection Co", + description: "Active run projection contract", + status: "active", + }); +} + +function createWorkspace(db: AppDatabase, companyId: string, localPath: string): string { + return db.core.workspaces.saveWorkspace({ + companyId, + name: "Primary", + localPath, + repoUrl: "", + repoRef: "", + isPrimary: true, + }); +} + +function createAgent(db: AppDatabase, companyId: string, workspaceId: string): string { + return db.core.agents.saveAgent({ + companyId, + name: "Projection Agent", + role: "Engineer", + title: "Engineer", + status: "idle", + reportsTo: null, + connectorId: "openai-codex", + workspaceId, + model: "gpt-5.4", + capabilities: "Tests active run projections", + budgetMonthlyUsd: 100, + }); +} + +describe("active run projection", () => { + let tempDir: string; + let db: AppDatabase; + + beforeEach(async () => { + tempDir = await mkdtemp(join(tmpdir(), "agentcompany-active-run-projection-")); + db = new AppDatabase(tempDir); + db.init(); + }); + + afterEach(async () => { + db.close(); + await rm(tempDir, { recursive: true, force: true }); + }); + + it("ignores terminal run leases when projecting activeRunId", () => { + const companyId = createCompany(db); + const workspaceId = createWorkspace(db, companyId, tempDir); + const agentId = createAgent(db, companyId, workspaceId); + const taskId = db.core.tasks.saveTask({ + companyId, + title: "Recoverable task", + description: "A stale terminal lease must not hide this task from recovery.", + assigneeAgentId: agentId, + workspaceId, + priority: "high", + status: "todo", + }); + const run = db.core.runs.createRun({ + companyId, + taskId, + agentId, + workspaceId: null, + connectorId: "openai-codex", + }); + db.core.runs.finishRun({ + runId: run.id, + status: "succeeded", + summary: "Done", + errorMessage: null, + exitCode: 0, + signal: null, + model: null, + unattributedCost: true, + }); + const now = new Date().toISOString(); + (db as unknown as RawDb).db + .prepare( + `insert into task_leases ( + task_id, company_id, run_id, agent_id, acquired_at, renewed_at, expires_at, created_at, updated_at + ) values (?, ?, ?, ?, ?, ?, ?, ?, ?)`, + ) + .run(taskId, companyId, run.id, agentId, now, now, now, now, now); + + expect(db.core.entity.getTask(taskId).activeRunId).toBeNull(); + expect(db.core.tasks.listAgentTasks(agentId, companyId, ["todo"])[0]?.activeRunId).toBeNull(); + expect(db.projection.snapshot.listSnapshot({ companyId }).tasks.find((task) => task.id === taskId)?.activeRunId).toBeNull(); + }); +}); diff --git a/tests/unit/approval-review-feedback-writeback.test.ts b/tests/unit/approval-review-feedback-writeback.test.ts index 3aeea474..b0863cdb 100644 --- a/tests/unit/approval-review-feedback-writeback.test.ts +++ b/tests/unit/approval-review-feedback-writeback.test.ts @@ -4,6 +4,7 @@ import { tmpdir } from "node:os"; import { afterEach, describe, expect, it, vi } from "vitest"; import { AppDatabase } from "../../src/main/database"; import { handleApprovalResolved } from "../../src/main/automation/automation-approval-handler"; +import { createHeartbeat } from "../../src/main/orchestration/heartbeat"; const tempDirs: string[] = []; @@ -18,7 +19,9 @@ afterEach(async () => { }); describe("deliverable review feedback writeback", () => { - it("persists high-importance knowledge when the board requests revisions on a deliverable", async () => { + it.each(["rejected", "revision_requested"] as const)( + "persists high-importance knowledge when the board returns %s on a deliverable", + async (decision) => { const profileDir = await createTempDir("agentcompany-review-feedback-"); const db = new AppDatabase(profileDir); db.init(); @@ -87,7 +90,7 @@ describe("deliverable review feedback writeback", () => { db.core.approvals.decideApproval({ approvalId, - state: "revision_requested", + state: decision, decisionNote: "Add stronger competitor evidence and a clearer recommendation.", }); @@ -105,7 +108,7 @@ describe("deliverable review feedback writeback", () => { handleTaskStatusChange: vi.fn(), runWorkflowPipeline: async () => {}, trackTimer: vi.fn(), - }, approvalId, companyId, "revision_requested"); + }, approvalId, companyId, decision); const knowledgeEntries = db.projection.snapshot.listSnapshot().knowledgeBase.filter((entry) => entry.companyId === companyId); expect(knowledgeEntries).toHaveLength(1); @@ -113,13 +116,147 @@ describe("deliverable review feedback writeback", () => { expect(knowledgeEntries[0]?.importance).toBe("high"); expect(knowledgeEntries[0]?.referencedEntityType).toBe("task"); expect(knowledgeEntries[0]?.referencedEntityId).toBe(taskId); - expect(knowledgeEntries[0]?.content).toContain("revision_requested"); + expect(knowledgeEntries[0]?.content).toContain(decision); expect(knowledgeEntries[0]?.content).toContain("stronger competitor evidence"); expect(wakeAgentIfPossible).toHaveBeenCalledWith(agentId, companyId, "revision_needed"); - expect(db.core.entity.getTask(taskId).status).toBe("in_progress"); + expect(wakeAgentIfPossible).toHaveBeenCalledTimes(1); + expect(db.core.entity.getTask(taskId).status).toBe("todo"); const document = db.projection.snapshot.listSnapshot().documents.find((entry) => entry.id === documentId); expect(document?.status).toBe("draft"); + db.close(); + }, + ); + + it("makes a revision request immediately checkoutable by the assignee heartbeat", async () => { + const profileDir = await createTempDir("agentcompany-review-checkout-"); + const db = new AppDatabase(profileDir); + db.init(); + + db.core.connectors.saveConnector({ + id: "openai-codex", + model: "gpt-5.4", + envBindingText: "", + notes: "Test connector", + }); + + const companyId = db.core.company.saveCompany({ + name: "Review Checkout Co", + description: "Tests review revision checkout", + status: "active", + }); + const workspaceId = db.core.workspaces.saveWorkspace({ + companyId, + name: "Primary", + localPath: profileDir, + repoUrl: "", + repoRef: "", + isPrimary: true, + }); + const agentId = db.core.agents.saveAgent({ + companyId, + name: "Research Lead", + role: "Researcher", + title: "Research Lead", + status: "active", + reportsTo: null, + connectorId: "openai-codex", + workspaceId, + model: "gpt-5.4", + capabilities: "Research and synthesis", + budgetMonthlyUsd: 180, + department: "research", + }); + const taskId = db.core.tasks.saveTask({ + companyId, + title: "Draft market report", + description: "Produce a market report with concrete evidence and business insight.", + assigneeAgentId: agentId, + workspaceId, + priority: "high", + status: "in_review", + }); + const documentId = db.core.documents.saveDocument({ + companyId, + type: "knowledge_article", + title: "Market report", + content: "Draft report body", + authorAgentId: agentId, + status: "in_review", + }); + const approvalId = db.core.approvals.requestApproval({ + companyId, + relatedTaskId: taskId, + requestedByAgentId: agentId, + type: "deliverable_review", + payloadSummary: "Review market report deliverable", + impactSummary: "Board review required before the report is finalized.", + payloadJson: JSON.stringify({ documentIds: [documentId] }), + }); + + db.core.approvals.decideApproval({ + approvalId, + state: "revision_requested", + decisionNote: "Add stronger competitor evidence and a clearer recommendation.", + }); + + const dispatchRunToWorker = vi.fn(); + const heartbeat = createHeartbeat( + { + db, + logger: { debug: vi.fn(), info: vi.fn(), warn: vi.fn(), error: vi.fn() }, + publishDomainChanged: vi.fn(), + emitEvent: vi.fn(), + notify: vi.fn(), + createRunToken: () => "test-token", + getApiPort: () => 3210, + dispatchAutomation: vi.fn(), + hasWorkerReady: () => true, + enqueueWorkerRun: vi.fn(), + isQuitting: () => false, + listRuntimeSkillEntries: () => [], + budgetService: { + autoCloseExpiredIncidents: vi.fn(() => 0), + evaluateBudgetGate: vi.fn(() => ({ allowed: true, blockedBy: [] })), + } as never, + checkpointEngine: { shouldPauseHeartbeat: () => false }, + }, + () => ({ + workspace: null, + lockMode: "none", + source: "none", + observationPath: profileDir, + }), + () => null, + dispatchRunToWorker, + (filter) => db.core.runs.hasActiveRun(filter), + ); + + handleApprovalResolved({ + db, + logger: { info: () => {}, warn: () => {}, error: () => {} }, + publishDomainChanged: () => {}, + notify: vi.fn(), + wakeAgentIfPossible: heartbeat.wakeAgentIfPossible, + queueBrowserActionExecution: () => {}, + isQuitting: () => false, + }, { + dispatchAutomation: vi.fn(), + handleTaskStatusChange: vi.fn(), + runWorkflowPipeline: async () => {}, + trackTimer: vi.fn(), + }, approvalId, companyId, "revision_requested"); + + const task = db.core.entity.getTask(taskId); + expect(task.status).toBe("in_progress"); + expect(db.core.runs.hasActiveRun({ companyId, taskId, agentId })).toBe(true); + expect(dispatchRunToWorker).toHaveBeenCalledTimes(1); + expect(dispatchRunToWorker).toHaveBeenCalledWith(expect.objectContaining({ + companyId, + task: expect.objectContaining({ id: taskId }), + trigger: "revision_needed", + })); + db.close(); }); }); diff --git a/tests/unit/queued-run-recovery.test.ts b/tests/unit/queued-run-recovery.test.ts new file mode 100644 index 00000000..d05062cf --- /dev/null +++ b/tests/unit/queued-run-recovery.test.ts @@ -0,0 +1,185 @@ +import { mkdtemp, rm } from "node:fs/promises"; +import { join } from "node:path"; +import { tmpdir } from "node:os"; +import { afterEach, describe, expect, it, vi } from "vitest"; +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"; + +vi.mock("electron", () => ({ + safeStorage: { + isEncryptionAvailable: () => false, + getSelectedStorageBackend: () => "basic_text", + encryptString: (value: string) => Buffer.from(value, "utf8"), + decryptString: (value: Buffer) => value.toString("utf8"), + }, +})); + +const tempDirs: string[] = []; + +async function createTempDir(prefix: string) { + const dir = await mkdtemp(join(tmpdir(), prefix)); + tempDirs.push(dir); + return dir; +} + +afterEach(async () => { + await Promise.all(tempDirs.splice(0).map((dir) => rm(dir, { recursive: true, force: true }))); +}); + +function createLogger() { + return { + debug: vi.fn(), + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + }; +} + +function createHarness(db: AppDatabase) { + const logger = createLogger(); + const enqueueWorkerRun = vi.fn(); + const budgetService = createBudgetEnforcementService({ + db, + logger, + notify: vi.fn(), + publishDomainChanged: vi.fn(), + }); + const deps = { + db, + logger, + publishDomainChanged: vi.fn(), + emitEvent: vi.fn(), + notify: vi.fn(), + createRunToken: () => "token", + getApiPort: () => 3210, + dispatchAutomation: vi.fn(), + hasWorkerReady: () => true, + enqueueWorkerRun, + isQuitting: () => false, + listRuntimeSkillEntries: () => [], + budgetService, + checkpointEngine: { shouldPauseHeartbeat: () => false }, + }; + const runDispatch = createRunDispatch(deps); + const taskRunner = createTaskRunner( + deps, + runDispatch.resolveWorkspace, + runDispatch.getRunEligibilityFailure, + runDispatch.dispatchRunToWorker, + (filter) => db.core.runs.hasActiveRun(filter), + ); + return { enqueueWorkerRun, taskRunner }; +} + +function createCompanyGraph(db: AppDatabase, profileDir: string) { + db.core.connectors.updateConnectorHealth({ + id: "openai-codex", + status: "ready", + authState: "authenticated", + version: "test", + lastError: null, + }); + const companyId = db.core.company.saveCompany({ + name: "Queued Recovery Co", + description: "Recovers queued runs deterministically", + status: "active", + }); + const workspaceId = db.core.workspaces.saveWorkspace({ + companyId, + name: "Primary", + localPath: profileDir, + repoUrl: "", + repoRef: "", + isPrimary: true, + }); + const agentId = db.core.agents.saveAgent({ + companyId, + name: "Engineer", + role: "Engineer", + title: "Engineer", + status: "idle", + reportsTo: null, + connectorId: "openai-codex", + workspaceId, + model: "gpt-5.4", + capabilities: "Runs queued recovery tasks", + budgetMonthlyUsd: 100, + }); + return { companyId, workspaceId, agentId }; +} + +function createQueuedTaskRun(db: AppDatabase, input: { + companyId: string; + workspaceId: string; + agentId: string; + title: string; +}) { + const taskId = db.core.tasks.saveTask({ + companyId: input.companyId, + title: input.title, + description: "Queued before worker startup recovery.", + assigneeAgentId: input.agentId, + workspaceId: input.workspaceId, + priority: "high", + status: "todo", + }); + const run = db.core.runs.createTaskRun({ + companyId: input.companyId, + taskId, + agentId: input.agentId, + workspaceId: input.workspaceId, + connectorId: "openai-codex", + }); + if (!run) throw new Error("Expected queued run to be created."); + return { taskId, runId: run.id }; +} + +describe("queued run recovery", () => { + it("dispatches the first queued run and releases same-agent queued siblings", async () => { + const profileDir = await createTempDir("agentcompany-queued-recovery-"); + const db = new AppDatabase(profileDir); + db.init(); + const { companyId, workspaceId, agentId } = createCompanyGraph(db, profileDir); + const first = createQueuedTaskRun(db, { companyId, workspaceId, agentId, title: "First queued task" }); + const second = createQueuedTaskRun(db, { companyId, workspaceId, agentId, title: "Second queued task" }); + const { enqueueWorkerRun, taskRunner } = createHarness(db); + + taskRunner.resumeQueuedRuns(); + + expect(enqueueWorkerRun).toHaveBeenCalledOnce(); + expect(enqueueWorkerRun.mock.calls[0]?.[0]).toEqual(expect.objectContaining({ + runId: first.runId, + taskId: first.taskId, + agentId, + trigger: "recovery", + })); + expect(db.core.runs.getRun(second.runId).status).toBe("cancelled"); + expect(db.core.entity.getTask(second.taskId).status).toBe("todo"); + db.close(); + }); + + it("releases queued runs when recovery is blocked by connector readiness", async () => { + const profileDir = await createTempDir("agentcompany-queued-connector-block-"); + const db = new AppDatabase(profileDir); + db.init(); + const { companyId, workspaceId, agentId } = createCompanyGraph(db, profileDir); + const queued = createQueuedTaskRun(db, { companyId, workspaceId, agentId, title: "Connector blocked task" }); + db.core.connectors.updateConnectorHealth({ + id: "openai-codex", + status: "auth_required", + authState: "missing", + version: "test", + lastError: "Login required", + }); + const { enqueueWorkerRun, taskRunner } = createHarness(db); + + taskRunner.resumeQueuedRuns(); + + expect(enqueueWorkerRun).not.toHaveBeenCalled(); + expect(db.core.runs.getRun(queued.runId).status).toBe("cancelled"); + expect(db.core.entity.getTask(queued.taskId).status).toBe("todo"); + db.close(); + }); +}); diff --git a/tests/unit/scheduler-heartbeat.test.ts b/tests/unit/scheduler-heartbeat.test.ts index 1b56ca40..82d91ecf 100644 --- a/tests/unit/scheduler-heartbeat.test.ts +++ b/tests/unit/scheduler-heartbeat.test.ts @@ -199,6 +199,38 @@ describe("unified scheduler heartbeat cadence", () => { expect(wakeAgentIfPossible).toHaveBeenCalledWith(agentId, companyId, "timer"); }); + it("does not wake assigned work for an agent with heartbeat disabled", () => { + const companyId = createCompany(db); + createConnector(db); + const workspaceId = createWorkspace(db, companyId, tempDir); + const agentId = createAgent(db, companyId, workspaceId); + db.core.agents.setHeartbeat(agentId, false, 1); + db.core.tasks.saveTask({ + companyId, + title: "Paused operating work", + description: "Manual heartbeat disable should pause periodic recovery.", + assigneeAgentId: agentId, + workspaceId, + priority: "high", + status: "todo", + }); + const wakeAgentIfPossible = vi.fn(); + const scheduler = createScheduler( + createDeps(db), + () => false, + wakeAgentIfPossible, + ); + + try { + scheduler.startAllSchedulers(); + vi.advanceTimersByTime(60_000); + } finally { + scheduler.stopAllSchedulers(); + } + + expect(wakeAgentIfPossible).not.toHaveBeenCalled(); + }); + it("wakes an agent with a queued wakeup request on heartbeat cadence", () => { const companyId = createCompany(db); createConnector(db); diff --git a/tests/unit/worker-completion-boundary.test.ts b/tests/unit/worker-completion-boundary.test.ts index c68da7ee..d72cda6d 100644 --- a/tests/unit/worker-completion-boundary.test.ts +++ b/tests/unit/worker-completion-boundary.test.ts @@ -24,19 +24,45 @@ function runFinishedMessage(runId: string) { }; } -function createHandlerDeps(calls: string[]) { - const releaseWorkspaceLock = vi.fn(() => calls.push("releaseWorkspaceLock")); +function createHandlerDeps(calls: string[], options: { + taskId?: string | null; + taskStatus?: string; + sessionAccumulateThrows?: boolean; + processRunCompletionThrows?: boolean; + activityThrows?: boolean; + emitEventThrows?: boolean; + wakeupThrows?: boolean; + releaseWorkspaceThrows?: boolean; + markRunCompletionFailureThrows?: boolean; +} = {}) { + const releaseWorkspaceLock = vi.fn(() => { + calls.push("releaseWorkspaceLock"); + if (options.releaseWorkspaceThrows) { + throw new Error("workspace release failed"); + } + }); const finishRun = vi.fn(() => { calls.push("finishRun"); return { budgetHardStopped: false }; }); const rotateSession = vi.fn(() => calls.push("rotateSession")); - const processRunCompletion = vi.fn(() => calls.push("processRunCompletion")); + const processRunCompletion = vi.fn(() => { + calls.push("processRunCompletion"); + if (options.processRunCompletionThrows) { + throw new Error("processRunCompletion failed"); + } + }); const releaseRunResources = vi.fn((): RunResourceReleaseResult => { calls.push("releaseRunResources"); return { ok: true as const }; }); - const markRunCompletionFailure = vi.fn(() => calls.push("markRunCompletionFailure")); + const markRunCompletionFailure = vi.fn(() => { + calls.push("markRunCompletionFailure"); + if (options.markRunCompletionFailureThrows) { + throw new Error("mark completion failed"); + } + }); + const advanceTaskStatus = vi.fn(() => calls.push("advanceTaskStatus")); const db = { core: { @@ -45,14 +71,21 @@ function createHandlerDeps(calls: string[]) { id: RUN_ID, companyId: "company-1", agentId: "agent-1", - taskId: null, + taskId: options.taskId ?? null, workspaceId: "workspace-1", })), finishRun, markRunCompletionFailure, }, workspaces: { releaseWorkspaceLock }, - activity: { addActivity: vi.fn(() => calls.push("addActivity")) }, + activity: { + addActivity: vi.fn(() => { + calls.push("addActivity"); + if (options.activityThrows) { + throw new Error("activity write failed"); + } + }), + }, entity: { getAgent: vi.fn(() => ({ id: "agent-1", @@ -60,11 +93,19 @@ function createHandlerDeps(calls: string[]) { reportsTo: "manager-1", sessionId: "session-1", })), + getTask: vi.fn(() => ({ + id: options.taskId ?? "task-1", + companyId: "company-1", + agentId: "agent-1", + title: "Completion task", + status: options.taskStatus ?? "in_progress", + })), countActiveGoalsForCompany: vi.fn(() => 0), }, tasks: { getSessionById: vi.fn(() => ({ id: "session-1" })), listAgentTasks: vi.fn(() => []), + advanceTaskStatus, }, agents: { getDirectReports: vi.fn(() => []), @@ -76,14 +117,24 @@ function createHandlerDeps(calls: string[]) { }, automation: { wakeup: { - promoteWakeupRequests: vi.fn(() => 0), + promoteWakeupRequests: vi.fn(() => { + if (options.wakeupThrows) { + throw new Error("wakeup promotion failed"); + } + return 0; + }), getPreferredWakeTrigger: vi.fn(() => null), }, }, }; const sessionService = { - accumulateTokenUsage: vi.fn(() => calls.push("accumulateTokenUsage")), + accumulateTokenUsage: vi.fn(() => { + calls.push("accumulateTokenUsage"); + if (options.sessionAccumulateThrows) { + throw new Error("session accumulation failed"); + } + }), evaluateRotation: vi.fn(() => ({ shouldRotate: true, reason: "unit-test" })), rotateSession, }; @@ -98,7 +149,12 @@ function createHandlerDeps(calls: string[]) { error: vi.fn(), }, isQuitting: vi.fn(() => false), - emitEvent: vi.fn(() => calls.push("emitEvent")), + emitEvent: vi.fn(() => { + calls.push("emitEvent"); + if (options.emitEventThrows) { + throw new Error("emit failed"); + } + }), wakeAgentIfPossible: vi.fn(() => null), processRunCompletion, notify: vi.fn(), @@ -111,6 +167,7 @@ function createHandlerDeps(calls: string[]) { processRunCompletion, releaseRunResources, markRunCompletionFailure, + advanceTaskStatus, }; } @@ -227,6 +284,265 @@ describe("worker completion boundary", () => { expect(worker.postMessage).toHaveBeenCalledWith({ type: "run-completion-ack", runId: RUN_ID }); }); + it("continues completion when artifact failure marking throws", async () => { + const calls: string[] = []; + const { + deps, + releaseWorkspaceLock, + processRunCompletion, + releaseRunResources, + markRunCompletionFailure, + } = createHandlerDeps(calls, { markRunCompletionFailureThrows: true }); + releaseRunResources.mockImplementationOnce(() => { + calls.push("releaseRunResources"); + return { ok: false as const, errorMessage: "completion_reconcile failed" }; + }); + const worker = { postMessage: vi.fn(() => calls.push("completionAck")) }; + + await handleWorkerMessage( + runFinishedMessage(RUN_ID), + worker as never, + deps as never, + workerMessageSchema, + false, + 2, + 30_000, + new Map(), + 3, + new Set(), + new Map(), + ); + + expect(markRunCompletionFailure).toHaveBeenCalledWith(RUN_ID, "completion_reconcile failed"); + expect(processRunCompletion).toHaveBeenCalledWith( + expect.objectContaining({ id: RUN_ID }), + expect.objectContaining({ + status: "failed", + errorMessage: "completion_reconcile failed", + usageStatus: "error_stop", + }), + ); + expect(releaseWorkspaceLock).toHaveBeenCalledWith("workspace-1", RUN_ID); + expect(worker.postMessage).toHaveBeenCalledWith({ type: "run-completion-ack", runId: RUN_ID }); + }); + + it("continues completion automation when session accumulation fails", async () => { + const calls: string[] = []; + const { + deps, + releaseWorkspaceLock, + processRunCompletion, + } = createHandlerDeps(calls, { sessionAccumulateThrows: true }); + const worker = { postMessage: vi.fn(() => calls.push("completionAck")) }; + + await handleWorkerMessage( + runFinishedMessage(RUN_ID), + worker as never, + deps as never, + workerMessageSchema, + false, + 2, + 30_000, + new Map(), + 3, + new Set(), + new Map(), + ); + + expect(processRunCompletion).toHaveBeenCalled(); + expect(releaseWorkspaceLock).toHaveBeenCalledWith("workspace-1", RUN_ID); + expect(worker.postMessage).toHaveBeenCalledWith({ type: "run-completion-ack", runId: RUN_ID }); + }); + + it("marks completion failed and blocks the task when completion automation throws", async () => { + const calls: string[] = []; + const { + deps, + releaseWorkspaceLock, + markRunCompletionFailure, + advanceTaskStatus, + } = createHandlerDeps(calls, { + taskId: "task-1", + taskStatus: "in_progress", + processRunCompletionThrows: true, + }); + const worker = { postMessage: vi.fn(() => calls.push("completionAck")) }; + + await handleWorkerMessage( + runFinishedMessage(RUN_ID), + worker as never, + deps as never, + workerMessageSchema, + false, + 2, + 30_000, + new Map(), + 3, + new Set(), + new Map(), + ); + + expect(markRunCompletionFailure).toHaveBeenCalledWith(RUN_ID, "processRunCompletion failed"); + expect(advanceTaskStatus).toHaveBeenCalledWith( + "task-1", + "blocked", + expect.stringContaining("Run completion automation failed: processRunCompletion failed"), + ); + expect(releaseWorkspaceLock).toHaveBeenCalledWith("workspace-1", RUN_ID); + expect(worker.postMessage).toHaveBeenCalledWith({ type: "run-completion-ack", runId: RUN_ID }); + }); + + it("releases workspace and acknowledges completion when activity logging fails", async () => { + const calls: string[] = []; + const { + deps, + releaseWorkspaceLock, + processRunCompletion, + } = createHandlerDeps(calls, { activityThrows: true }); + const worker = { postMessage: vi.fn(() => calls.push("completionAck")) }; + + await handleWorkerMessage( + runFinishedMessage(RUN_ID), + worker as never, + deps as never, + workerMessageSchema, + false, + 2, + 30_000, + new Map(), + 3, + new Set(), + new Map(), + ); + + expect(processRunCompletion).toHaveBeenCalled(); + expect(releaseWorkspaceLock).toHaveBeenCalledWith("workspace-1", RUN_ID); + expect(worker.postMessage).toHaveBeenCalledWith({ type: "run-completion-ack", runId: RUN_ID }); + }); + + it("releases workspace and acknowledges completion when renderer event emission fails", async () => { + const calls: string[] = []; + const { + deps, + releaseWorkspaceLock, + processRunCompletion, + } = createHandlerDeps(calls, { emitEventThrows: true }); + const worker = { postMessage: vi.fn(() => calls.push("completionAck")) }; + + await handleWorkerMessage( + runFinishedMessage(RUN_ID), + worker as never, + deps as never, + workerMessageSchema, + false, + 2, + 30_000, + new Map(), + 3, + new Set(), + new Map(), + ); + + expect(processRunCompletion).toHaveBeenCalled(); + expect(releaseWorkspaceLock).toHaveBeenCalledWith("workspace-1", RUN_ID); + expect(worker.postMessage).toHaveBeenCalledWith({ type: "run-completion-ack", runId: RUN_ID }); + }); + + it("releases workspace and acknowledges completion when deferred wake promotion fails", async () => { + const calls: string[] = []; + const { + deps, + releaseWorkspaceLock, + processRunCompletion, + } = createHandlerDeps(calls, { wakeupThrows: true }); + const worker = { postMessage: vi.fn(() => calls.push("completionAck")) }; + + await handleWorkerMessage( + runFinishedMessage(RUN_ID), + worker as never, + deps as never, + workerMessageSchema, + false, + 2, + 30_000, + new Map(), + 3, + new Set(), + new Map(), + ); + + expect(processRunCompletion).toHaveBeenCalled(); + expect(releaseWorkspaceLock).toHaveBeenCalledWith("workspace-1", RUN_ID); + expect(worker.postMessage).toHaveBeenCalledWith({ type: "run-completion-ack", runId: RUN_ID }); + }); + + it("acknowledges completion when workspace release fails", async () => { + const calls: string[] = []; + const { + deps, + releaseWorkspaceLock, + processRunCompletion, + } = createHandlerDeps(calls, { releaseWorkspaceThrows: true }); + const worker = { postMessage: vi.fn(() => calls.push("completionAck")) }; + + await handleWorkerMessage( + runFinishedMessage(RUN_ID), + worker as never, + deps as never, + workerMessageSchema, + false, + 2, + 30_000, + new Map(), + 3, + new Set(), + new Map(), + ); + + expect(processRunCompletion).toHaveBeenCalled(); + expect(releaseWorkspaceLock).toHaveBeenCalledWith("workspace-1", RUN_ID); + expect(worker.postMessage).toHaveBeenCalledWith({ type: "run-completion-ack", runId: RUN_ID }); + }); + + it("blocks the task and acknowledges completion when marking completion failure throws", async () => { + const calls: string[] = []; + const { + deps, + releaseWorkspaceLock, + markRunCompletionFailure, + advanceTaskStatus, + } = createHandlerDeps(calls, { + taskId: "task-1", + taskStatus: "in_progress", + processRunCompletionThrows: true, + markRunCompletionFailureThrows: true, + }); + const worker = { postMessage: vi.fn(() => calls.push("completionAck")) }; + + await handleWorkerMessage( + runFinishedMessage(RUN_ID), + worker as never, + deps as never, + workerMessageSchema, + false, + 2, + 30_000, + new Map(), + 3, + new Set(), + new Map(), + ); + + expect(markRunCompletionFailure).toHaveBeenCalledWith(RUN_ID, "processRunCompletion failed"); + expect(advanceTaskStatus).toHaveBeenCalledWith( + "task-1", + "blocked", + expect.stringContaining("Run completion automation failed: processRunCompletion failed"), + ); + expect(releaseWorkspaceLock).toHaveBeenCalledWith("workspace-1", RUN_ID); + expect(worker.postMessage).toHaveBeenCalledWith({ type: "run-completion-ack", runId: RUN_ID }); + }); + it("keeps same-workspace runs queued until parent acknowledges completion", async () => { const parentPort = new EventEmitter() as EventEmitter & { postMessage: ReturnType; diff --git a/tests/unit/worker-crash-recovery.test.ts b/tests/unit/worker-crash-recovery.test.ts new file mode 100644 index 00000000..94771f86 --- /dev/null +++ b/tests/unit/worker-crash-recovery.test.ts @@ -0,0 +1,201 @@ +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 { handleWorkerExit, requeueQueuedRunsAfterWorkerRestart } from "../../src/main/worker/entry/crash-recovery"; + +function createLogger() { + return { + debug: vi.fn(), + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + }; +} + +function createCompany(db: AppDatabase): string { + return db.core.company.saveCompany({ + name: "Crash Recovery Co", + description: "Worker crash recovery contract test", + status: "active", + }); +} + +function createConnector(db: AppDatabase): void { + db.core.connectors.saveConnector({ + id: "openai-codex", + model: "gpt-5.4", + envBindingText: "", + notes: "Test connector", + }); +} + +function createWorkspace(db: AppDatabase, companyId: string, localPath: string): string { + return db.core.workspaces.saveWorkspace({ + companyId, + name: "Primary", + localPath, + repoUrl: "", + repoRef: "", + isPrimary: true, + }); +} + +function createAgent(db: AppDatabase, companyId: string, workspaceId: string): string { + return db.core.agents.saveAgent({ + companyId, + name: "Recovery Agent", + role: "Engineer", + title: "Engineer", + status: "active", + reportsTo: null, + connectorId: "openai-codex", + workspaceId, + model: "gpt-5.4", + capabilities: "Continues work after runtime crashes", + budgetMonthlyUsd: 100, + }); +} + +describe("worker crash recovery", () => { + let tempDir: string; + let db: AppDatabase; + + beforeEach(async () => { + tempDir = await mkdtemp(join(tmpdir(), "agentcompany-worker-crash-")); + db = new AppDatabase(tempDir); + db.init(); + }); + + afterEach(async () => { + db.close(); + await rm(tempDir, { recursive: true, force: true }); + }); + + it("returns interrupted running tasks to checkoutable work and wakes the original agent", () => { + const companyId = createCompany(db); + createConnector(db); + const workspaceId = createWorkspace(db, companyId, tempDir); + const agentId = createAgent(db, companyId, workspaceId); + const taskId = db.core.tasks.saveTask({ + companyId, + title: "Continue Phase 2", + description: "This task must remain recoverable after the worker process exits.", + assigneeAgentId: agentId, + workspaceId, + priority: "high", + status: "todo", + }); + const run = db.core.runs.createTaskRun({ + companyId, + taskId, + agentId, + workspaceId, + connectorId: "openai-codex", + }); + expect(run).not.toBeNull(); + db.core.runs.markRunStatus(run!.id, "running", "Started"); + expect(db.core.entity.getTask(taskId).status).toBe("in_progress"); + expect(db.core.runs.hasActiveRun({ companyId, agentId })).toBe(true); + + const wakeAgentIfPossible = vi.fn(() => null); + handleWorkerExit( + 1, + { + db, + logger: createLogger(), + isQuitting: () => false, + publishDomainChanged: vi.fn(), + emitEvent: vi.fn(), + wakeAgentIfPossible, + }, + ); + + expect(db.core.runs.getRun(run!.id).status).toBe("interrupted"); + expect(db.core.entity.getTask(taskId).status).toBe("todo"); + expect(db.core.runs.hasActiveRun({ companyId, agentId })).toBe(false); + expect(wakeAgentIfPossible).toHaveBeenCalledWith(agentId, companyId, "recovery"); + + const retryRun = db.core.runs.createTaskRun({ + companyId, + taskId, + agentId, + workspaceId, + connectorId: "openai-codex", + }); + expect(retryRun).not.toBeNull(); + }); + + it("requeues queued runs after the replacement worker is ready", () => { + const companyId = createCompany(db); + createConnector(db); + const workspaceId = createWorkspace(db, companyId, tempDir); + const agentId = createAgent(db, companyId, workspaceId); + const taskId = db.core.tasks.saveTask({ + companyId, + title: "Queued recovery task", + description: "This run was queued when the worker process exited.", + assigneeAgentId: agentId, + workspaceId, + priority: "high", + status: "todo", + }); + const run = db.core.runs.createTaskRun({ + companyId, + taskId, + agentId, + workspaceId, + connectorId: "openai-codex", + }); + expect(run).not.toBeNull(); + expect(db.core.runs.getRun(run!.id).status).toBe("queued"); + expect(db.core.runs.hasActiveRun({ companyId, agentId })).toBe(true); + + const dispatchRunToWorker = vi.fn(); + const requeueCandidates = handleWorkerExit( + 1, + { + db, + logger: createLogger(), + isQuitting: () => false, + publishDomainChanged: vi.fn(), + emitEvent: vi.fn(), + wakeAgentIfPossible: vi.fn(() => null), + }, + ); + + expect(dispatchRunToWorker).not.toHaveBeenCalled(); + expect(requeueCandidates).toHaveLength(1); + expect(db.core.runs.getRun(run!.id).status).toBe("queued"); + + requeueQueuedRunsAfterWorkerRestart( + requeueCandidates, + { + db, + logger: createLogger(), + dispatchRunToWorker, + }, + ); + + expect(dispatchRunToWorker).toHaveBeenCalledOnce(); + expect(dispatchRunToWorker).toHaveBeenCalledWith(expect.objectContaining({ + runId: run!.id, + companyId, + task: expect.objectContaining({ id: taskId }), + agent: expect.objectContaining({ id: agentId }), + trigger: "recovery", + })); + expect(db.core.runs.hasActiveRun({ companyId, agentId })).toBe(true); + }); +}); diff --git a/tests/unit/worker-service-shutdown.test.ts b/tests/unit/worker-service-shutdown.test.ts index 72698174..909828a4 100644 --- a/tests/unit/worker-service-shutdown.test.ts +++ b/tests/unit/worker-service-shutdown.test.ts @@ -1,5 +1,5 @@ import { EventEmitter } from "node:events"; -import { describe, expect, it, vi } from "vitest"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; const mocks = vi.hoisted(() => ({ forkMock: vi.fn(), @@ -25,6 +25,15 @@ function createWorker() { } describe("worker service shutdown", () => { + beforeEach(() => { + mocks.forkMock.mockReset(); + }); + + afterEach(() => { + vi.clearAllTimers(); + vi.useRealTimers(); + }); + it("keeps completed run outcomes when cleaning shutdown runs", async () => { const worker = createWorker(); mocks.forkMock.mockReturnValueOnce(worker); @@ -83,4 +92,128 @@ describe("worker service shutdown", () => { status: "interrupted", })); }); + + it("clears the crashed worker reference before recovery runs", () => { + vi.useFakeTimers(); + const worker = createWorker(); + mocks.forkMock.mockReturnValueOnce(worker); + const publishDomainChanged = vi.fn(); + const service = createWorkerService({ + db: { + projection: { + snapshot: { + listSnapshot: vi.fn(() => ({ runs: [] })), + }, + }, + }, + logger: { debug: vi.fn(), info: vi.fn(), warn: vi.fn(), error: vi.fn() }, + workerPath: "/tmp/worker.js", + isQuitting: () => false, + publishDomainChanged, + emitEvent: vi.fn(), + wakeAgentIfPossible: vi.fn(() => null), + dispatchRunToWorker: vi.fn(), + processRunCompletion: vi.fn(), + onRunStarted: vi.fn(), + onRunCompleted: vi.fn(), + notify: vi.fn(), + budgetService: undefined, + sessionService: undefined, + getRuntimeBridge: vi.fn(() => null), + getRuntimeMessageInjector: vi.fn(() => null), + } as never); + + service.createWorker(); + expect(service.getWorker()).toBe(worker); + worker.emit("exit", 1); + + expect(service.getWorker()).toBeNull(); + expect(publishDomainChanged).toHaveBeenCalledOnce(); + }); + + it("requeues queued crash runs after the replacement worker is created", () => { + vi.useFakeTimers(); + const crashedWorker = createWorker(); + const replacementWorker = createWorker(); + mocks.forkMock + .mockReturnValueOnce(crashedWorker) + .mockReturnValueOnce(replacementWorker); + const queuedRun = { + id: "run-queued", + status: "queued", + taskId: "task-1", + agentId: "agent-1", + companyId: "company-1", + workspaceId: null, + connectorId: "openai-codex", + }; + const dispatchRunToWorker = vi.fn(); + const service = createWorkerService({ + db: { + projection: { + snapshot: { + listSnapshot: vi.fn(() => ({ + runs: [queuedRun], + companies: [{ id: "company-1", name: "Company", description: "Recovered company" }], + })), + }, + }, + core: { + runs: { + getRun: vi.fn(() => queuedRun), + }, + entity: { + getTask: vi.fn(() => ({ + id: "task-1", + companyId: "company-1", + workspaceId: null, + projectId: null, + requiresFilesystem: false, + })), + getAgent: vi.fn(() => ({ + id: "agent-1", + companyId: "company-1", + connectorId: "openai-codex", + workspaceId: null, + })), + }, + connectors: { + getConnector: vi.fn(() => ({ id: "openai-codex" })), + }, + workspaces: { + releaseWorkspaceLock: vi.fn(), + getCompanyObservationWorkspace: vi.fn(() => null), + }, + }, + }, + logger: { debug: vi.fn(), info: vi.fn(), warn: vi.fn(), error: vi.fn() }, + workerPath: "/tmp/worker.js", + isQuitting: () => false, + publishDomainChanged: vi.fn(), + emitEvent: vi.fn(), + wakeAgentIfPossible: vi.fn(() => null), + dispatchRunToWorker, + processRunCompletion: vi.fn(), + onRunStarted: vi.fn(), + onRunCompleted: vi.fn(), + notify: vi.fn(), + budgetService: undefined, + sessionService: undefined, + getRuntimeBridge: vi.fn(() => null), + getRuntimeMessageInjector: vi.fn(() => null), + } as never); + + service.createWorker(); + crashedWorker.emit("exit", 1); + expect(dispatchRunToWorker).not.toHaveBeenCalled(); + + vi.advanceTimersByTime(1_000); + + expect(service.getWorker()).toBe(replacementWorker); + expect(dispatchRunToWorker).toHaveBeenCalledWith(expect.objectContaining({ + runId: "run-queued", + companyId: "company-1", + trigger: "recovery", + })); + }); });