diff --git a/server/src/__tests__/heartbeat-opencode-k8s-timer-no-work.test.ts b/server/src/__tests__/heartbeat-opencode-k8s-timer-no-work.test.ts new file mode 100644 index 00000000000..9e7f745e311 --- /dev/null +++ b/server/src/__tests__/heartbeat-opencode-k8s-timer-no-work.test.ts @@ -0,0 +1,207 @@ +import { randomUUID } from "node:crypto"; +import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; +import { eq } from "drizzle-orm"; +import { + agents, + agentRuntimeState, + agentWakeupRequests, + companySkills, + companies, + createDb, + heartbeatRunEvents, + 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 opencode_k8s timer no-work tests on this host: ${embeddedPostgresSupport.reason ?? "unsupported environment"}`, + ); +} + +describeEmbeddedPostgres("opencode_k8s timer no-work suppression", () => { + let db!: ReturnType; + let tempDb: Awaited> | null = null; + + beforeAll(async () => { + tempDb = await startEmbeddedPostgresTestDatabase("heartbeat-opencode-k8s-timer-no-work-"); + db = createDb(tempDb.connectionString); + }); + + afterEach(async () => { + await db.delete(heartbeatRunEvents); + await db.delete(issues); + await db.delete(heartbeatRuns); + await db.delete(agentWakeupRequests); + await db.delete(agentRuntimeState); + await db.delete(companySkills); + await db.delete(agents); + await db.delete(companies); + }); + + afterAll(async () => { + await tempDb?.cleanup(); + }); + + async function seedOpencodeK8sTimerAgent(input: { + companyId: string; + agentId: string; + lastHeartbeatAt: Date; + }) { + const issuePrefix = `T${input.companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`; + + await db.insert(companies).values({ + id: input.companyId, + name: "Paperclip", + issuePrefix, + requireBoardApprovalForNewAgents: false, + }); + + await db.insert(agents).values({ + id: input.agentId, + companyId: input.companyId, + name: "Staff Engineer", + role: "engineer", + status: "active", + adapterType: "opencode_k8s", + adapterConfig: {}, + runtimeConfig: { + heartbeat: { + enabled: true, + intervalSec: 60, + wakeOnDemand: true, + maxConcurrentRuns: 1, + }, + }, + permissions: {}, + lastHeartbeatAt: input.lastHeartbeatAt, + createdAt: input.lastHeartbeatAt, + updatedAt: input.lastHeartbeatAt, + }); + + return { issuePrefix }; + } + + async function saturateAgentConcurrency(input: { + companyId: string; + agentId: string; + now: Date; + }) { + await db.insert(heartbeatRuns).values({ + id: randomUUID(), + companyId: input.companyId, + agentId: input.agentId, + invocationSource: "assignment", + triggerDetail: "system", + status: "running", + contextSnapshot: { + taskKey: `issue:${randomUUID()}`, + wakeReason: "test_busy_slot", + }, + startedAt: input.now, + updatedAt: input.now, + createdAt: input.now, + }); + } + + it("skips opencode_k8s timer ticks when the agent has no assigned live work", async () => { + const companyId = randomUUID(); + const agentId = randomUUID(); + const now = new Date("2026-05-25T20:30:00.000Z"); + const heartbeat = heartbeatService(db); + + await seedOpencodeK8sTimerAgent({ + companyId, + agentId, + lastHeartbeatAt: new Date("2026-05-25T20:28:00.000Z"), + }); + await saturateAgentConcurrency({ companyId, agentId, now }); + + const result = await heartbeat.tickTimers(now); + + expect(result).toMatchObject({ checked: 1, enqueued: 0, skipped: 1 }); + + const runs = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.agentId, agentId)); + expect(runs).toHaveLength(1); + expect(runs[0]?.contextSnapshot).toMatchObject({ wakeReason: "test_busy_slot" }); + + const wakeups = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.agentId, agentId)); + expect(wakeups).toHaveLength(1); + expect(wakeups[0]).toMatchObject({ + source: "timer", + triggerDetail: "system", + reason: "no_in_flight_work", + status: "skipped", + }); + + const agent = await db + .select({ lastHeartbeatAt: agents.lastHeartbeatAt }) + .from(agents) + .where(eq(agents.id, agentId)) + .then((rows) => rows[0]); + expect(agent?.lastHeartbeatAt?.toISOString()).toBe(now.toISOString()); + + const immediateRetry = await heartbeat.tickTimers(new Date("2026-05-25T20:30:10.000Z")); + expect(immediateRetry).toMatchObject({ checked: 1, enqueued: 0, skipped: 0 }); + + const wakeupsAfterImmediateRetry = await db + .select() + .from(agentWakeupRequests) + .where(eq(agentWakeupRequests.agentId, agentId)); + expect(wakeupsAfterImmediateRetry).toHaveLength(1); + }); + + it("queues opencode_k8s timer ticks when the agent has assigned live work", async () => { + const companyId = randomUUID(); + const agentId = randomUUID(); + const issueId = randomUUID(); + const now = new Date("2026-05-25T20:30:00.000Z"); + const heartbeat = heartbeatService(db); + const { issuePrefix } = await seedOpencodeK8sTimerAgent({ + companyId, + agentId, + lastHeartbeatAt: new Date("2026-05-25T20:28:00.000Z"), + }); + await saturateAgentConcurrency({ companyId, agentId, now }); + + await db.insert(issues).values({ + id: issueId, + companyId, + title: "Actionable work", + status: "todo", + priority: "medium", + assigneeAgentId: agentId, + issueNumber: 1, + identifier: `${issuePrefix}-1`, + }); + + const result = await heartbeat.tickTimers(now); + + expect(result).toMatchObject({ checked: 1, enqueued: 1, skipped: 0 }); + + const runs = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.agentId, agentId)); + expect(runs).toHaveLength(2); + const timerRun = runs.find((run) => run.invocationSource === "timer"); + expect(timerRun).toMatchObject({ + invocationSource: "timer", + triggerDetail: "system", + status: "queued", + }); + + const wakeups = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.agentId, agentId)); + expect(wakeups).toHaveLength(1); + expect(wakeups[0]).toMatchObject({ + source: "timer", + reason: "heartbeat_timer", + status: "queued", + }); + }); +}); diff --git a/server/src/__tests__/heartbeat-retry-scheduling.test.ts b/server/src/__tests__/heartbeat-retry-scheduling.test.ts index ce30a0e184a..e097fe19f8e 100644 --- a/server/src/__tests__/heartbeat-retry-scheduling.test.ts +++ b/server/src/__tests__/heartbeat-retry-scheduling.test.ts @@ -19,6 +19,7 @@ import { import { BOUNDED_TRANSIENT_HEARTBEAT_RETRY_DELAYS_MS, heartbeatService, + shouldScheduleAutomaticRunRetry, } from "../services/heartbeat.ts"; const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); @@ -219,6 +220,56 @@ describeEmbeddedPostgres("heartbeat bounded retry scheduling", () => { expect(promotedRun?.status).toBe("queued"); }); + it("treats idempotent GitHub PR-review adapter failures as retry-eligible", () => { + expect( + shouldScheduleAutomaticRunRetry({ + errorCode: "adapter_failed", + resultJson: {}, + contextSnapshot: { + wakeReason: "github_pr_opened", + reviewKind: "pr_review", + githubPrNumber: 976, + }, + }), + ).toBe(true); + + expect( + shouldScheduleAutomaticRunRetry({ + errorCode: "process_lost", + resultJson: {}, + contextSnapshot: { + wakeReason: "github_pr_review_submitted", + reviewKind: "pr_review", + githubPrNumber: 976, + }, + }), + ).toBe(true); + }); + + it("does not retry plain adapter failures when the wake is not an idempotent PR review", () => { + expect( + shouldScheduleAutomaticRunRetry({ + errorCode: "adapter_failed", + resultJson: {}, + contextSnapshot: { + issueId: randomUUID(), + wakeReason: "issue_assigned", + }, + }), + ).toBe(false); + + expect( + shouldScheduleAutomaticRunRetry({ + errorCode: "process_lost", + resultJson: {}, + contextSnapshot: { + issueId: randomUUID(), + wakeReason: "issue_assigned", + }, + }), + ).toBe(false); + }); + it("does not defer a new assignee behind the previous assignee's scheduled retry", async () => { const companyId = randomUUID(); const oldAgentId = randomUUID(); diff --git a/server/src/__tests__/issue-stale-execution-lock-routes.test.ts b/server/src/__tests__/issue-stale-execution-lock-routes.test.ts index 0f40d210783..0465170912d 100644 --- a/server/src/__tests__/issue-stale-execution-lock-routes.test.ts +++ b/server/src/__tests__/issue-stale-execution-lock-routes.test.ts @@ -66,11 +66,12 @@ describeEmbeddedPostgres("stale issue execution lock routes", () => { return app; } - async function seedCompanyAgentAndRuns() { + async function seedCompanyAgentAndRuns(options: { staleRunStatus?: string } = {}) { const companyId = randomUUID(); const agentId = randomUUID(); const failedRunId = randomUUID(); const currentRunId = randomUUID(); + const staleRunStatus = options.staleRunStatus ?? "failed"; await db.insert(companies).values({ id: companyId, @@ -94,7 +95,7 @@ describeEmbeddedPostgres("stale issue execution lock routes", () => { id: failedRunId, companyId, agentId, - status: "failed", + status: staleRunStatus, invocationSource: "manual", finishedAt: new Date(), }, @@ -171,6 +172,73 @@ describeEmbeddedPostgres("stale issue execution lock routes", () => { }); }); + it("allows a same-agent current run to close an issue owned by a stale adapter_failed checkout run", async () => { + const { companyId, agentId, failedRunId, currentRunId } = await seedCompanyAgentAndRuns({ + staleRunStatus: "adapter_failed", + }); + const issueId = randomUUID(); + await db.insert(issues).values({ + id: issueId, + companyId, + title: "Routine close after adapter wedge", + status: "in_progress", + priority: "high", + assigneeAgentId: agentId, + checkoutRunId: failedRunId, + executionRunId: failedRunId, + executionAgentNameKey: "codexcoder", + executionLockedAt: new Date(), + }); + + const res = await request(createApp(agentActor(companyId, agentId, currentRunId))) + .patch(`/api/issues/${issueId}`) + .send({ status: "done" }); + + expect(res.status, JSON.stringify(res.body)).toBe(200); + expect(res.body.status).toBe("done"); + + const row = await db + .select({ + status: issues.status, + checkoutRunId: issues.checkoutRunId, + executionRunId: issues.executionRunId, + }) + .from(issues) + .where(eq(issues.id, issueId)) + .then((rows) => rows[0]); + expect(row).toEqual({ + status: "done", + checkoutRunId: null, + executionRunId: null, + }); + }); + + it("keeps live different-run ownership protected", async () => { + const { companyId, agentId, failedRunId, currentRunId } = await seedCompanyAgentAndRuns({ + staleRunStatus: "running", + }); + const issueId = randomUUID(); + await db.insert(issues).values({ + id: issueId, + companyId, + title: "Live run conflict", + status: "in_progress", + priority: "high", + assigneeAgentId: agentId, + checkoutRunId: failedRunId, + executionRunId: failedRunId, + executionAgentNameKey: "codexcoder", + executionLockedAt: new Date(), + }); + + const res = await request(createApp(agentActor(companyId, agentId, currentRunId))) + .patch(`/api/issues/${issueId}`) + .send({ status: "done" }); + + expect(res.status, JSON.stringify(res.body)).toBe(409); + expect(res.body.error).toBe("Issue run ownership conflict"); + }); + it("allows the rightful assignee to release after the owning run failed", async () => { const { companyId, agentId, failedRunId, currentRunId } = await seedCompanyAgentAndRuns(); const issueId = randomUUID(); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index d831b045fdf..e33291ff621 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -372,6 +372,17 @@ function readTransientRecoveryContractFromRun( return null; } +export function shouldScheduleAutomaticRunRetry( + run: Pick, +) { + if (readTransientRecoveryContractFromRun(run)) return true; + + if (run.errorCode !== "adapter_failed" && run.errorCode !== "process_lost") return false; + + const prReview = derivePaperclipPrReview(parseObject(run.contextSnapshot)); + return prReview?.reviewKind === "pr_review"; +} + function mergeAdapterRecoveryMetadata(input: { resultJson: Record | null | undefined; errorFamily?: string | null; @@ -9322,7 +9333,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) }, }); } - } else if (outcome === "failed" && readTransientRecoveryContractFromRun(livenessRun)) { + } else if (outcome === "failed" && shouldScheduleAutomaticRunRetry(livenessRun)) { await scheduleBoundedRetryForRun(livenessRun, agent); } const issueCommentPolicyResult = await finalizeIssueCommentPolicy(livenessRun, agent); @@ -11421,6 +11432,49 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) let skipped = 0; let idleSkipped = 0; + const writeNoInFlightWorkSkip = async (agent: typeof agents.$inferSelect) => { + await db.insert(agentWakeupRequests).values({ + companyId: agent.companyId, + agentId: agent.id, + source: "timer", + triggerDetail: "system", + reason: "no_in_flight_work", + payload: { + skipped: "no_in_flight_work", + assignedLiveIssueCount: 0, + }, + status: "skipped", + requestedByActorType: "system", + requestedByActorId: "heartbeat_scheduler", + finishedAt: now, + }); + + await db + .update(agents) + .set({ lastHeartbeatAt: now, updatedAt: now }) + .where(eq(agents.id, agent.id)); + }; + + const opencodeK8sAgentIds = allAgents + .filter((agent) => agent.adapterType === "opencode_k8s") + .map((agent) => agent.id); + const assignedLiveWorkAgentIds = new Set(); + if (opencodeK8sAgentIds.length > 0) { + const assignedLiveWorkRows = await db + .select({ agentId: issues.assigneeAgentId }) + .from(issues) + .where( + and( + inArray(issues.assigneeAgentId, opencodeK8sAgentIds), + inArray(issues.status, ["todo", "in_progress", "in_review"]), + isNull(issues.hiddenAt), + ), + ); + for (const row of assignedLiveWorkRows) { + if (row.agentId) assignedLiveWorkAgentIds.add(row.agentId); + } + } + for (const agent of allAgents) { if (agent.status === "paused" || agent.status === "terminated" || agent.status === "pending_approval") continue; const policy = parseHeartbeatPolicy(agent); @@ -11448,6 +11502,16 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) } } + if (agent.adapterType === "opencode_k8s" && !assignedLiveWorkAgentIds.has(agent.id)) { + await writeNoInFlightWorkSkip(agent); + logger.info( + { agentId: agent.id, agentName: agent.name, adapterType: agent.adapterType, skipped: "no_in_flight_work" }, + "opencode_k8s timer wakeup skipped because agent has no assigned live work", + ); + skipped += 1; + continue; + } + const run = await enqueueWakeup(agent.id, { source: "timer", triggerDetail: "system", diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index a34de23c2ee..d436e1fd47c 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -387,7 +387,14 @@ function sameRunLock(checkoutRunId: string | null, actorRunId: string | null) { return checkoutRunId == null; } -const TERMINAL_HEARTBEAT_RUN_STATUSES = new Set(["succeeded", "failed", "cancelled", "timed_out"]); +const TERMINAL_HEARTBEAT_RUN_STATUSES = new Set([ + "succeeded", + "failed", + "error", + "adapter_failed", + "cancelled", + "timed_out", +]); const ISSUE_LIST_DESCRIPTION_MAX_CHARS = 1200; function escapeLikePattern(value: string): string { diff --git a/server/src/services/recovery/successful-run-handoff.test.ts b/server/src/services/recovery/successful-run-handoff.test.ts index 77e950f7fbf..46fa801f45c 100644 --- a/server/src/services/recovery/successful-run-handoff.test.ts +++ b/server/src/services/recovery/successful-run-handoff.test.ts @@ -168,6 +168,26 @@ describe("successful run handoff decision", () => { }); }); + it("does not re-run a successful GitHub PR review after only the issue disposition write failed", () => { + expect(decide({ + run: { + ...run, + contextSnapshot: { + issueId: "issue-1", + taskId: "issue-1", + wakeReason: "github_pr_review_requested", + reviewKind: "pr_review", + githubRepoFullName: "Blockcast/linux-amt", + githubPrNumber: 51, + githubHeadSha: "54900568", + }, + } as any, + })).toEqual({ + kind: "skip", + reason: "successful PR review run already may have emitted an external side effect", + }); + }); + it("does not queue for issue monitor maintenance runs", () => { expect(decide({ run: { diff --git a/server/src/services/recovery/successful-run-handoff.ts b/server/src/services/recovery/successful-run-handoff.ts index 3332501876f..59d7ed4f8d3 100644 --- a/server/src/services/recovery/successful-run-handoff.ts +++ b/server/src/services/recovery/successful-run-handoff.ts @@ -295,6 +295,12 @@ function isIssueMonitorMaintenanceRun(run: HeartbeatRunRow) { return Boolean(wakeReason?.startsWith("issue_monitor") || source?.startsWith("issue.monitor")); } +function isGithubPrReviewRun(run: HeartbeatRunRow) { + const context = readRecord(run.contextSnapshot); + const reviewKind = readString(context.reviewKind); + return reviewKind === "pr_review"; +} + function isProductiveSuccessfulRun(input: { livenessState: RunLivenessState | null; detectedProgressSummary: string | null; @@ -350,6 +356,9 @@ export function decideSuccessfulRunHandoff(input: { if (run.status !== "succeeded") return { kind: "skip", reason: "source run did not succeed" }; if (isCorrectiveHandoffRun(run)) return { kind: "skip", reason: "source run is already a corrective handoff run" }; if (isIssueMonitorMaintenanceRun(run)) return { kind: "skip", reason: "issue monitor run owns its own recovery path" }; + if (isGithubPrReviewRun(run)) { + return { kind: "skip", reason: "successful PR review run already may have emitted an external side effect" }; + } if (run.issueCommentStatus === "retry_queued" || run.issueCommentStatus === "retry_exhausted") { return { kind: "skip", reason: "missing issue comment retry owns the next action" }; }