diff --git a/AgentCompany-New/src/main/app/startup-recovery.ts b/AgentCompany-New/src/main/app/startup-recovery.ts new file mode 100644 index 00000000..968df157 --- /dev/null +++ b/AgentCompany-New/src/main/app/startup-recovery.ts @@ -0,0 +1,37 @@ +import type { DatabaseSync } from "node:sqlite"; +import { AuditService } from "../services/audit-service"; + +const INTERRUPTED_MESSAGE = "Interrupted by application shutdown before completion."; + +/** + * A single app instance executes runs in-process, so any task_run still + * `running` at boot was interrupted by a previous shutdown and has no + * executor. Marking it failed — together with its in-flight foreground + * provider requests — returns the task to the heartbeat, which enqueues a + * fresh run within one tick. Background provider requests are left alone: + * the provider background worker resumes those from their external id. + */ +export function recoverInterruptedWork(db: DatabaseSync): void { + const now = new Date().toISOString(); + const audit = new AuditService(db); + const runs = db.prepare("select id, org_id, task_id from task_runs where state = 'running'") + .all() as Array<{ id: string; org_id: string; task_id: string }>; + for (const run of runs) { + db.prepare("update task_runs set state = 'failed', completed_at = ?, error_message = ? where id = ?") + .run(now, INTERRUPTED_MESSAGE, run.id); + audit.record({ + orgId: run.org_id, + entityType: "task_run", + entityId: run.id, + actorType: "system", + actorId: null, + action: "failed", + payload: { taskId: run.task_id, reason: "interrupted_by_shutdown" }, + }); + } + db.prepare(` + update provider_requests + set state = 'failed', completed_at = ?, error_message = ? + where execution_mode = 'foreground' and state in ('pending', 'streaming') + `).run(now, INTERRUPTED_MESSAGE); +} diff --git a/AgentCompany-New/src/main/index.ts b/AgentCompany-New/src/main/index.ts index 3e3db4ba..b42b8fc2 100644 --- a/AgentCompany-New/src/main/index.ts +++ b/AgentCompany-New/src/main/index.ts @@ -5,6 +5,7 @@ import { DatabaseConnection } from "./database/connection"; import { createMainWindow } from "./app/create-window"; import { registerIpcHandlers } from "./app/ipc-handlers"; import { startProviderBackgroundWorker } from "./app/provider-background-worker"; +import { recoverInterruptedWork } from "./app/startup-recovery"; import { startTaskQueueWorker } from "./app/task-queue-worker"; const __dirname = dirname(fileURLToPath(import.meta.url)); @@ -17,6 +18,7 @@ if (!hasSingleInstanceLock) { app.whenReady().then(() => { const profileDir = process.env.AGENTCOMPANY_NEW_PROFILE_DIR ?? join(app.getPath("userData"), "profile"); const connection = new DatabaseConnection(join(profileDir, "agentcompany.db")); + connection.transaction(() => recoverInterruptedWork(connection.db)); const win = createMainWindow({ preloadPath: join(__dirname, "../preload/index.js"), rendererDist: join(__dirname, "../renderer"), diff --git a/AgentCompany-New/tests/unit/startup-recovery.test.ts b/AgentCompany-New/tests/unit/startup-recovery.test.ts new file mode 100644 index 00000000..e254ccc0 --- /dev/null +++ b/AgentCompany-New/tests/unit/startup-recovery.test.ts @@ -0,0 +1,90 @@ +import { join } from "node:path"; +import { tmpdir } from "node:os"; +import { randomUUID } from "node:crypto"; +import { describe, expect, test } from "vitest"; +import { DatabaseConnection } from "../../src/main/database/connection"; +import { OnboardingService } from "../../src/main/services/onboarding-service"; +import { TaskExecutionService } from "../../src/main/services/task-execution-service"; +import { claimNextQueuedRun } from "../../src/main/app/task-queue-worker"; +import { recoverInterruptedWork } from "../../src/main/app/startup-recovery"; +import { createId } from "../../src/shared/ids"; + +describe("startup recovery", () => { + test("orphaned running runs fail at boot and the heartbeat re-enqueues the task", () => { + const connection = new DatabaseConnection(join(tmpdir(), `ac-recover-${randomUUID()}.db`)); + const onboarding = new OnboardingService({ + db: connection.db, + saveProviderCredential: ({ orgId, providerId, model }) => { + connection.db.prepare("update providers set credential_id = ?, model = ?, enabled = 1 where id = ? and org_id = ?") + .run("cred_test", model, providerId, orgId); + }, + }); + + let orgId = ""; + let providerId = ""; + connection.transaction(() => { + const summary = onboarding.bootstrap({ + companyName: "Acme Operations", + companyDescription: "Coordinate agents with auditable approvals.", + employeeName: "Dana Lee", + workspacePath: "", + executionMode: "ai_ceo", + autoApproveHires: true, + reviewDeliverables: false, + goalTitle: "Ship a verified control plane", + goalDescription: "Bring the desktop operating loop online.", + providerKey: "openai", + providerSecret: "test-secret", + providerModel: "gpt-5.1", + }); + orgId = summary.orgId; + providerId = summary.providerId; + }); + + // The previous session claimed the run and left in-flight provider requests. + const claimed = claimNextQueuedRun(connection); + expect(claimed).toBeTruthy(); + const insertRequest = (id: string, executionMode: string, state: string) => { + connection.db.prepare(` + insert into provider_requests( + id, org_id, provider_id, provider_key, workspace_id, model, prompt_sha256, + response_text, raw_response_json, state, error_message, input_char_count, + output_char_count, execution_mode, policy_decision, external_response_id, + stream_event_count, cancelled_at, created_at, completed_at + ) + values (?, ?, ?, 'openai', null, 'gpt-5.1', 'sha', null, null, ?, null, 10, 0, ?, 'allowed', ?, 0, null, ?, null) + `).run(id, orgId, providerId, state, executionMode, executionMode === "background" ? "resp_ext" : null, new Date().toISOString()); + }; + const foregroundId = createId("preq"); + const backgroundId = createId("preq"); + insertRequest(foregroundId, "foreground", "pending"); + insertRequest(backgroundId, "background", "pending"); + + connection.transaction(() => recoverInterruptedWork(connection.db)); + + const run = connection.db.prepare("select state, error_message, completed_at from task_runs where id = ?") + .get(claimed!.id) as { state: string; error_message: string; completed_at: string }; + expect(run.state).toBe("failed"); + expect(run.error_message).toContain("Interrupted"); + expect(run.completed_at).toBeTruthy(); + + // Foreground orphans fail; background ones stay for the background worker. + expect(connection.db.prepare("select state from provider_requests where id = ?").get(foregroundId)).toMatchObject({ state: "failed" }); + expect(connection.db.prepare("select state from provider_requests where id = ?").get(backgroundId)).toMatchObject({ state: "pending" }); + + // The interruption is auditable and the heartbeat immediately retries the task. + const audit = connection.db.prepare( + "select count(*) as n from audit_events where org_id = ? and entity_type = 'task_run' and action = 'failed'", + ).get(orgId) as { n: number }; + expect(audit.n).toBe(1); + const taskExecution = new TaskExecutionService(connection.db); + const enqueued = connection.transaction(() => taskExecution.enqueueHeartbeatRuns()); + expect(enqueued).toHaveLength(1); + + // Recovery is idempotent: a second boot changes nothing. + connection.transaction(() => recoverInterruptedWork(connection.db)); + const failedRuns = connection.db.prepare("select count(*) as n from task_runs where state = 'failed'").get() as { n: number }; + expect(failedRuns.n).toBe(1); + connection.close(); + }); +});