From 636d8626db2f52394da68a907cbf96ae77047e19 Mon Sep 17 00:00:00 2001 From: Dotta Date: Sun, 26 Apr 2026 07:50:07 -0500 Subject: [PATCH] Cancel stale queued heartbeats when issue graph changes Queued runs were promoted to running even after the target issue's ownership or status had moved on, causing agents to re-read and re-test already-completed work after dependents had advanced (bench evidence: pap-2299-sneaker-local-src-20260425-2). claimQueuedRun now performs a staleness check at claim time and cancels the run with a descriptive errorCode when: - the issue assignee no longer matches the run's agent (and the wake is not a tree-control interaction wake like a comment/mention) - the issue reached terminal status (done/cancelled) without an explicit resume intent or follow-up wake comment id - the issue is in_review with an executionState whose currentParticipant is not this agent and the wake is not comment-driven Cancel mirrors cancelQueuedRunForBlockedDependencies: marks the run cancelled, skips the wakeup request, releases the issue execution lock, and emits a lifecycle event. Co-Authored-By: Paperclip --- ...heartbeat-stale-queue-invalidation.test.ts | 466 ++++++++++++++++++ server/src/services/heartbeat.ts | 156 ++++++ 2 files changed, 622 insertions(+) create mode 100644 server/src/__tests__/heartbeat-stale-queue-invalidation.test.ts diff --git a/server/src/__tests__/heartbeat-stale-queue-invalidation.test.ts b/server/src/__tests__/heartbeat-stale-queue-invalidation.test.ts new file mode 100644 index 00000000000..6f57084c47a --- /dev/null +++ b/server/src/__tests__/heartbeat-stale-queue-invalidation.test.ts @@ -0,0 +1,466 @@ +import { randomUUID } from "node:crypto"; +import { eq, sql } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; +import { + activityLog, + agents, + agentRuntimeState, + agentWakeupRequests, + companies, + companySkills, + createDb, + documentRevisions, + documents, + heartbeatRunEvents, + heartbeatRuns, + issueComments, + issueDocuments, + issueRelations, + issueTreeHolds, + issues, +} from "@paperclipai/db"; +import { + getEmbeddedPostgresTestSupport, + startEmbeddedPostgresTestDatabase, +} from "./helpers/embedded-postgres.js"; +import { heartbeatService } from "../services/heartbeat.ts"; +import { runningProcesses } from "../adapters/index.ts"; + +const mockAdapterExecute = vi.hoisted(() => + vi.fn(async () => ({ + exitCode: 0, + signal: null, + timedOut: false, + errorMessage: null, + summary: "Stale-queue invalidation test run.", + provider: "test", + model: "test-model", + })), +); + +vi.mock("../adapters/index.ts", async () => { + const actual = await vi.importActual("../adapters/index.ts"); + return { + ...actual, + getServerAdapter: vi.fn(() => ({ + supportsLocalAgentJwt: false, + execute: mockAdapterExecute, + })), + }; +}); + +const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; + +if (!embeddedPostgresSupport.supported) { + console.warn( + `Skipping embedded Postgres heartbeat stale-queue invalidation tests on this host: ${embeddedPostgresSupport.reason ?? "unsupported environment"}`, + ); +} + +async function ensureIssueRelationsTable(db: ReturnType) { + await db.execute(sql.raw(` + CREATE TABLE IF NOT EXISTS "issue_relations" ( + "id" uuid PRIMARY KEY DEFAULT gen_random_uuid(), + "company_id" uuid NOT NULL, + "issue_id" uuid NOT NULL, + "related_issue_id" uuid NOT NULL, + "type" text NOT NULL, + "created_by_agent_id" uuid, + "created_by_user_id" text, + "created_at" timestamptz NOT NULL DEFAULT now(), + "updated_at" timestamptz NOT NULL DEFAULT now() + ); + `)); +} + +async function waitForCondition(fn: () => Promise, timeoutMs = 3_000) { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (await fn()) return true; + await new Promise((resolve) => setTimeout(resolve, 50)); + } + return fn(); +} + +type SeedOptions = { + agentName?: string; + agentRole?: string; + maxConcurrentRuns?: number; +}; + +type SeedResult = { + companyId: string; + agentId: string; +}; + +describeEmbeddedPostgres("heartbeat stale queued-run invalidation", () => { + let db!: ReturnType; + let heartbeat!: ReturnType; + let tempDb: Awaited> | null = null; + + beforeAll(async () => { + tempDb = await startEmbeddedPostgresTestDatabase("paperclip-heartbeat-stale-queue-"); + db = createDb(tempDb.connectionString); + heartbeat = heartbeatService(db); + await ensureIssueRelationsTable(db); + }, 20_000); + + afterEach(async () => { + mockAdapterExecute.mockReset(); + mockAdapterExecute.mockImplementation(async () => ({ + exitCode: 0, + signal: null, + timedOut: false, + errorMessage: null, + summary: "Stale-queue invalidation test run.", + provider: "test", + model: "test-model", + })); + runningProcesses.clear(); + let idlePolls = 0; + for (let attempt = 0; attempt < 100; attempt += 1) { + const runs = await db + .select({ status: heartbeatRuns.status }) + .from(heartbeatRuns); + const hasActiveRun = runs.some((run) => run.status === "queued" || run.status === "running"); + if (!hasActiveRun) { + idlePolls += 1; + if (idlePolls >= 3) break; + } else { + idlePolls = 0; + } + await new Promise((resolve) => setTimeout(resolve, 50)); + } + await new Promise((resolve) => setTimeout(resolve, 50)); + await db.delete(activityLog); + await db.delete(companySkills); + await db.delete(issueComments); + await db.delete(issueDocuments); + await db.delete(documentRevisions); + await db.delete(documents); + await db.delete(issueRelations); + await db.delete(issueTreeHolds); + await db.delete(issues); + await db.delete(heartbeatRunEvents); + await db.delete(activityLog); + await db.delete(heartbeatRuns); + await db.delete(agentWakeupRequests); + await db.delete(agentRuntimeState); + await db.delete(agents); + await db.delete(companies); + }); + + afterAll(async () => { + await tempDb?.cleanup(); + }); + + async function seedCompanyAndAgent(opts: SeedOptions = {}): Promise { + const companyId = randomUUID(); + const agentId = randomUUID(); + await db.insert(companies).values({ + id: companyId, + name: "Paperclip", + issuePrefix: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`, + requireBoardApprovalForNewAgents: false, + }); + await db.insert(agents).values({ + id: agentId, + companyId, + name: opts.agentName ?? "ClaudeCoder", + role: opts.agentRole ?? "engineer", + status: "active", + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: { + heartbeat: { + wakeOnDemand: true, + maxConcurrentRuns: opts.maxConcurrentRuns ?? 1, + }, + }, + permissions: {}, + }); + return { companyId, agentId }; + } + + async function seedQueuedRun(input: { + companyId: string; + agentId: string; + issueId: string; + wakeReason: string; + contextExtras?: Record; + invocationSource?: "assignment" | "automation"; + }) { + const wakeupRequestId = randomUUID(); + const runId = randomUUID(); + await db.insert(agentWakeupRequests).values({ + id: wakeupRequestId, + companyId: input.companyId, + agentId: input.agentId, + source: input.invocationSource ?? "assignment", + triggerDetail: "system", + reason: input.wakeReason, + payload: { issueId: input.issueId }, + status: "queued", + }); + await db.insert(heartbeatRuns).values({ + id: runId, + companyId: input.companyId, + agentId: input.agentId, + invocationSource: input.invocationSource ?? "assignment", + triggerDetail: "system", + status: "queued", + wakeupRequestId, + contextSnapshot: { + issueId: input.issueId, + wakeReason: input.wakeReason, + ...(input.contextExtras ?? {}), + }, + }); + await db + .update(agentWakeupRequests) + .set({ runId }) + .where(eq(agentWakeupRequests.id, wakeupRequestId)); + return { runId, wakeupRequestId }; + } + + it("cancels queued runs when the issue assignee changes before the run starts", async () => { + const { companyId, agentId } = await seedCompanyAndAgent({ agentName: "OriginalCoder" }); + const replacementAgentId = randomUUID(); + await db.insert(agents).values({ + id: replacementAgentId, + companyId, + name: "ReplacementCoder", + role: "engineer", + status: "active", + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: { + heartbeat: { + wakeOnDemand: true, + maxConcurrentRuns: 1, + }, + }, + permissions: {}, + }); + + const issueId = randomUUID(); + await db.insert(issues).values({ + id: issueId, + companyId, + title: "Reassigned task", + status: "in_progress", + priority: "high", + assigneeAgentId: replacementAgentId, + }); + + const { runId, wakeupRequestId } = await seedQueuedRun({ + companyId, + agentId, + issueId, + wakeReason: "issue_assigned", + }); + + await heartbeat.resumeQueuedRuns(); + + await waitForCondition(async () => { + const run = await db + .select({ status: heartbeatRuns.status }) + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, runId)) + .then((rows) => rows[0] ?? null); + return run?.status === "cancelled"; + }); + + const [run, wakeup] = await Promise.all([ + db + .select({ + status: heartbeatRuns.status, + errorCode: heartbeatRuns.errorCode, + resultJson: heartbeatRuns.resultJson, + }) + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, runId)) + .then((rows) => rows[0] ?? null), + db + .select({ status: agentWakeupRequests.status, error: agentWakeupRequests.error }) + .from(agentWakeupRequests) + .where(eq(agentWakeupRequests.id, wakeupRequestId)) + .then((rows) => rows[0] ?? null), + ]); + + expect(run?.status).toBe("cancelled"); + expect(run?.errorCode).toBe("issue_assignee_changed"); + expect(run?.resultJson).toMatchObject({ stopReason: "issue_assignee_changed" }); + expect(wakeup?.status).toBe("skipped"); + expect(wakeup?.error).toContain("assignee changed"); + expect(mockAdapterExecute).not.toHaveBeenCalled(); + }); + + it("cancels queued runs when the issue reaches a terminal status before the run starts", async () => { + const { companyId, agentId } = await seedCompanyAndAgent(); + const issueId = randomUUID(); + await db.insert(issues).values({ + id: issueId, + companyId, + title: "Already-completed task", + status: "done", + priority: "medium", + assigneeAgentId: agentId, + }); + + const { runId, wakeupRequestId } = await seedQueuedRun({ + companyId, + agentId, + issueId, + wakeReason: "issue_assigned", + }); + + await heartbeat.resumeQueuedRuns(); + + await waitForCondition(async () => { + const run = await db + .select({ status: heartbeatRuns.status }) + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, runId)) + .then((rows) => rows[0] ?? null); + return run?.status === "cancelled"; + }); + + const [run, wakeup] = await Promise.all([ + db + .select({ status: heartbeatRuns.status, errorCode: heartbeatRuns.errorCode }) + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, runId)) + .then((rows) => rows[0] ?? null), + db + .select({ status: agentWakeupRequests.status }) + .from(agentWakeupRequests) + .where(eq(agentWakeupRequests.id, wakeupRequestId)) + .then((rows) => rows[0] ?? null), + ]); + + expect(run?.status).toBe("cancelled"); + expect(run?.errorCode).toBe("issue_terminal_status"); + expect(wakeup?.status).toBe("skipped"); + expect(mockAdapterExecute).not.toHaveBeenCalled(); + }); + + it("still runs comment-driven wakes on in_review issues even when the agent is no longer the current participant", async () => { + const { companyId, agentId } = await seedCompanyAndAgent(); + const otherAgentId = randomUUID(); + await db.insert(agents).values({ + id: otherAgentId, + companyId, + name: "ReviewerAgent", + role: "qa", + status: "active", + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: { heartbeat: { wakeOnDemand: true, maxConcurrentRuns: 1 } }, + permissions: {}, + }); + + const issueId = randomUUID(); + const commentId = randomUUID(); + await db.insert(issues).values({ + id: issueId, + companyId, + title: "In-review task with comment feedback", + status: "in_review", + priority: "medium", + assigneeAgentId: agentId, + executionState: { + status: "pending", + currentStageId: randomUUID(), + currentStageIndex: 0, + currentStageType: "review", + currentParticipant: { type: "agent", agentId: otherAgentId, userId: null }, + returnAssignee: { type: "agent", agentId, userId: null }, + reviewRequest: null, + completedStageIds: [], + lastDecisionId: null, + lastDecisionOutcome: null, + }, + }); + await db.insert(issueComments).values({ + id: commentId, + companyId, + issueId, + authorAgentId: otherAgentId, + body: "Review feedback comment", + }); + + const { runId } = await seedQueuedRun({ + companyId, + agentId, + issueId, + wakeReason: "issue_commented", + invocationSource: "automation", + contextExtras: { + commentId, + wakeCommentId: commentId, + source: "issue.comment", + }, + }); + + await heartbeat.resumeQueuedRuns(); + + await waitForCondition(async () => { + const run = await db + .select({ status: heartbeatRuns.status }) + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, runId)) + .then((rows) => rows[0] ?? null); + return run?.status === "succeeded"; + }); + + const run = await db + .select({ status: heartbeatRuns.status, errorCode: heartbeatRuns.errorCode }) + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, runId)) + .then((rows) => rows[0] ?? null); + expect(run?.status).toBe("succeeded"); + expect(run?.errorCode).toBeNull(); + }); + + it("baseline: runs queued runs when the issue is in_progress with the same assignee", async () => { + const { companyId, agentId } = await seedCompanyAndAgent(); + const issueId = randomUUID(); + await db.insert(issues).values({ + id: issueId, + companyId, + title: "Still actionable", + status: "in_progress", + priority: "medium", + assigneeAgentId: agentId, + }); + + const { runId } = await seedQueuedRun({ + companyId, + agentId, + issueId, + wakeReason: "issue_assigned", + }); + + await heartbeat.resumeQueuedRuns(); + + await waitForCondition(async () => { + const run = await db + .select({ status: heartbeatRuns.status }) + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, runId)) + .then((rows) => rows[0] ?? null); + return run?.status === "succeeded"; + }); + + const run = await db + .select({ status: heartbeatRuns.status, errorCode: heartbeatRuns.errorCode }) + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, runId)) + .then((rows) => rows[0] ?? null); + expect(run?.status).toBe("succeeded"); + expect(run?.errorCode).toBeNull(); + expect(mockAdapterExecute).toHaveBeenCalledTimes(1); + }); +}); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 0d2c715f4da..c8e33372d05 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -77,6 +77,7 @@ import { sanitizeRuntimeServiceBaseEnv, } from "./workspace-runtime.js"; import { issueService } from "./issues.js"; +import { parseIssueExecutionState } from "./issue-execution-policy.js"; import { ISSUE_TREE_CONTROL_INTERACTION_WAKE_REASONS, isVerifiedIssueTreeControlInteractionWake, @@ -3792,6 +3793,16 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) logger.info({ runId: run.id, issueId, unresolvedBlockerCount }, "claimQueuedRun: cancelled blocked queued run"); return null; } + + const staleness = await evaluateQueuedRunStaleness(run, issueId, context); + if (staleness.stale) { + await cancelQueuedRunForStaleIssue(run, issueId, staleness); + logger.info( + { runId: run.id, issueId, errorCode: staleness.errorCode }, + "claimQueuedRun: cancelled stale queued run", + ); + return null; + } } const claimedAt = new Date(); @@ -3912,6 +3923,151 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) return cancelled; } + type QueuedRunStaleness = + | { stale: false } + | { + stale: true; + reason: string; + errorCode: + | "issue_not_found" + | "issue_assignee_changed" + | "issue_terminal_status" + | "issue_review_participant_changed"; + details: Record; + }; + + async function evaluateQueuedRunStaleness( + run: typeof heartbeatRuns.$inferSelect, + issueId: string, + context: Record, + ): Promise { + const issue = await db + .select({ + id: issues.id, + status: issues.status, + assigneeAgentId: issues.assigneeAgentId, + executionState: issues.executionState, + }) + .from(issues) + .where(and(eq(issues.id, issueId), eq(issues.companyId, run.companyId))) + .then((rows) => rows[0] ?? null); + + if (!issue) { + return { + stale: true, + errorCode: "issue_not_found", + reason: "Cancelled because the target issue no longer exists", + details: { issueId }, + }; + } + + const wakeCommentId = deriveCommentId(context, null); + const isInteractionWake = allowsIssueInteractionWake(context); + const resumeIntent = context.resumeIntent === true || context.followUpRequested === true; + + if (issue.assigneeAgentId !== run.agentId && !isInteractionWake) { + return { + stale: true, + errorCode: "issue_assignee_changed", + reason: + "Cancelled because issue assignee changed before the queued run could start; the new owner will be woken instead", + details: { + issueId, + previousAssigneeAgentId: run.agentId, + currentAssigneeAgentId: issue.assigneeAgentId, + }, + }; + } + + if (issue.status === "done" || issue.status === "cancelled") { + if (!resumeIntent && !wakeCommentId) { + return { + stale: true, + errorCode: "issue_terminal_status", + reason: `Cancelled because issue reached terminal status (${issue.status}) before the queued run could start`, + details: { issueId, currentStatus: issue.status }, + }; + } + } + + if (issue.status === "in_review") { + const executionState = parseIssueExecutionState(issue.executionState); + const currentParticipant = executionState?.currentParticipant ?? null; + if (currentParticipant) { + const participantMatches = + currentParticipant.type === "agent" && currentParticipant.agentId === run.agentId; + if (!participantMatches && !wakeCommentId) { + return { + stale: true, + errorCode: "issue_review_participant_changed", + reason: + "Cancelled because the in-review participant changed before the queued run could start; the current participant will be woken instead", + details: { + issueId, + currentStageType: executionState?.currentStageType ?? null, + currentParticipant, + }, + }; + } + } + } + + return { stale: false }; + } + + async function cancelQueuedRunForStaleIssue( + run: typeof heartbeatRuns.$inferSelect, + issueId: string, + staleness: Extract, + ) { + const now = new Date(); + const cancelled = await setRunStatus(run.id, "cancelled", { + finishedAt: now, + error: staleness.reason, + errorCode: staleness.errorCode, + resultJson: { + ...parseObject(run.resultJson), + stopReason: staleness.errorCode, + effectiveTimeoutSec: 0, + timeoutConfigured: false, + timeoutSource: "stale_queued_run_gate", + timeoutFired: false, + }, + }); + if (!cancelled) return null; + + await setWakeupStatus(run.wakeupRequestId, "skipped", { + finishedAt: now, + error: staleness.reason, + }); + + await db + .update(issues) + .set({ + executionRunId: null, + executionAgentNameKey: null, + executionLockedAt: null, + updatedAt: now, + }) + .where( + and( + eq(issues.companyId, run.companyId), + eq(issues.id, issueId), + eq(issues.executionRunId, run.id), + ), + ); + + await appendRunEvent(cancelled, await nextRunEventSeq(cancelled.id), { + eventType: "lifecycle", + stream: "system", + level: "warn", + message: staleness.reason, + payload: staleness.details, + }); + + return cancelled; + } + async function finalizeAgentStatus( agentId: string, outcome: "succeeded" | "failed" | "cancelled" | "timed_out",