From 93206f73fa65bb6709c039854a1eb8c57c63c2e3 Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Wed, 3 Jun 2026 21:07:14 -0700 Subject: [PATCH] fix: Stop archived companies from waking agents (#7478) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip orchestrates AI agents for zero-human companies > - Each agent has a heartbeat scheduler that wakes it on timers and on events; every wake spawns an adapter (Claude / Codex / …) run that bills the operator's subscription > - When an operator archives a company, the agents inside it remain in invokable states — the heartbeat scheduler never consults company status — so timers keep firing and event-driven wakes (comments, mentions, blockers-resolved, etc.) keep cascading > - On real deployments this silently drains the operator's subscription: idle archived companies wake their CEOs hourly, plus any cross-company event cascade > - This pull request enforces "archived ⇒ never spawns a run" as a structural invariant by guarding the wake path AND cascading agent state on archive/reactivate > - The benefit is that archived companies stop billing the operator, and the UI/queue stays consistent with the invariant ## What Changed - `server/src/services/heartbeat.ts`: - `enqueueWakeup()` loads the company and short-circuits when status is not `active`. Background sources (timer, automation, events) write a `company.inactive` skipped wake and return `null`; explicit user invokes throw a `conflict` so the UI surfaces the real reason. - `tickTimers()` joins agents to active companies so the scheduler does not iterate archived-company agents at all (no skip-row noise). - `server/src/services/companies.ts`: - `archive(id, actor?)` pauses runnable agents with `pauseReason = "company_archived"` inside the transaction (preserving `pending_approval`, `terminated`, and agents paused for unrelated reasons), then cancels `queued`/`running` heartbeat runs after the transaction commits. - `update(id, data, actor?)` reverses the cascade only for agents whose `pauseReason === "company_archived"` on the `archived → active` transition; manually-paused agents stay paused. - Both methods emit activity-log entries (`company.archived` with `agentsPaused` + `runsCancelled`, `company.reactivated` with `agentsRestored`) so the audit trail fires regardless of caller. - `packages/shared/src/constants.ts` + `server/src/services/budgets.ts`: add `company_archived` to the legal `PauseReason` union so the restorable marker is a first-class value. - `packages/db/src/migrations/0094_backfill_archived_company_agent_pauses.sql`: backfill so existing archived-company agents become `paused / company_archived` (excludes `pending_approval`). - `ui/src/lib/activity-format.ts`: add the `company.reactivated` label. ## Verification - `npx vitest run src/__tests__/companies-service.test.ts` — archive cascade, reactivate cascade, and activity-log entries (with counts) all pass. - `npx vitest run src/__tests__/heartbeat-archived-company-guard.test.ts` — timer + on-demand + event-wake paths all blocked for archived companies; `company.inactive` skipped-wake row written; user-initiated wakes throw `conflict`. - `pnpm typecheck` — clean. - Manual repro from the bug description: archive a company, wait an interval / post a comment on one of its issues, observe zero new heartbeat runs. ## Risks - Migration `0094` is a single bulk UPDATE on `agents` joined to archived `companies`. On large deployments it briefly holds row locks on archived-company agent rows; should be quick because the predicate is narrow (`status NOT IN (paused, terminated, pending_approval)` and `companies.status = 'archived'`). - New `pauseReason` value (`company_archived`) is opaque to older clients that only know the previous union. Acceptable because the union is read as plain text and the contract is sync'd in the same change. - Behavior change for users: invoking an agent in an archived company now fails with a conflict instead of silently spawning a run. Intended. ## Model Used - Claude (Anthropic) — model `claude-opus-4-7` ("Opus 4.7"), Claude Code CLI, with tool use (Read/Edit/Bash/Grep). No extended thinking mode. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [ ] If this change affects the UI, I have included before/after screenshots — N/A, no UI changes beyond an activity-log label string - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] I will address all Greptile and reviewer comments before requesting merge ## Related Work Fixes #1348 (`[Bug] Archived companies still running heartbeats and consuming tokens`). Prior attempts and parallel work in this area: - #1365 and #1429 by @hungdqdesign (March 2026) — both closed without merging. Same three-layer shape (`tickTimers` / `enqueueWakeup` / `resumeQueuedRuns` + archive-route cancellation) targeting #1348. Credit for first publicly proposing the wake-path-guard approach. - #5865 by @stubbi (May 2026, open) — adds the same `companies.status != 'archived'` joins to `tickTimers`, `enqueueWakeup`, `resumeQueuedRuns`, **and** routines `tickScheduledTriggers`, bundled with plugin-table tenant isolation (`plugin_entities` / `plugin_job_runs` / `plugin_logs` / `plugin_webhook_deliveries` get a `companyId` FK with `ON DELETE CASCADE`). This PR is narrower — it does not touch routines or plugin tables — but adds the **archive cascade** (pause agents with `pauseReason = "company_archived"`), the **reactivate reverse** (un-pause only that subset), the **`company_archived` pause-reason marker**, and a **backfill migration** for pre-existing archived companies, which #5865 does not include. Happy to coordinate sequencing or rebase if #5865 lands first. --- ...backfill_archived_company_agent_pauses.sql | 10 + packages/db/src/migrations/meta/_journal.json | 9 +- packages/shared/src/constants.ts | 2 +- .../src/__tests__/companies-service.test.ts | 754 +++++++++++++++++- .../heartbeat-archived-company-guard.test.ts | 208 +++++ server/src/routes/companies.ts | 60 +- server/src/services/budgets.ts | 3 +- server/src/services/companies.ts | 187 ++++- server/src/services/heartbeat.ts | 73 +- ui/src/lib/activity-format.ts | 1 + 10 files changed, 1247 insertions(+), 60 deletions(-) create mode 100644 packages/db/src/migrations/0094_backfill_archived_company_agent_pauses.sql create mode 100644 server/src/__tests__/heartbeat-archived-company-guard.test.ts diff --git a/packages/db/src/migrations/0094_backfill_archived_company_agent_pauses.sql b/packages/db/src/migrations/0094_backfill_archived_company_agent_pauses.sql new file mode 100644 index 00000000..30443496 --- /dev/null +++ b/packages/db/src/migrations/0094_backfill_archived_company_agent_pauses.sql @@ -0,0 +1,10 @@ +UPDATE "agents" AS a +SET + "status" = 'paused', + "pause_reason" = 'company_archived', + "paused_at" = now(), + "updated_at" = now() +FROM "companies" AS c +WHERE c."id" = a."company_id" + AND c."status" = 'archived' + AND a."status" NOT IN ('paused', 'terminated', 'pending_approval'); diff --git a/packages/db/src/migrations/meta/_journal.json b/packages/db/src/migrations/meta/_journal.json index 991a61ce..00f92ef6 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -659,6 +659,13 @@ "when": 1780040470886, "tag": "0093_giant_green_goblin", "breakpoints": true + }, + { + "idx": 94, + "version": "7", + "when": 1780533900000, + "tag": "0094_backfill_archived_company_agent_pauses", + "breakpoints": true } ] -} \ No newline at end of file +} diff --git a/packages/shared/src/constants.ts b/packages/shared/src/constants.ts index c6933bd7..fb58a85b 100644 --- a/packages/shared/src/constants.ts +++ b/packages/shared/src/constants.ts @@ -405,7 +405,7 @@ export type RoutineRunStatus = (typeof ROUTINE_RUN_STATUSES)[number]; export const ROUTINE_RUN_SOURCES = ["schedule", "manual", "api", "webhook"] as const; export type RoutineRunSource = (typeof ROUTINE_RUN_SOURCES)[number]; -export const PAUSE_REASONS = ["manual", "budget", "system"] as const; +export const PAUSE_REASONS = ["manual", "budget", "system", "company_archived"] as const; export type PauseReason = (typeof PAUSE_REASONS)[number]; export const PROJECT_COLORS = [ diff --git a/server/src/__tests__/companies-service.test.ts b/server/src/__tests__/companies-service.test.ts index 8977b041..2301b33b 100644 --- a/server/src/__tests__/companies-service.test.ts +++ b/server/src/__tests__/companies-service.test.ts @@ -1,5 +1,15 @@ +import { randomUUID } from "node:crypto"; import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; -import { companies, createDb } from "@paperclipai/db"; +import { and, eq } from "drizzle-orm"; +import { + activityLog, + agents, + agentWakeupRequests, + companies, + createDb, + heartbeatRunEvents, + heartbeatRuns, +} from "@paperclipai/db"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase, @@ -25,6 +35,11 @@ describeEmbeddedPostgres("companyService", () => { }, 20_000); afterEach(async () => { + await db.delete(heartbeatRunEvents); + await db.delete(heartbeatRuns); + await db.delete(agentWakeupRequests); + await db.delete(activityLog); + await db.delete(agents); await db.delete(companies); }); @@ -47,4 +62,741 @@ describeEmbeddedPostgres("companyService", () => { const rows = await db.select({ issuePrefix: companies.issuePrefix }).from(companies); expect(rows.map((row) => row.issuePrefix).sort()).toEqual(["ARO", "AROA"]); }); + + it("archives companies by pausing runnable agents and cancelling active runs", async () => { + const companyId = randomUUID(); + const runningAgentId = randomUUID(); + const idleAgentId = randomUUID(); + const errorAgentId = randomUUID(); + const pausedAgentId = randomUUID(); + const pendingAgentId = randomUUID(); + const terminatedAgentId = randomUUID(); + const wakeupRequestId = randomUUID(); + const runId = randomUUID(); + + await db.insert(companies).values({ + id: companyId, + name: "Archive Test Co", + issuePrefix: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`, + requireBoardApprovalForNewAgents: false, + }); + + await db.insert(agents).values([ + { + id: runningAgentId, + companyId, + name: "Running Agent", + role: "engineer", + status: "running", + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: {}, + permissions: {}, + }, + { + id: idleAgentId, + companyId, + name: "Idle Agent", + role: "engineer", + status: "idle", + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: {}, + permissions: {}, + }, + { + id: errorAgentId, + companyId, + name: "Error Agent", + role: "engineer", + status: "error", + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: {}, + permissions: {}, + }, + { + id: pausedAgentId, + companyId, + name: "Paused Agent", + role: "engineer", + status: "paused", + pauseReason: "manual", + pausedAt: new Date("2026-06-01T00:00:00Z"), + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: {}, + permissions: {}, + }, + { + id: pendingAgentId, + companyId, + name: "Pending Agent", + role: "engineer", + status: "pending_approval", + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: {}, + permissions: {}, + }, + { + id: terminatedAgentId, + companyId, + name: "Terminated Agent", + role: "engineer", + status: "terminated", + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: {}, + permissions: {}, + }, + ]); + + await db.insert(agentWakeupRequests).values({ + id: wakeupRequestId, + companyId, + agentId: runningAgentId, + source: "timer", + status: "queued", + }); + + await db.insert(heartbeatRuns).values({ + id: runId, + companyId, + agentId: runningAgentId, + invocationSource: "timer", + status: "running", + wakeupRequestId, + }); + + const archived = await companyService(db).archive(companyId, { + actorType: "user", + actorId: "test-user", + agentId: null, + runId: null, + }); + + expect(archived?.status).toBe("archived"); + + const archiveActivity = await db + .select({ + actorType: activityLog.actorType, + actorId: activityLog.actorId, + details: activityLog.details, + }) + .from(activityLog) + .where(and( + eq(activityLog.companyId, companyId), + eq(activityLog.action, "company.archived"), + )); + expect(archiveActivity).toHaveLength(1); + expect(archiveActivity[0]).toMatchObject({ + actorType: "user", + actorId: "test-user", + details: { agentsPaused: 3, runsCancelled: 1 }, + }); + + const rows = await db + .select({ + id: agents.id, + status: agents.status, + pauseReason: agents.pauseReason, + }) + .from(agents); + + const byId = new Map(rows.map((row) => [row.id, row])); + expect(byId.get(runningAgentId)).toMatchObject({ status: "paused", pauseReason: "company_archived" }); + expect(byId.get(idleAgentId)).toMatchObject({ status: "paused", pauseReason: "company_archived" }); + expect(byId.get(errorAgentId)).toMatchObject({ status: "paused", pauseReason: "company_archived" }); + expect(byId.get(pausedAgentId)).toMatchObject({ status: "paused", pauseReason: "manual" }); + expect(byId.get(pendingAgentId)).toMatchObject({ status: "pending_approval", pauseReason: null }); + expect(byId.get(terminatedAgentId)).toMatchObject({ status: "terminated", pauseReason: null }); + + const run = await db + .select({ + status: heartbeatRuns.status, + error: heartbeatRuns.error, + }) + .from(heartbeatRuns) + .then((result) => result[0] ?? null); + expect(run).toMatchObject({ + status: "cancelled", + error: "Cancelled because the company was archived", + }); + + const wakeup = await db + .select({ + status: agentWakeupRequests.status, + error: agentWakeupRequests.error, + }) + .from(agentWakeupRequests) + .then((result) => result[0] ?? null); + expect(wakeup).toMatchObject({ + status: "cancelled", + error: "Cancelled because the company was archived", + }); + }); + + it("reactivates only agents paused because the company was archived", async () => { + const companyId = randomUUID(); + const archivedPausedAgentId = randomUUID(); + const manualPausedAgentId = randomUUID(); + const pendingAgentId = randomUUID(); + + await db.insert(companies).values({ + id: companyId, + name: "Reactivate Test Co", + status: "archived", + issuePrefix: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`, + requireBoardApprovalForNewAgents: false, + }); + + await db.insert(agents).values([ + { + id: archivedPausedAgentId, + companyId, + name: "Archived Paused Agent", + role: "engineer", + status: "paused", + pauseReason: "company_archived", + pausedAt: new Date("2026-06-01T00:00:00Z"), + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: {}, + permissions: {}, + }, + { + id: manualPausedAgentId, + companyId, + name: "Manual Paused Agent", + role: "engineer", + status: "paused", + pauseReason: "manual", + pausedAt: new Date("2026-06-01T00:00:00Z"), + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: {}, + permissions: {}, + }, + { + id: pendingAgentId, + companyId, + name: "Pending Approval Agent", + role: "engineer", + status: "pending_approval", + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: {}, + permissions: {}, + }, + ]); + + const reactivated = await companyService(db).update( + companyId, + { status: "active" }, + { actorType: "user", actorId: "test-user", agentId: null, runId: null }, + ); + + expect(reactivated?.status).toBe("active"); + + const reactivateActivity = await db + .select({ + actorType: activityLog.actorType, + actorId: activityLog.actorId, + details: activityLog.details, + }) + .from(activityLog) + .where(and( + eq(activityLog.companyId, companyId), + eq(activityLog.action, "company.reactivated"), + )); + expect(reactivateActivity).toHaveLength(1); + expect(reactivateActivity[0]).toMatchObject({ + actorType: "user", + actorId: "test-user", + details: { agentsRestored: 1 }, + }); + + const rows = await db + .select({ + id: agents.id, + status: agents.status, + pauseReason: agents.pauseReason, + pausedAt: agents.pausedAt, + }) + .from(agents); + + const byId = new Map(rows.map((row) => [row.id, row])); + expect(byId.get(archivedPausedAgentId)).toMatchObject({ + status: "idle", + pauseReason: null, + pausedAt: null, + }); + expect(byId.get(manualPausedAgentId)).toMatchObject({ + status: "paused", + pauseReason: "manual", + }); + expect(byId.get(pendingAgentId)).toMatchObject({ + status: "pending_approval", + pauseReason: null, + }); + }); + + it("runs the archive cascade when update() transitions a company to archived", async () => { + const companyId = randomUUID(); + const runningAgentId = randomUUID(); + const idleAgentId = randomUUID(); + const pendingAgentId = randomUUID(); + const wakeupRequestId = randomUUID(); + const runId = randomUUID(); + + await db.insert(companies).values({ + id: companyId, + name: "Update Archive Test Co", + issuePrefix: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`, + requireBoardApprovalForNewAgents: false, + }); + + await db.insert(agents).values([ + { + id: runningAgentId, + companyId, + name: "Running Agent", + role: "engineer", + status: "running", + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: {}, + permissions: {}, + }, + { + id: idleAgentId, + companyId, + name: "Idle Agent", + role: "engineer", + status: "idle", + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: {}, + permissions: {}, + }, + { + id: pendingAgentId, + companyId, + name: "Pending Agent", + role: "engineer", + status: "pending_approval", + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: {}, + permissions: {}, + }, + ]); + + await db.insert(agentWakeupRequests).values({ + id: wakeupRequestId, + companyId, + agentId: runningAgentId, + source: "timer", + status: "queued", + }); + + await db.insert(heartbeatRuns).values({ + id: runId, + companyId, + agentId: runningAgentId, + invocationSource: "timer", + status: "running", + wakeupRequestId, + }); + + const archived = await companyService(db).update( + companyId, + { status: "archived" }, + { actorType: "user", actorId: "test-user", agentId: null, runId: null }, + ); + + expect(archived?.status).toBe("archived"); + + const rows = await db + .select({ id: agents.id, status: agents.status, pauseReason: agents.pauseReason }) + .from(agents); + const byId = new Map(rows.map((row) => [row.id, row])); + expect(byId.get(runningAgentId)).toMatchObject({ status: "paused", pauseReason: "company_archived" }); + expect(byId.get(idleAgentId)).toMatchObject({ status: "paused", pauseReason: "company_archived" }); + expect(byId.get(pendingAgentId)).toMatchObject({ status: "pending_approval", pauseReason: null }); + + const run = await db + .select({ status: heartbeatRuns.status, error: heartbeatRuns.error }) + .from(heartbeatRuns) + .then((result) => result[0] ?? null); + expect(run).toMatchObject({ + status: "cancelled", + error: "Cancelled because the company was archived", + }); + + const archiveActivity = await db + .select({ + actorType: activityLog.actorType, + actorId: activityLog.actorId, + details: activityLog.details, + }) + .from(activityLog) + .where(and( + eq(activityLog.companyId, companyId), + eq(activityLog.action, "company.archived"), + )); + expect(archiveActivity).toHaveLength(1); + expect(archiveActivity[0]).toMatchObject({ + actorType: "user", + actorId: "test-user", + details: { agentsPaused: 2, runsCancelled: 1 }, + }); + }); + + it("reactivates company_archived agents even when going via paused state (archived → paused → active)", async () => { + const companyId = randomUUID(); + const archivedPausedAgentId = randomUUID(); + const manualPausedAgentId = randomUUID(); + + await db.insert(companies).values({ + id: companyId, + name: "Indirect Reactivate Test Co", + status: "paused", + issuePrefix: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`, + requireBoardApprovalForNewAgents: false, + }); + + await db.insert(agents).values([ + { + id: archivedPausedAgentId, + companyId, + name: "Archived Paused Agent", + role: "engineer", + status: "paused", + pauseReason: "company_archived", + pausedAt: new Date("2026-06-01T00:00:00Z"), + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: {}, + permissions: {}, + }, + { + id: manualPausedAgentId, + companyId, + name: "Manual Paused Agent", + role: "engineer", + status: "paused", + pauseReason: "manual", + pausedAt: new Date("2026-06-01T00:00:00Z"), + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: {}, + permissions: {}, + }, + ]); + + const reactivated = await companyService(db).update( + companyId, + { status: "active" }, + { actorType: "user", actorId: "test-user", agentId: null, runId: null }, + ); + + expect(reactivated?.status).toBe("active"); + + const rows = await db + .select({ id: agents.id, status: agents.status, pauseReason: agents.pauseReason }) + .from(agents); + const byId = new Map(rows.map((row) => [row.id, row])); + expect(byId.get(archivedPausedAgentId)).toMatchObject({ status: "idle", pauseReason: null }); + expect(byId.get(manualPausedAgentId)).toMatchObject({ status: "paused", pauseReason: "manual" }); + + const reactivateActivity = await db + .select({ details: activityLog.details }) + .from(activityLog) + .where(and( + eq(activityLog.companyId, companyId), + eq(activityLog.action, "company.reactivated"), + )); + expect(reactivateActivity).toHaveLength(1); + expect(reactivateActivity[0]).toMatchObject({ details: { agentsRestored: 1 } }); + }); + + it("emits company.reactivated for archived → active even when no agents need restoring", async () => { + const companyId = randomUUID(); + const terminatedAgentId = randomUUID(); + + await db.insert(companies).values({ + id: companyId, + name: "Empty Reactivate Co", + status: "archived", + issuePrefix: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`, + requireBoardApprovalForNewAgents: false, + }); + + await db.insert(agents).values({ + id: terminatedAgentId, + companyId, + name: "Terminated Agent", + role: "engineer", + status: "terminated", + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: {}, + permissions: {}, + }); + + const reactivated = await companyService(db).update( + companyId, + { status: "active" }, + { actorType: "user", actorId: "test-user", agentId: null, runId: null }, + ); + + expect(reactivated?.status).toBe("active"); + + const reactivateActivity = await db + .select({ details: activityLog.details }) + .from(activityLog) + .where(and( + eq(activityLog.companyId, companyId), + eq(activityLog.action, "company.reactivated"), + )); + expect(reactivateActivity).toHaveLength(1); + expect(reactivateActivity[0]).toMatchObject({ details: { agentsRestored: 0 } }); + }); + + it("does not emit company.reactivated when paused → active restores no archive-paused agents", async () => { + const companyId = randomUUID(); + const manualPausedAgentId = randomUUID(); + + await db.insert(companies).values({ + id: companyId, + name: "Plain Unpause Co", + status: "paused", + issuePrefix: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`, + requireBoardApprovalForNewAgents: false, + }); + + await db.insert(agents).values({ + id: manualPausedAgentId, + companyId, + name: "Manual Paused Agent", + role: "engineer", + status: "paused", + pauseReason: "manual", + pausedAt: new Date("2026-06-01T00:00:00Z"), + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: {}, + permissions: {}, + }); + + const reactivated = await companyService(db).update( + companyId, + { status: "active" }, + { actorType: "user", actorId: "test-user", agentId: null, runId: null }, + ); + + expect(reactivated?.status).toBe("active"); + + const reactivateActivity = await db + .select({ id: activityLog.id }) + .from(activityLog) + .where(and( + eq(activityLog.companyId, companyId), + eq(activityLog.action, "company.reactivated"), + )); + expect(reactivateActivity).toHaveLength(0); + + const agent = await db + .select({ status: agents.status, pauseReason: agents.pauseReason }) + .from(agents) + .then((rows) => rows[0] ?? null); + expect(agent).toMatchObject({ status: "paused", pauseReason: "manual" }); + }); + + it("cancels orphan queued wakeup requests with no runId during archive", async () => { + const companyId = randomUUID(); + const agentId = randomUUID(); + const orphanWakeupId = randomUUID(); + const runWakeupId = randomUUID(); + const runId = randomUUID(); + + await db.insert(companies).values({ + id: companyId, + name: "Orphan Wakeup Co", + issuePrefix: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`, + requireBoardApprovalForNewAgents: false, + }); + + await db.insert(agents).values({ + id: agentId, + companyId, + name: "Idle Agent", + role: "engineer", + status: "idle", + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: {}, + permissions: {}, + }); + + await db.insert(agentWakeupRequests).values([ + { + id: orphanWakeupId, + companyId, + agentId, + source: "automation", + status: "queued", + }, + { + id: runWakeupId, + companyId, + agentId, + source: "timer", + status: "queued", + }, + ]); + + await db.insert(heartbeatRuns).values({ + id: runId, + companyId, + agentId, + invocationSource: "timer", + status: "running", + wakeupRequestId: runWakeupId, + }); + + const archived = await companyService(db).archive(companyId, { + actorType: "user", + actorId: "test-user", + agentId: null, + runId: null, + }); + expect(archived?.status).toBe("archived"); + + const wakeups = await db + .select({ + id: agentWakeupRequests.id, + status: agentWakeupRequests.status, + error: agentWakeupRequests.error, + }) + .from(agentWakeupRequests); + const byId = new Map(wakeups.map((row) => [row.id, row])); + expect(byId.get(orphanWakeupId)).toMatchObject({ + status: "cancelled", + error: "Cancelled because the company was archived", + }); + expect(byId.get(runWakeupId)).toMatchObject({ + status: "cancelled", + error: "Cancelled because the company was archived", + }); + }); + + it("archive() is idempotent — re-archiving emits no second cascade or activity entry", async () => { + const companyId = randomUUID(); + const agentId = randomUUID(); + + await db.insert(companies).values({ + id: companyId, + name: "Idempotent Archive Test Co", + issuePrefix: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`, + requireBoardApprovalForNewAgents: false, + }); + + await db.insert(agents).values({ + id: agentId, + companyId, + name: "Idle Agent", + role: "engineer", + status: "idle", + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: {}, + permissions: {}, + }); + + const actor = { actorType: "user" as const, actorId: "test-user", agentId: null, runId: null }; + const first = await companyService(db).archive(companyId, actor); + expect(first?.status).toBe("archived"); + + const second = await companyService(db).archive(companyId, actor); + expect(second?.status).toBe("archived"); + + const archiveActivity = await db + .select({ details: activityLog.details }) + .from(activityLog) + .where(and( + eq(activityLog.companyId, companyId), + eq(activityLog.action, "company.archived"), + )); + expect(archiveActivity).toHaveLength(1); + expect(archiveActivity[0]).toMatchObject({ details: { agentsPaused: 1, runsCancelled: 0 } }); + }); + + it("runs the archive cascade when update() transitions a paused company to archived", async () => { + const companyId = randomUUID(); + const idleAgentId = randomUUID(); + const runId = randomUUID(); + + await db.insert(companies).values({ + id: companyId, + name: "Paused To Archived Test Co", + status: "paused", + issuePrefix: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`, + requireBoardApprovalForNewAgents: false, + }); + + await db.insert(agents).values({ + id: idleAgentId, + companyId, + name: "Idle Agent", + role: "engineer", + status: "idle", + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: {}, + permissions: {}, + }); + + await db.insert(heartbeatRuns).values({ + id: runId, + companyId, + agentId: idleAgentId, + invocationSource: "timer", + status: "queued", + }); + + const archived = await companyService(db).update( + companyId, + { status: "archived" }, + { actorType: "user", actorId: "test-user", agentId: null, runId: null }, + ); + + expect(archived?.status).toBe("archived"); + + const agent = await db + .select({ status: agents.status, pauseReason: agents.pauseReason }) + .from(agents) + .then((rows) => rows[0] ?? null); + expect(agent).toMatchObject({ status: "paused", pauseReason: "company_archived" }); + + const run = await db + .select({ status: heartbeatRuns.status }) + .from(heartbeatRuns) + .then((rows) => rows[0] ?? null); + expect(run?.status).toBe("cancelled"); + + const archiveActivity = await db + .select({ details: activityLog.details }) + .from(activityLog) + .where(and( + eq(activityLog.companyId, companyId), + eq(activityLog.action, "company.archived"), + )); + expect(archiveActivity).toHaveLength(1); + expect(archiveActivity[0]).toMatchObject({ + details: { agentsPaused: 1, runsCancelled: 1 }, + }); + }); }); diff --git a/server/src/__tests__/heartbeat-archived-company-guard.test.ts b/server/src/__tests__/heartbeat-archived-company-guard.test.ts new file mode 100644 index 00000000..b0eb6b54 --- /dev/null +++ b/server/src/__tests__/heartbeat-archived-company-guard.test.ts @@ -0,0 +1,208 @@ +import { randomUUID } from "node:crypto"; +import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; +import { + agents, + agentWakeupRequests, + companies, + createDb, + heartbeatRuns, + issues, +} from "@paperclipai/db"; +import { + getEmbeddedPostgresTestSupport, + startEmbeddedPostgresTestDatabase, +} from "./helpers/embedded-postgres.js"; +import { heartbeatService } from "../services/heartbeat.ts"; + +const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; + +if (!embeddedPostgresSupport.supported) { + console.warn( + `Skipping embedded Postgres archived-company heartbeat guard tests on this host: ${embeddedPostgresSupport.reason ?? "unsupported environment"}`, + ); +} + +describeEmbeddedPostgres("heartbeat archived-company guard", () => { + let db!: ReturnType; + let tempDb: Awaited> | null = null; + + beforeAll(async () => { + tempDb = await startEmbeddedPostgresTestDatabase("heartbeat-archived-company-guard-"); + db = createDb(tempDb.connectionString); + }, 20_000); + + afterEach(async () => { + await db.delete(heartbeatRuns); + await db.delete(agentWakeupRequests); + await db.delete(issues); + await db.delete(agents); + await db.delete(companies); + }); + + afterAll(async () => { + await tempDb?.cleanup(); + }); + + async function insertArchivedAgent() { + const companyId = randomUUID(); + const agentId = randomUUID(); + + await db.insert(companies).values({ + id: companyId, + name: "Archived Co", + status: "archived", + issuePrefix: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`, + requireBoardApprovalForNewAgents: false, + }); + + await db.insert(agents).values({ + id: agentId, + companyId, + name: "Archived Agent", + role: "engineer", + status: "idle", + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: { + heartbeat: { + enabled: true, + intervalSec: 60, + wakeOnDemand: true, + }, + }, + permissions: {}, + }); + + return { companyId, agentId }; + } + + it("does not iterate archived-company agents in tickTimers", async () => { + const { agentId } = await insertArchivedAgent(); + + const heartbeat = heartbeatService(db); + const result = await heartbeat.tickTimers(new Date("2026-06-04T00:10:00Z")); + + expect(result).toMatchObject({ + checked: 0, + enqueued: 0, + skipped: 0, + }); + + const runCount = await db + .select() + .from(heartbeatRuns) + .then((rows) => rows.filter((row) => row.agentId === agentId).length); + expect(runCount).toBe(0); + }); + + it("skips background wakeups for non-active companies with a company.inactive reason", async () => { + const { agentId } = await insertArchivedAgent(); + + const heartbeat = heartbeatService(db); + const run = await heartbeat.wakeup(agentId, { + source: "automation", + triggerDetail: "system", + reason: "issue_commented", + payload: { issueId: randomUUID(), commentId: randomUUID() }, + requestedByActorType: "system", + requestedByActorId: "comment_wake", + }); + + expect(run).toBeNull(); + + const wakeup = await db + .select({ + agentId: agentWakeupRequests.agentId, + status: agentWakeupRequests.status, + reason: agentWakeupRequests.reason, + error: agentWakeupRequests.error, + }) + .from(agentWakeupRequests) + .then((rows) => rows.find((row) => row.agentId === agentId) ?? null); + + expect(wakeup).toMatchObject({ + status: "skipped", + reason: "company.inactive", + error: "Wake suppressed because company status is archived", + }); + }); + + it("does not advance issue monitors for archived companies", async () => { + const { companyId, agentId } = await insertArchivedAgent(); + const issueId = randomUUID(); + const monitorScheduledAt = new Date("2026-06-04T00:00:00Z"); + + await db.insert(issues).values({ + id: issueId, + companyId, + title: "Archived-company monitor issue", + status: "in_progress", + assigneeAgentId: agentId, + monitorNextCheckAt: monitorScheduledAt, + monitorAttemptCount: 0, + }); + + const heartbeat = heartbeatService(db); + await heartbeat.tickTimers(new Date("2026-06-04T00:10:00Z")); + + const row = await db + .select({ + monitorNextCheckAt: issues.monitorNextCheckAt, + monitorWakeRequestedAt: issues.monitorWakeRequestedAt, + monitorLastTriggeredAt: issues.monitorLastTriggeredAt, + monitorAttemptCount: issues.monitorAttemptCount, + }) + .from(issues) + .then((rows) => rows[0] ?? null); + + expect(row?.monitorWakeRequestedAt).toBeNull(); + expect(row?.monitorLastTriggeredAt).toBeNull(); + expect(row?.monitorAttemptCount).toBe(0); + expect(row?.monitorNextCheckAt?.getTime()).toBe(monitorScheduledAt.getTime()); + }); + + it("does not resume queued runs for archived companies", async () => { + const { companyId, agentId } = await insertArchivedAgent(); + const runId = randomUUID(); + + await db.insert(heartbeatRuns).values({ + id: runId, + companyId, + agentId, + invocationSource: "timer", + status: "queued", + }); + + const heartbeat = heartbeatService(db); + await heartbeat.resumeQueuedRuns(); + + const status = await db + .select({ status: heartbeatRuns.status }) + .from(heartbeatRuns) + .then((rows) => rows[0]?.status ?? null); + expect(status).toBe("queued"); + }); + + it("rejects explicit user invokes for non-active companies", async () => { + const { agentId } = await insertArchivedAgent(); + + const heartbeat = heartbeatService(db); + + await expect(heartbeat.wakeup(agentId, { + source: "on_demand", + triggerDetail: "manual", + requestedByActorType: "user", + requestedByActorId: "board-user", + })).rejects.toMatchObject({ + status: 409, + details: { status: "archived" }, + }); + + const runCount = await db + .select() + .from(heartbeatRuns) + .then((rows) => rows.filter((row) => row.agentId === agentId).length); + expect(runCount).toBe(0); + }); +}); diff --git a/server/src/routes/companies.ts b/server/src/routes/companies.ts index 3fe9664f..2704f502 100644 --- a/server/src/routes/companies.ts +++ b/server/src/routes/companies.ts @@ -1,6 +1,8 @@ import { randomUUID } from "node:crypto"; import { Router, type Request } from "express"; +import { and, count as countFn, eq } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; +import { agents as agentsTable } from "@paperclipai/db"; import { DEFAULT_FEEDBACK_DATA_SHARING_TERMS_VERSION, companyPortabilityExportSchema, @@ -366,22 +368,46 @@ export function companyRoutes(db: Db, storage?: StorageService) { } } - const company = await svc.update(companyId, body); + const transitionsToArchived = + body.status === "archived" && existingCompany.status !== "archived"; + const transitionsArchivedToActive = + body.status === "active" && existingCompany.status === "archived"; + let transitionsPausedToActiveWithArchivePausedAgents = false; + if (body.status === "active" && existingCompany.status === "paused") { + const [archivedPausedCount] = await db + .select({ value: countFn() }) + .from(agentsTable) + .where(and( + eq(agentsTable.companyId, companyId), + eq(agentsTable.status, "paused"), + eq(agentsTable.pauseReason, "company_archived"), + )); + transitionsPausedToActiveWithArchivePausedAgents = + Number(archivedPausedCount?.value ?? 0) > 0; + } + const lifecycleEventEmittedByService = + transitionsToArchived || + transitionsArchivedToActive || + transitionsPausedToActiveWithArchivePausedAgents; + + const company = await svc.update(companyId, body, actor); if (!company) { res.status(404).json({ error: "Company not found" }); return; } - await logActivity(db, { - companyId, - actorType: actor.actorType, - actorId: actor.actorId, - agentId: actor.agentId, - runId: actor.runId, - action: "company.updated", - entityType: "company", - entityId: companyId, - details: body, - }); + if (!lifecycleEventEmittedByService) { + await logActivity(db, { + companyId, + actorType: actor.actorType, + actorId: actor.actorId, + agentId: actor.agentId, + runId: actor.runId, + action: "company.updated", + entityType: "company", + entityId: companyId, + details: body, + }); + } res.json(company); }); @@ -412,19 +438,11 @@ export function companyRoutes(db: Db, storage?: StorageService) { assertBoard(req); const companyId = req.params.companyId as string; assertCompanyAccess(req, companyId); - const company = await svc.archive(companyId); + const company = await svc.archive(companyId, getActorInfo(req)); if (!company) { res.status(404).json({ error: "Company not found" }); return; } - await logActivity(db, { - companyId, - actorType: "user", - actorId: req.actor.userId ?? "board", - action: "company.archived", - entityType: "company", - entityId: companyId, - }); res.json(company); }); diff --git a/server/src/services/budgets.ts b/server/src/services/budgets.ts index 48575438..84d6663e 100644 --- a/server/src/services/budgets.ts +++ b/server/src/services/budgets.ts @@ -14,6 +14,7 @@ import type { BudgetIncidentResolutionInput, BudgetMetric, BudgetOverview, + PauseReason, BudgetPolicy, BudgetPolicySummary, BudgetPolicyUpsertInput, @@ -28,7 +29,7 @@ type ScopeRecord = { companyId: string; name: string; paused: boolean; - pauseReason: "manual" | "budget" | "system" | null; + pauseReason: PauseReason | null; }; type PolicyRow = typeof budgetPolicies.$inferSelect; diff --git a/server/src/services/companies.ts b/server/src/services/companies.ts index e69abb9e..92ac9562 100644 --- a/server/src/services/companies.ts +++ b/server/src/services/companies.ts @@ -1,4 +1,4 @@ -import { and, count, eq, gte, inArray, lt, sql } from "drizzle-orm"; +import { and, count, eq, gte, inArray, isNull, lt, notInArray, sql } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; import { companies, @@ -31,10 +31,95 @@ import { } from "@paperclipai/db"; import { notFound, unprocessable } from "../errors.js"; import { environmentService } from "./environments.js"; +import { heartbeatService } from "./heartbeat.js"; +import { logActivity } from "./activity-log.js"; + +export interface CompanyActivityActor { + actorType: "user" | "agent" | "system" | "plugin"; + actorId: string; + agentId?: string | null; + runId?: string | null; +} + +const SYSTEM_COMPANY_ACTOR: CompanyActivityActor = { + actorType: "system", + actorId: "system", + agentId: null, + runId: null, +}; export function companyService(db: Db) { const ISSUE_PREFIX_FALLBACK = "CMP"; const environmentsSvc = environmentService(db); + const heartbeat = heartbeatService(db); + + type CompanyTx = Parameters[0]>[0]; + + async function applyArchiveCascadeInTx(tx: CompanyTx, id: string) { + const pausedAgentRows = await tx + .update(agents) + .set({ + status: "paused", + pauseReason: "company_archived", + pausedAt: new Date(), + updatedAt: new Date(), + }) + .where(and( + eq(agents.companyId, id), + notInArray(agents.status, ["paused", "terminated", "pending_approval"]), + )) + .returning({ id: agents.id }); + + const activeRunIds = await tx + .select({ id: heartbeatRuns.id }) + .from(heartbeatRuns) + .where(and( + eq(heartbeatRuns.companyId, id), + inArray(heartbeatRuns.status, ["queued", "running"]), + )) + .then((rows) => rows.map((row) => row.id)); + + await tx + .update(agentWakeupRequests) + .set({ + status: "cancelled", + error: "Cancelled because the company was archived", + finishedAt: new Date(), + updatedAt: new Date(), + }) + .where(and( + eq(agentWakeupRequests.companyId, id), + inArray(agentWakeupRequests.status, ["queued", "deferred_issue_execution", "claimed"]), + isNull(agentWakeupRequests.runId), + )); + + return { agentsPaused: pausedAgentRows.length, activeRunIds }; + } + + async function finalizeArchive( + id: string, + actor: CompanyActivityActor, + cascade: { agentsPaused: number; activeRunIds: string[] }, + ) { + for (const runId of cascade.activeRunIds) { + await heartbeat.cancelRun(runId, "Cancelled because the company was archived"); + } + + await logActivity(db, { + companyId: id, + actorType: actor.actorType, + actorId: actor.actorId, + agentId: actor.agentId ?? null, + runId: actor.runId ?? null, + action: "company.archived", + entityType: "company", + entityId: id, + details: { + agentsPaused: cascade.agentsPaused, + runsCancelled: cascade.activeRunIds.length, + }, + }); + } const companySelection = { id: companies.id, @@ -185,17 +270,20 @@ export function companyService(db: Db) { return enrichCompany(hydrated); }, - update: ( + update: async ( id: string, data: Partial & { logoAssetId?: string | null }, - ) => - db.transaction(async (tx) => { + actor: CompanyActivityActor = SYSTEM_COMPANY_ACTOR, + ) => { + const result = await db.transaction(async (tx) => { const existing = await getCompanyQuery(tx) .where(eq(companies.id, id)) .then((rows) => rows[0] ?? null); if (!existing) return null; const { logoAssetId, ...companyPatch } = data; + const willReactivate = existing.status !== "active" && companyPatch.status === "active"; + const willArchive = existing.status !== "archived" && companyPatch.status === "archived"; if (logoAssetId !== undefined && logoAssetId !== null) { const nextLogoAsset = await tx @@ -217,6 +305,27 @@ export function companyService(db: Db) { .then((rows) => rows[0] ?? null); if (!updated) return null; + let agentsRestored = 0; + if (willReactivate) { + const restoredRows = await tx + .update(agents) + .set({ + status: "idle", + pauseReason: null, + pausedAt: null, + updatedAt: new Date(), + }) + .where(and( + eq(agents.companyId, id), + eq(agents.status, "paused"), + eq(agents.pauseReason, "company_archived"), + )) + .returning({ id: agents.id }); + agentsRestored = restoredRows.length; + } + + const archiveCascade = willArchive ? await applyArchiveCascadeInTx(tx, id) : null; + if (logoAssetId === null) { await tx.delete(companyLogos).where(eq(companyLogos.companyId, id)); } else if (logoAssetId !== undefined) { @@ -244,25 +353,73 @@ export function companyService(db: Db) { logoAssetId: logoAssetId === undefined ? existing.logoAssetId : logoAssetId, }], tx); - return enrichCompany(hydrated); - }), + const shouldLogReactivation = willReactivate && + (existing.status === "archived" || agentsRestored > 0); - archive: (id: string) => - db.transaction(async (tx) => { - const updated = await tx - .update(companies) - .set({ status: "archived", updatedAt: new Date() }) + return { + company: enrichCompany(hydrated), + reactivated: shouldLogReactivation ? { agentsRestored } : null, + archiveCascade, + }; + }); + if (!result) return null; + if (result.reactivated) { + await logActivity(db, { + companyId: id, + actorType: actor.actorType, + actorId: actor.actorId, + agentId: actor.agentId ?? null, + runId: actor.runId ?? null, + action: "company.reactivated", + entityType: "company", + entityId: id, + details: { agentsRestored: result.reactivated.agentsRestored }, + }); + } + if (result.archiveCascade) { + await finalizeArchive(id, actor, result.archiveCascade); + } + return result.company; + }, + + archive: async (id: string, actor: CompanyActivityActor = SYSTEM_COMPANY_ACTOR) => { + const result = await db.transaction(async (tx) => { + const existing = await tx + .select({ status: companies.status }) + .from(companies) .where(eq(companies.id, id)) - .returning() .then((rows) => rows[0] ?? null); - if (!updated) return null; + if (!existing) return null; + + const wasAlreadyArchived = existing.status === "archived"; + + if (!wasAlreadyArchived) { + await tx + .update(companies) + .set({ status: "archived", updatedAt: new Date() }) + .where(eq(companies.id, id)); + } + + const cascade = wasAlreadyArchived ? null : await applyArchiveCascadeInTx(tx, id); + const row = await getCompanyQuery(tx) .where(eq(companies.id, id)) .then((rows) => rows[0] ?? null); if (!row) return null; const [hydrated] = await hydrateCompanySpend([row], tx); - return enrichCompany(hydrated); - }), + return { + company: enrichCompany(hydrated), + cascade, + }; + }); + if (!result) return null; + + if (result.cascade) { + await finalizeArchive(id, actor, result.cascade); + } + + return result.company; + }, remove: (id: string) => db.transaction(async (tx) => { diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 09a1e5fe..06b202ec 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -29,6 +29,7 @@ import { activityLog, approvals, companySkills as companySkillsTable, + companies, documentAnnotationComments, documentAnnotationThreads, documentRevisions, @@ -3350,8 +3351,10 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) const dueMonitors = await db .select(issueMonitorDispatchColumns) .from(issues) + .innerJoin(companies, eq(companies.id, issues.companyId)) .where( and( + eq(companies.status, "active"), sql`${issues.monitorNextCheckAt} is not null`, lte(issues.monitorNextCheckAt, now), isNull(issues.assigneeUserId), @@ -6830,7 +6833,11 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) const queuedRuns = await db .select({ agentId: heartbeatRuns.agentId }) .from(heartbeatRuns) - .where(eq(heartbeatRuns.status, "queued")); + .innerJoin(companies, eq(companies.id, heartbeatRuns.companyId)) + .where(and( + eq(heartbeatRuns.status, "queued"), + eq(companies.status, "active"), + )); const agentIds = [...new Set(queuedRuns.map((r) => r.agentId))]; for (const agentId of agentIds) { @@ -9005,6 +9012,44 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) const agent = await getAgent(agentId); if (!agent) throw notFound("Agent not found"); + + const writeSkippedRequest = async ( + skipReason: string, + patch: Partial = {}, + ) => { + await db.insert(agentWakeupRequests).values({ + companyId: agent.companyId, + agentId, + source, + triggerDetail, + reason: skipReason, + payload, + status: "skipped", + requestedByActorType: opts.requestedByActorType ?? null, + requestedByActorId: opts.requestedByActorId ?? null, + idempotencyKey: opts.idempotencyKey ?? null, + finishedAt: new Date(), + ...patch, + }); + }; + + const company = await db + .select({ status: companies.status }) + .from(companies) + .where(eq(companies.id, agent.companyId)) + .then((rows) => rows[0] ?? null); + + if (!company || company.status !== "active") { + const companyStatus = company?.status ?? "missing"; + if (opts.requestedByActorType === "user") { + throw conflict("Company is not active", { status: companyStatus }); + } + await writeSkippedRequest("company.inactive", { + error: `Wake suppressed because company status is ${companyStatus}`, + }); + return null; + } + const explicitResumeSession = await resolveExplicitResumeSessionOverride(agent, payload, taskKey); if (explicitResumeSession) { enrichedContextSnapshot.resumeFromRunId = explicitResumeSession.resumeFromRunId; @@ -9027,22 +9072,6 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) await resolveSessionBeforeForWakeup(agent, effectiveTaskKey); const continuationAttempt = readContinuationAttempt(enrichedContextSnapshot.livenessContinuationAttempt); - const writeSkippedRequest = async (skipReason: string) => { - await db.insert(agentWakeupRequests).values({ - companyId: agent.companyId, - agentId, - source, - triggerDetail, - reason: skipReason, - payload, - status: "skipped", - requestedByActorType: opts.requestedByActorType ?? null, - requestedByActorId: opts.requestedByActorId ?? null, - idempotencyKey: opts.idempotencyKey ?? null, - finishedAt: new Date(), - }); - }; - let projectId = readNonEmptyString(enrichedContextSnapshot.projectId); if (!projectId && issueId) { // Look up by either UUID or identifier (e.g. "ENV-13"), but always scope @@ -10195,7 +10224,11 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) buildRunOutputSilence, tickTimers: async (now = new Date()) => { - const allAgents = await db.select().from(agents); + const allAgents = await db + .select({ ...getTableColumns(agents) }) + .from(agents) + .innerJoin(companies, eq(companies.id, agents.companyId)) + .where(eq(companies.status, "active")); let checked = 0; let enqueued = 0; let skipped = 0; @@ -10235,9 +10268,9 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) }; }, - cancelRun: (runId: string) => cancelRunInternal(runId), + cancelRun: (runId: string, reason?: string) => cancelRunInternal(runId, reason), - cancelActiveForAgent: (agentId: string) => cancelActiveForAgentInternal(agentId), + cancelActiveForAgent: (agentId: string, reason?: string) => cancelActiveForAgentInternal(agentId, reason), cancelBudgetScopeWork, diff --git a/ui/src/lib/activity-format.ts b/ui/src/lib/activity-format.ts index ff09c225..aa89611e 100644 --- a/ui/src/lib/activity-format.ts +++ b/ui/src/lib/activity-format.ts @@ -78,6 +78,7 @@ const ACTIVITY_ROW_VERBS: Record = { "company.created": "created company", "company.updated": "updated company", "company.archived": "archived", + "company.reactivated": "reactivated", "company.budget_updated": "updated budget for", };