diff --git a/AgentCompany-New/src/main/app/startup-recovery.ts b/AgentCompany-New/src/main/app/startup-recovery.ts index 968df157..230232d1 100644 --- a/AgentCompany-New/src/main/app/startup-recovery.ts +++ b/AgentCompany-New/src/main/app/startup-recovery.ts @@ -6,10 +6,12 @@ const INTERRUPTED_MESSAGE = "Interrupted by application shutdown before completi /** * 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. + * executor. The run returns to the queue — interruption is not failure, so + * the work continues without spending the task's retry budget — and the + * queue worker claims it within one tick. In-flight provider requests from + * the dead process cannot be resumed: foreground ones fail, and background + * ones fail only when they never received the external response id the + * provider background worker needs to resume them. */ export function recoverInterruptedWork(db: DatabaseSync): void { const now = new Date().toISOString(); @@ -17,21 +19,21 @@ export function recoverInterruptedWork(db: DatabaseSync): void { 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); + db.prepare("update task_runs set state = 'queued', started_at = null where id = ?").run(run.id); audit.record({ orgId: run.org_id, entityType: "task_run", entityId: run.id, actorType: "system", actorId: null, - action: "failed", + action: "requeued", 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') + where state in ('pending', 'streaming') + and (execution_mode = 'foreground' or external_response_id is null) `).run(now, INTERRUPTED_MESSAGE); } diff --git a/AgentCompany-New/src/main/index.ts b/AgentCompany-New/src/main/index.ts index b42b8fc2..40a97167 100644 --- a/AgentCompany-New/src/main/index.ts +++ b/AgentCompany-New/src/main/index.ts @@ -9,48 +9,49 @@ import { recoverInterruptedWork } from "./app/startup-recovery"; import { startTaskQueueWorker } from "./app/task-queue-worker"; const __dirname = dirname(fileURLToPath(import.meta.url)); -const hasSingleInstanceLock = app.requestSingleInstanceLock(); -if (!hasSingleInstanceLock) { +// A losing second instance must never touch the database: quit is +// asynchronous, so all startup work stays behind the lock check. +if (!app.requestSingleInstanceLock()) { app.quit(); +} else { + 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"), + devServerUrl: process.env.ELECTRON_RENDERER_URL, + }); + + registerIpcHandlers({ + connection, + mainWindow: win, + artifactRoot: join(profileDir, "artifacts"), + }); + const stopProviderBackgroundWorker = startProviderBackgroundWorker({ + db: connection.db, + mainWindow: win, + }); + const autonomousExecution = process.env.AGENTCOMPANY_NEW_AUTORUN !== "0"; + const stopTaskQueueWorker = autonomousExecution + ? startTaskQueueWorker({ connection, mainWindow: win }) + : () => {}; + + app.on("second-instance", () => { + if (win.isMinimized()) win.restore(); + win.focus(); + }); + + app.on("before-quit", () => { + stopProviderBackgroundWorker(); + stopTaskQueueWorker(); + connection.close(); + }); + }); + + app.on("window-all-closed", () => { + if (process.platform !== "darwin") app.quit(); + }); } - -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"), - devServerUrl: process.env.ELECTRON_RENDERER_URL, - }); - - registerIpcHandlers({ - connection, - mainWindow: win, - artifactRoot: join(profileDir, "artifacts"), - }); - const stopProviderBackgroundWorker = startProviderBackgroundWorker({ - db: connection.db, - mainWindow: win, - }); - const autonomousExecution = process.env.AGENTCOMPANY_NEW_AUTORUN !== "0"; - const stopTaskQueueWorker = autonomousExecution - ? startTaskQueueWorker({ connection, mainWindow: win }) - : () => {}; - - app.on("second-instance", () => { - if (win.isMinimized()) win.restore(); - win.focus(); - }); - - app.on("before-quit", () => { - stopProviderBackgroundWorker(); - stopTaskQueueWorker(); - connection.close(); - }); -}); - -app.on("window-all-closed", () => { - if (process.platform !== "darwin") app.quit(); -}); diff --git a/AgentCompany-New/src/main/services/org-governance-service.ts b/AgentCompany-New/src/main/services/org-governance-service.ts index 73793aab..dbf985ee 100644 --- a/AgentCompany-New/src/main/services/org-governance-service.ts +++ b/AgentCompany-New/src/main/services/org-governance-service.ts @@ -26,20 +26,30 @@ export class OrgGovernanceService { `).run(createId("arr"), orgId, "*", RISK_LEVEL.LOW, ROUTE_TO_TYPE.EMPLOYEE, boardEmployeeId, now); } - enforceHiringPolicy(input: CreateAgentInput): HiringPolicyDecision { - const requested = input.department.trim(); + /** Resolves the canonical department name (case-insensitive), or throws naming the valid choices. */ + requireCanonicalDepartment(orgId: string, department: string): string { + const requested = department.trim(); if (!requested) throw new Error("Department is required."); - const policy = this.db.prepare(` - select department_name, max_headcount, monthly_budget_limit_cents, require_approval_above_cents - from hiring_policies + const row = this.db.prepare(` + select department_name from hiring_policies where org_id = ? and lower(department_name) = lower(?) - `).get(input.orgId, requested) as HiringPolicyRow | undefined; - if (!policy) { + `).get(orgId, requested) as { department_name: string } | undefined; + if (!row) { const known = this.db.prepare("select department_name from hiring_policies where org_id = ? order by department_name asc") - .all(input.orgId) as Array<{ department_name: string }>; - throw new Error(`Unknown department "${requested}". Departments: ${known.map((row) => row.department_name).join(", ")}.`); + .all(orgId) as Array<{ department_name: string }>; + throw new Error(`Unknown department "${requested}". Departments: ${known.map((entry) => entry.department_name).join(", ")}.`); } - const departmentName = policy.department_name; + return row.department_name; + } + + enforceHiringPolicy(input: CreateAgentInput): HiringPolicyDecision { + const departmentName = this.requireCanonicalDepartment(input.orgId, input.department); + const policy = this.db.prepare(` + select max_headcount, monthly_budget_limit_cents, require_approval_above_cents + from hiring_policies + where org_id = ? and department_name = ? + `).get(input.orgId, departmentName) as HiringPolicyRow | undefined; + if (!policy) throw new Error("Hiring policy is required for department."); const usage = this.db.prepare(` select count(*) as headcount, coalesce(sum(budget_cents), 0) as budget_cents from agents @@ -126,7 +136,6 @@ export class OrgGovernanceService { } interface HiringPolicyRow { - department_name: string; max_headcount: number; monthly_budget_limit_cents: number; require_approval_above_cents: number; diff --git a/AgentCompany-New/src/main/services/organization-service.ts b/AgentCompany-New/src/main/services/organization-service.ts index dd9891b6..c640f68f 100644 --- a/AgentCompany-New/src/main/services/organization-service.ts +++ b/AgentCompany-New/src/main/services/organization-service.ts @@ -200,7 +200,7 @@ export class OrganizationService { input.orgId, "agent_hire", `Review hire: ${input.name.trim()}`, - `Department: ${input.department.trim()}\nRole: ${input.role.trim()}\nBudget: ${input.budgetCents}`, + `Department: ${hiringPolicy.departmentName}\nRole: ${input.role.trim()}\nBudget: ${input.budgetCents}`, RISK_LEVEL.MEDIUM, "system", null, @@ -228,13 +228,13 @@ export class OrganizationService { .get(input.modelProviderId, input.orgId) as { id: string } | undefined; if (!provider) throw new Error("Agent provider was not found."); } - const department = input.department.trim(); + const department = this.governance.requireCanonicalDepartment(input.orgId, input.department); const budgetCents = Math.max(0, Math.round(input.budgetCents)); const result = this.db.prepare(` update agents set role = ?, title = ?, role_level = ?, department = ?, budget_cents = ?, model_provider_id = ?, model_id = ? where id = ? and org_id = ? - `).run(role, input.title.trim() || role, input.roleLevel.trim() || "staff", department || "General", budgetCents, input.modelProviderId, input.modelId, input.agentId, input.orgId); + `).run(role, input.title.trim() || role, input.roleLevel.trim() || "staff", department, budgetCents, input.modelProviderId, input.modelId, input.agentId, input.orgId); if (result.changes === 0) throw new Error("Agent update failed."); this.audit.record({ orgId: input.orgId, diff --git a/AgentCompany-New/tests/unit/startup-recovery.test.ts b/AgentCompany-New/tests/unit/startup-recovery.test.ts index e254ccc0..c5a6e309 100644 --- a/AgentCompany-New/tests/unit/startup-recovery.test.ts +++ b/AgentCompany-New/tests/unit/startup-recovery.test.ts @@ -10,7 +10,7 @@ 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", () => { + test("orphaned running runs return to the queue at boot without spending retry budget", () => { const connection = new DatabaseConnection(join(tmpdir(), `ac-recover-${randomUUID()}.db`)); const onboarding = new OnboardingService({ db: connection.db, @@ -44,7 +44,7 @@ describe("startup recovery", () => { // 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) => { + const insertRequest = (id: string, executionMode: string, externalResponseId: string | null) => { connection.db.prepare(` insert into provider_requests( id, org_id, provider_id, provider_key, workspace_id, model, prompt_sha256, @@ -52,39 +52,44 @@ describe("startup recovery", () => { 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()); + values (?, ?, ?, 'openai', null, 'gpt-5.1', 'sha', null, null, 'pending', null, 10, 0, ?, 'allowed', ?, 0, null, ?, null) + `).run(id, orgId, providerId, executionMode, externalResponseId, new Date().toISOString()); }; const foregroundId = createId("preq"); - const backgroundId = createId("preq"); - insertRequest(foregroundId, "foreground", "pending"); - insertRequest(backgroundId, "background", "pending"); + const acceptedBackgroundId = createId("preq"); + const unacceptedBackgroundId = createId("preq"); + insertRequest(foregroundId, "foreground", null); + insertRequest(acceptedBackgroundId, "background", "resp_ext"); + insertRequest(unacceptedBackgroundId, "background", null); 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. + // Interruption is not failure: the run returns to the queue and is claimed again. + const run = connection.db.prepare("select state, started_at, error_message from task_runs where id = ?") + .get(claimed!.id) as { state: string; started_at: string | null; error_message: string | null }; + expect(run).toMatchObject({ state: "queued", started_at: null, error_message: null }); const audit = connection.db.prepare( - "select count(*) as n from audit_events where org_id = ? and entity_type = 'task_run' and action = 'failed'", + "select count(*) as n from audit_events where org_id = ? and entity_type = 'task_run' and action = 'requeued'", ).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); + const reclaimed = claimNextQueuedRun(connection); + expect(reclaimed?.id).toBe(claimed!.id); - // Recovery is idempotent: a second boot changes nothing. + // The heartbeat does not double-enqueue the task while its run is live. + const taskExecution = new TaskExecutionService(connection.db); + expect(connection.transaction(() => taskExecution.enqueueHeartbeatRuns())).toEqual([]); + + // Dead in-flight requests fail unless the background worker can resume them by external id. + 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(acceptedBackgroundId)).toMatchObject({ state: "pending" }); + expect(connection.db.prepare("select state from provider_requests where id = ?").get(unacceptedBackgroundId)).toMatchObject({ state: "failed" }); + + // Recovery is idempotent for runs it already requeued and claimed work it does not own. 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); + const requeues = connection.db.prepare( + "select count(*) as n from audit_events where org_id = ? and entity_type = 'task_run' and action = 'requeued'", + ).get(orgId) as { n: number }; + expect(requeues.n).toBe(2); connection.close(); }); });