diff --git a/server/src/modules/agent-lifecycle/__tests__/driver-scope.test.ts b/server/src/modules/agent-lifecycle/__tests__/driver-scope.test.ts index 2f1999565b7..da514753201 100644 --- a/server/src/modules/agent-lifecycle/__tests__/driver-scope.test.ts +++ b/server/src/modules/agent-lifecycle/__tests__/driver-scope.test.ts @@ -8,10 +8,13 @@ import { createLifecycleDriver } from "../../../services/agent-lifecycle-driver. describe("lifecycle host stop scope", () => { it("pauses only the requested agent without loading its company peers", async () => { - const db = { select: vi.fn(() => { throw new Error("No peer lookup expected"); }) } as unknown as Db; + // The pause stamps its cause on the agent's own runs (one update), never a peer lookup. + const update = vi.fn(() => ({ set: () => ({ where: async () => [] }) })); + const db = { select: vi.fn(() => { throw new Error("No peer lookup expected"); }), update } as unknown as Db; const driver = createLifecycleDriver(db, {} as never); - expect(await driver.runHost({ id: "target", companyId: "company", lifecycleState: "pausing", - lifecycleVersion: 2, lifecycleRequiredPluginIds: [], lifecycleError: null, lifecycleOperation: null })).toBe("complete"); + expect(await driver.runHost({ id: "target", companyId: "company", lifecycleState: "pausing", lifecycleHolds: ["manual"], + lifecycleVersion: 2, lifecycleRequiredPluginIds: [], lifecycleError: null, lifecycleOperation: null } as never)).toBe("complete"); + expect(update).toHaveBeenCalledOnce(); // A pause keeps the messages it deferred (work-intent outbox). expect(stop).toHaveBeenCalledWith(["target"], "Agent lifecycle stop requested", { preservePauseDeferred: true }); expect(db.select).not.toHaveBeenCalled(); diff --git a/server/src/modules/agent-lifecycle/__tests__/lifecycle.test.ts b/server/src/modules/agent-lifecycle/__tests__/lifecycle.test.ts index e55f627582a..ea0b6ba6da7 100644 --- a/server/src/modules/agent-lifecycle/__tests__/lifecycle.test.ts +++ b/server/src/modules/agent-lifecycle/__tests__/lifecycle.test.ts @@ -371,6 +371,19 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu expect(await db.select().from(principalPermissionGrants).where(eq(principalPermissionGrants.companyId, companyId))).toHaveLength(0); }); + // H1 bug B: a pause's stop is told apart from termination by the cause the + // pause records on each run it stops. + it("records what a pause stopped each active run for", async () => { + const agent = await hire(); + await db.update(companies).set({ status: "paused", pauseReason: "budget" }).where(eq(companies.id, companyId)); + await db.update(agents).set({ lifecycleState: "pausing", lifecycleHolds: ["company_paused"] }).where(eq(agents.id, agent.id)); + const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId: agent.id, status: "queued" }).returning(); + await createLifecycleDriver(db, {} as never).runHost(await current(agent.id) as never); + const [stopped] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)); + expect(stopped.status).toBe("cancelled"); + expect(stopped.resultJson).toMatchObject({ pauseStop: { holds: ["company_paused"], companyPauseReason: "budget" } }); + }); + it("runs the saved harness test and keeps a failed configuration out of ready", async () => { const driver = createLifecycleDriver(db, {} as never); const work = worker(driver.runHost); diff --git a/server/src/modules/wake-queue/adapters/postgres.test.ts b/server/src/modules/wake-queue/adapters/postgres.test.ts index 831d8228d3d..761624744fa 100644 --- a/server/src/modules/wake-queue/adapters/postgres.test.ts +++ b/server/src/modules/wake-queue/adapters/postgres.test.ts @@ -3,7 +3,7 @@ import { DELIVERY_QUEUES } from "../../../services/delivery-work-notifications.j import { instanceSettingsService } from "../../../services/instance-settings.js"; import { randomBytes, randomUUID } from "node:crypto"; import { and, eq } from "drizzle-orm"; -import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; +import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; import type { Db } from "@paperclipai/db"; import { activityLog, @@ -754,6 +754,94 @@ describeEmbeddedPostgres("wake-queue postgres adapter", () => { expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wakeId)))[0].status).toBe("deferred_issue_execution"); }); + // H1 bug B: a budget pause stopped the agent's run, and the release blocked + // the task behind a reconciliation incident with no control. A pause is an + // intentional stop; resuming continues the task. + it.each(["Agent lifecycle stop requested", "Cancelled due to budget pause", "Cancelled due to agent pause"])( + "keeps a native task's status and opens no incident when a pause stopped its run (%s)", async (reason) => { + const companyId = await seedCompany(); + const agentId = await seedAgent({ companyId }); + const issueId = await seedIssue({ companyId, assigneeAgentId: agentId, status: "in_progress" }); + const runId = await seedRun({ companyId, agentId, status: "cancelled", contextSnapshot: { issueId } }); + await db.update(heartbeatRuns).set({ runtimeMode: "native", nativeIssueId: issueId, errorCode: "cancelled", + resultJson: { cancellation: { reason, source: "control_plane", expected: true, initiator: { type: "system" }, + recordedAt: new Date().toISOString() }, + ...(reason === "Agent lifecycle stop requested" ? { pauseStop: { holds: ["budget"], companyPauseReason: null } } : {}) } }).where(eq(heartbeatRuns.id, runId)); + await db.update(issues).set({ executionRunId: runId, checkoutRunId: runId }).where(eq(issues.id, issueId)); + const adapter = createPostgresWakeQueueAdapter(db, stubDeps); + await adapter.withIssueExecutionLock({ companyId, runId, now: new Date() }, async () => undefined as never); + expect((await db.select().from(issues).where(eq(issues.id, issueId)))[0]).toMatchObject({ status: "in_progress" }); + expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, issueId))).toHaveLength(0); + }); + + // Review: the full release (createReleaseIssueExecution), not only the + // native incident step, must leave a paused agent's task unblocked. + it.each(["native", "legacy"] as const)("keeps a %s task unblocked through the full release when its agent's pause stopped the run", async (runtimeMode) => { + const companyId = await seedCompany(); + const agentId = await seedAgent({ companyId }); + await db.update(agents).set({ status: "paused", lifecycleState: "pausing", lifecycleHolds: ["manual"] }).where(eq(agents.id, agentId)); + const issueId = await seedIssue({ companyId, assigneeAgentId: agentId, status: "in_progress" }); + const runId = await seedRun({ companyId, agentId, status: "cancelled", contextSnapshot: { issueId } }); + await db.update(heartbeatRuns).set({ runtimeMode, ...(runtimeMode === "native" ? { nativeIssueId: issueId } : {}), + errorCode: "cancelled", resultJson: { cancellation: { reason: "Agent lifecycle stop requested", source: "control_plane", + expected: true, initiator: { type: "system" } }, pauseStop: { holds: ["manual"], companyPauseReason: null } } }) + .where(eq(heartbeatRuns.id, runId)); + await db.update(issues).set({ executionRunId: runId, checkoutRunId: runId }).where(eq(issues.id, issueId)); + const escalateStrandedAssignedIssue = vi.fn(async () => undefined); + const release = createReleaseIssueExecution({ issueLock: createPostgresWakeQueueAdapter(db, stubDeps), + recovery: { escalateStrandedAssignedIssue, escalateStrandedRecoveryIssueInPlace: async () => undefined }, + }); + await release({ companyId, runId, now: new Date() }); + expect((await db.select().from(issues).where(eq(issues.id, issueId)))[0]).toMatchObject({ status: "in_progress" }); + expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, issueId))).toHaveLength(0); + expect(escalateStrandedAssignedIssue).not.toHaveBeenCalled(); + }); + + it("still opens a reconciliation incident when termination replaced the pause that stamped the run", async () => { + const companyId = await seedCompany(); + const agentId = await seedAgent({ companyId }); + await db.update(agents).set({ lifecycleState: "terminating" }).where(eq(agents.id, agentId)); + const issueId = await seedIssue({ companyId, assigneeAgentId: agentId, status: "in_progress" }); + const runId = await seedRun({ companyId, agentId, status: "cancelled", contextSnapshot: { issueId } }); + await db.update(heartbeatRuns).set({ runtimeMode: "native", nativeIssueId: issueId, errorCode: "cancelled", + resultJson: { cancellation: { reason: "Agent lifecycle stop requested", source: "control_plane", expected: true, + initiator: { type: "system" } }, pauseStop: { holds: ["budget"], companyPauseReason: null } } }) + .where(eq(heartbeatRuns.id, runId)); + const adapter = createPostgresWakeQueueAdapter(db, stubDeps); + await adapter.withIssueExecutionLock({ companyId, runId, now: new Date() }, async () => undefined as never); + expect((await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, issueId)))[0]) + .toMatchObject({ cause: "native_continuation_requires_reconciliation" }); + }); + + it("still opens a reconciliation incident when agent termination stopped the run", async () => { + // Termination and cleanup use the lifecycle stop reason too, with no pauseStop. + const companyId = await seedCompany(); + const agentId = await seedAgent({ companyId }); + const issueId = await seedIssue({ companyId, assigneeAgentId: agentId, status: "in_progress" }); + const runId = await seedRun({ companyId, agentId, status: "cancelled", contextSnapshot: { issueId } }); + await db.update(heartbeatRuns).set({ runtimeMode: "native", nativeIssueId: issueId, errorCode: "cancelled", + resultJson: { cancellation: { reason: "Agent lifecycle stop requested", source: "control_plane", + expected: true, initiator: { type: "system" } } } }).where(eq(heartbeatRuns.id, runId)); + const adapter = createPostgresWakeQueueAdapter(db, stubDeps); + await adapter.withIssueExecutionLock({ companyId, runId, now: new Date() }, async () => undefined as never); + expect((await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, issueId)))[0]) + .toMatchObject({ cause: "native_continuation_requires_reconciliation" }); + }); + + it("still opens a reconciliation incident for a native run cancelled by another system stop", async () => { + const companyId = await seedCompany(); + const agentId = await seedAgent({ companyId }); + const issueId = await seedIssue({ companyId, assigneeAgentId: agentId, status: "in_progress" }); + const runId = await seedRun({ companyId, agentId, status: "cancelled", contextSnapshot: { issueId } }); + await db.update(heartbeatRuns).set({ runtimeMode: "native", nativeIssueId: issueId, errorCode: "cancelled", + resultJson: { cancellation: { reason: "Cancelled because the company was archived", source: "control_plane", + expected: true, initiator: { type: "system" } } } }).where(eq(heartbeatRuns.id, runId)); + const adapter = createPostgresWakeQueueAdapter(db, stubDeps); + await adapter.withIssueExecutionLock({ companyId, runId, now: new Date() }, async () => undefined as never); + expect((await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, issueId)))[0]) + .toMatchObject({ cause: "native_continuation_requires_reconciliation" }); + }); + it.each(["in_progress", "blocked"])("preserves recovery ownership and queued messages when a native task fails from %s", async (status) => { const companyId = await seedCompany(); const agentId = await seedAgent({ companyId }); diff --git a/server/src/modules/wake-queue/adapters/postgres.ts b/server/src/modules/wake-queue/adapters/postgres.ts index c3bbf5219a8..d59c9340a0a 100644 --- a/server/src/modules/wake-queue/adapters/postgres.ts +++ b/server/src/modules/wake-queue/adapters/postgres.ts @@ -17,6 +17,7 @@ import { getExecutionBlocker, hasPendingRunCleanup } from "../../../services/exe import { nativeManualAdmissionHoldsExecution } from "../../../services/native-runtime/native-runner-ownership.js"; import { and, asc, eq, inArray, isNull, notInArray, or, sql } from "drizzle-orm"; import { readReleaseRun, type ReleaseRun } from "./release-run-read.js"; +import { isGovernancePauseStop } from "../../../services/governance-pause-stop.js"; import type { Db } from "@paperclipai/db"; import { deliveredQueuedCommentIds } from "../../../services/queued-comment-delivery.js"; import { extractIssueReferenceIdentifiers, isUuidLike } from "@paperclipai/shared"; @@ -124,6 +125,9 @@ export function isOperatorStoppedRunRow(row: Pick = new Set([ + AGENT_LIFECYCLE_STOP_REASON, BUDGET_PAUSE_STOP_REASON, AGENT_PAUSE_STOP_REASON]); + +const record = (value: unknown): Record => + value !== null && typeof value === "object" && !Array.isArray(value) ? value as Record : {}; + +export function isGovernancePauseStop(run: { status: string; resultJson?: Record | null }): boolean { + if (run.status !== "cancelled") return false; + const cancellation = record(run.resultJson?.cancellation); + if (record(cancellation.initiator).type !== "system" || cancellation.expected !== true + || typeof cancellation.reason !== "string" || !GOVERNANCE_PAUSE_STOP_REASONS.has(cancellation.reason)) return false; + // Termination and cleanup issue the same lifecycle stop reason. Only the + // pausing path records pauseStop, so require it for that reason. + return cancellation.reason !== AGENT_LIFECYCLE_STOP_REASON + || Object.keys(record(run.resultJson?.pauseStop)).length > 0; +}