Arnike/paperclip
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: Stop archived companies from waking agents (#7478)

Browse Source 
## 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.
This commit is contained in:
Devin Foley
2026-06-03 21:07:14 -07:00
committed by GitHub
parent 62863126a3
commit 93206f73fa
10 changed files with 1247 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/src/services/budgets.ts
+2 -1
View File
@@ -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;
server/src/services/companies.ts
+172 -15
View File
@@ -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<Parameters<typeof db.transaction>[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<typeof companies.$inferInsert> & { 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) => {
server/src/services/heartbeat.ts
+53 -20
View File
@@ -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<typeof agentWakeupRequests.$inferInsert> = {},
) => {
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,