diff --git a/server/src/__tests__/budgets-service.test.ts b/server/src/__tests__/budgets-service.test.ts index 1276e8f8b4a..02dca85e4c7 100644 --- a/server/src/__tests__/budgets-service.test.ts +++ b/server/src/__tests__/budgets-service.test.ts @@ -27,6 +27,7 @@ const mockLogActivity = vi.hoisted(() => vi.fn()); vi.mock("../services/activity-log.js", () => ({ logActivity: mockLogActivity, + persistActivity: vi.fn(async () => ({ activity: null, publication: null })), })); // Budget behavior is exercised against PostgreSQL below and in diff --git a/server/src/__tests__/heartbeat-native-restart-suspension.test.ts b/server/src/__tests__/heartbeat-native-restart-suspension.test.ts index 9ac9b952ed1..edf2f07afeb 100644 --- a/server/src/__tests__/heartbeat-native-restart-suspension.test.ts +++ b/server/src/__tests__/heartbeat-native-restart-suspension.test.ts @@ -372,6 +372,22 @@ describeEmbeddedPostgres("heartbeat native recovery interrupted by graceful shut .where(eq(heartbeatRuns.nativeIssueId, f.issueId))).toHaveLength(1); }, 90_000); + // H1 bug A: a budget pause cancelled the run 0.36 s after its result was + // accepted, then a hard kill took the controller. The dead controller's + // result lease held finalization, and with it the agent's pause, for the + // lease's 20 minutes. + it("finalizes an accepted result whose run a pause cancelled before a hard kill took its controller", async () => { + const f = await seedAcceptedResult("ACCEPTED-CANCELLED-THEN-SIGKILL", (runner) => `${runner}:${randomUUID()}`); + await killedController(f, deadController(currentProcessSpaceHash())); + await f.db.update(heartbeatRuns).set({ status: "cancelled", finishedAt: new Date(), errorCode: "cancelled" }) + .where(eq(heartbeatRuns.id, f.runId)); + const { owner } = await restart(f); + // The agent's pause waits only while an accepted result stays leased + // (agentExecutionsHaveStopped); finalization has now released it. + expect(owner.leaseOwner).toBeNull(); + expect(owner.phase).not.toBe("workspace_finalizing"); + }, 90_000); + it("releases a dead controller's lease whose expiry Postgres stored with microsecond precision", async () => { // The controller writes its lease with SQL time arithmetic, so the stored // expiry carries microseconds that a JavaScript Date cannot round-trip. diff --git a/server/src/services/agent-budget-hold-activity.test.ts b/server/src/services/agent-budget-hold-activity.test.ts new file mode 100644 index 00000000000..b4c73e4016a --- /dev/null +++ b/server/src/services/agent-budget-hold-activity.test.ts @@ -0,0 +1,55 @@ +import { and, eq } from "drizzle-orm"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { activityLog, agents, companies, createDb } from "@paperclipai/db"; +import { applyAgentBudgetHoldInTransaction } from "./agent-lifecycle.js"; +import { + EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase, +} from "../__tests__/helpers/embedded-postgres.js"; + +const support = await getEmbeddedPostgresTestSupport(); + +// H1 bug A: a budget hard stop paused the agent with no activity entry and +// no recorded reason. A governance stop must be explainable from the log. +describe.skipIf(!support.supported)("agent budget hold activity", () => { + let db!: ReturnType; + let tempDb: Awaited> | null = null; + beforeAll(async () => { + tempDb = await startEmbeddedPostgresTestDatabase("paperclip-agent-budget-hold-activity-"); + db = createDb(tempDb.connectionString); + }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS); + afterAll(async () => { + await db?.$client.end(); + await tempDb?.cleanup(); + }); + + it("records the budget pause and its release with the reason", async () => { + const [company] = await db.insert(companies).values({ name: "Hold", issuePrefix: "HLD" }).returning(); + const [agent] = await db.insert(agents).values({ companyId: company.id, name: "Worker" }).returning(); + const publications: unknown[] = []; + await db.transaction(tx => applyAgentBudgetHoldInTransaction(tx as never, company.id, agent.id, true, publications as never)); + // The entry is handed to the caller's post-commit publications. + expect(publications).toEqual([expect.objectContaining({ companyId: company.id })]); + const [paused] = await db.select().from(agents).where(eq(agents.id, agent.id)); + expect(paused).toMatchObject({ pauseReason: "budget", lifecycleHolds: ["budget"] }); + const entries = () => db.select({ action: activityLog.action, details: activityLog.details }).from(activityLog) + .where(and(eq(activityLog.companyId, company.id), eq(activityLog.entityId, agent.id))); + expect(await entries()).toEqual([{ action: "agent.paused", + details: expect.objectContaining({ reason: "budget", holds: ["budget"] }) }]); + // Applying the same hold again records nothing new. + await db.transaction(tx => applyAgentBudgetHoldInTransaction(tx as never, company.id, agent.id, true)); + expect(await entries()).toHaveLength(1); + await db.transaction(tx => applyAgentBudgetHoldInTransaction(tx as never, company.id, agent.id, false)); + expect((await entries()).map(entry => entry.action)).toEqual(["agent.paused", "agent.resumed"]); + }); + + it("records a budget hold release, not a resume, while another hold keeps the agent paused", async () => { + const [company] = await db.insert(companies).values({ name: "Held", issuePrefix: "HLE" }).returning(); + const [agent] = await db.insert(agents).values({ companyId: company.id, name: "Worker" }).returning(); + await db.transaction(tx => applyAgentBudgetHoldInTransaction(tx as never, company.id, agent.id, true)); + await db.update(agents).set({ lifecycleHolds: ["budget", "manual"] }).where(eq(agents.id, agent.id)); + await db.transaction(tx => applyAgentBudgetHoldInTransaction(tx as never, company.id, agent.id, false)); + const actions = (await db.select({ action: activityLog.action }).from(activityLog) + .where(and(eq(activityLog.companyId, company.id), eq(activityLog.entityId, agent.id)))).map(row => row.action); + expect(actions).toEqual(["agent.paused", "agent.budget_hold_released"]); + }); +}); diff --git a/server/src/services/agent-lifecycle.ts b/server/src/services/agent-lifecycle.ts index 63a54291ef4..a324b34b72f 100644 --- a/server/src/services/agent-lifecycle.ts +++ b/server/src/services/agent-lifecycle.ts @@ -9,8 +9,9 @@ import { isAgentStatusInvokable } from "@paperclipai/shared"; import { notifyNativeResponsePresentation } from "./native-response-presentation-notifications.js"; import { notifyChatInboundWakeNotices } from "./chat-inbound-wake-notice-notifications.js"; import { logger } from "../middleware/logger.js"; -import { companies, type Db } from "@paperclipai/db"; -import { eq } from "drizzle-orm"; +import { agents, companies, type Db } from "@paperclipai/db"; +import { and, eq } from "drizzle-orm"; +import { persistActivity, type ActivityPublication } from "./activity-log.js"; import * as lifecycle from "../modules/agent-lifecycle/index.js"; import { withAccountingTransaction } from "./accounting-transaction.js"; import { budgetServiceInTransaction, policyBlocks, type BudgetServiceHooks } from "./budgets.js"; @@ -110,8 +111,27 @@ export function startAgentLifecycle(db: Db, driver: lifecycle.LifecycleDriver, c } /** Budget evaluation owns the transaction and supplies the aggregate agent decision. */ -export function applyAgentBudgetHoldInTransaction(tx: Db, companyId: string, agentId: string, blocked: boolean) { - return applyAgentPolicyHoldsInTransaction(tx, createAgentLifecycleEffects(), companyId, agentId, { budget: blocked }); +/** Pass the transaction's post-commit `publications` so the entry reaches live + * subscribers after commit; without them it publishes nothing until reload. */ +export async function applyAgentBudgetHoldInTransaction(tx: Db, companyId: string, agentId: string, blocked: boolean, + publications?: ActivityPublication[]) { + const [before] = await tx.select({ holds: agents.lifecycleHolds }).from(agents) + .where(and(eq(agents.id, agentId), eq(agents.companyId, companyId))); + const updated = await applyAgentPolicyHoldsInTransaction(tx, createAgentLifecycleEffects(), companyId, agentId, { budget: blocked }); + // A budget pause is a governance stop: record it, and its release, with the + // reason, in the transaction that applies it. + const had = before?.holds?.includes("budget") ?? false; + const has = updated?.lifecycleHolds?.includes("budget") ?? had; + if (updated && had !== has) { + const { publication } = await persistActivity(tx, { companyId, actorType: "system", actorId: "budget", + // Releasing the budget hold resumes the agent only when no other hold + // (a manual or company pause) still keeps it paused. + agentId, action: has ? "agent.paused" : updated.lifecycleHolds.length === 0 ? "agent.resumed" : "agent.budget_hold_released", + entityType: "agent", entityId: agentId, + details: { reason: "budget", lifecycleState: updated.lifecycleState, holds: updated.lifecycleHolds } }); + if (publication) publications?.push(publication); + } + return updated; } export function fenceActiveCompanyAgentStopsInTransaction(tx: Db, companyId: string, status: "paused" | "archived") { diff --git a/server/src/services/budgets.ts b/server/src/services/budgets.ts index 135b62e96a2..51160723bad 100644 --- a/server/src/services/budgets.ts +++ b/server/src/services/budgets.ts @@ -349,7 +349,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica const now = new Date(); if (policy.scopeType === "agent") { const { applyAgentBudgetHoldInTransaction } = await import("./agent-lifecycle.js"); - await applyAgentBudgetHoldInTransaction(db, policy.companyId, policy.scopeId, true); + await applyAgentBudgetHoldInTransaction(db, policy.companyId, policy.scopeId, true, publications); return; } @@ -420,7 +420,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica await promoteDeferredWorkIntents(db, { companyId: policy.companyId, trigger: "pause_lifted" }); if (policy.scopeType === "agent") { const { applyAgentBudgetHoldInTransaction } = await import("./agent-lifecycle.js"); - await applyAgentBudgetHoldInTransaction(db, policy.companyId, policy.scopeId, false); + await applyAgentBudgetHoldInTransaction(db, policy.companyId, policy.scopeId, false, publications); return; } diff --git a/server/src/services/native-runtime/native-accepted-result-lease-settlement.ts b/server/src/services/native-runtime/native-accepted-result-lease-settlement.ts index 98b04eb5c29..928fc9dcc5c 100644 --- a/server/src/services/native-runtime/native-accepted-result-lease-settlement.ts +++ b/server/src/services/native-runtime/native-accepted-result-lease-settlement.ts @@ -1,4 +1,4 @@ -import { and, eq, gt, isNull, notLike, sql } from "drizzle-orm"; +import { and, eq, gt, inArray, isNull, notLike, sql } from "drizzle-orm"; import { heartbeatRuns, nativeRunFinalizations, type Db } from "@paperclipai/db"; import { notifyLiveRunRecovery } from "../live-run-recovery-notifications.js"; import { notifyNativeFinalizationWork } from "../native-finalization-notifications.js"; @@ -19,8 +19,9 @@ import { recordLegacyHostFallbackProof } from "./native-restart-recovery.js"; * * The shared reconciler calls this only after `proveOwnerGone` proved the * recorded controller gone. Settle expires exactly that lease: a lease taken - * over since the proof, a run no longer running, and remote copyback (fenced - * by its own controller lease) are left alone. It settles ownership only; the + * over since the proof, a run that already finished (other than a run cancelled + * after its result was accepted), and remote copyback (fenced by its own + * controller lease) are left alone. It settles ownership only; the * accepted result is finalized by the finalizer that the notifications wake. */ export async function settleNativeAcceptedResultLease(db: Db, record: InProgressRecord, @@ -43,7 +44,10 @@ export async function settleNativeAcceptedResultLease(db: Db, record: InProgress eq(heartbeatRuns.companyId, nativeRunFinalizations.companyId), eq(heartbeatRuns.nativeIssueId, nativeRunFinalizations.issueId))) .where(and(eq(nativeRunFinalizations.runId, record.recordId), eq(heartbeatRuns.runtimeMode, "native"), - eq(heartbeatRuns.status, "running"), gt(nativeRunFinalizations.leaseExpiresAt, now), + // A run cancelled after its result was accepted (for example by a + // budget pause) still owes finalization, and the dead lease would + // hold it, and the agent's pause, for the lease's full term. + inArray(heartbeatRuns.status, ["running", "cancelled"]), gt(nativeRunFinalizations.leaseExpiresAt, now), notLike(nativeRunFinalizations.leaseOwner, "native-finalizer:%"), notLike(nativeRunFinalizations.leaseOwner, "native-cleanup:%"))) .for("update", { of: nativeRunFinalizations });