Recover interrupted runs and requests at boot

A task_run still marked running at startup has no executor - the
single-instance app that owned it is gone - yet nothing failed it, the
queue worker only claims queued rows, and the heartbeat excludes tasks
with a running run, so one interrupted shutdown wedged the company
forever. Boot now fails orphaned running runs and their in-flight
foreground provider requests (auditably), which hands the task back to
the heartbeat for an automatic retry. Background provider requests are
untouched; the background worker resumes those by external id.
This commit is contained in:
ZacharyZhang-NY committed 2026-06-09 21:16:42 -04:00
1 parent 4e47da4d6b
commit f15ed8cb6f
3 files changed
+129

No files matched your search

@@ -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);
}
+2
View File
@@ -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"),
@@ -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();
});
});