From 7ad129ae327d0597161b5dbc5bee926bc6300677 Mon Sep 17 00:00:00 2001 From: Dotta Date: Sat, 11 Apr 2026 10:50:17 -0500 Subject: [PATCH] fix: reopen deferred queued comment wakes --- .../heartbeat-comment-wake-batching.test.ts | 186 ++++++++++++++++++ server/src/services/heartbeat.ts | 107 ++++++++-- 2 files changed, 276 insertions(+), 17 deletions(-) diff --git a/server/src/__tests__/heartbeat-comment-wake-batching.test.ts b/server/src/__tests__/heartbeat-comment-wake-batching.test.ts index 3626c2f5ff3..91e1112bd8a 100644 --- a/server/src/__tests__/heartbeat-comment-wake-batching.test.ts +++ b/server/src/__tests__/heartbeat-comment-wake-batching.test.ts @@ -424,6 +424,192 @@ describe("heartbeat comment wake batching", () => { } }, 120_000); + it("promotes deferred comment wakes after the active run closes the issue", async () => { + const gateway = await createControlledGatewayServer(); + const companyId = randomUUID(); + const agentId = randomUUID(); + const issueId = randomUUID(); + const issuePrefix = `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`; + const heartbeat = heartbeatService(db); + + try { + await db.insert(companies).values({ + id: companyId, + name: "Paperclip", + issuePrefix, + requireBoardApprovalForNewAgents: false, + }); + + await db.insert(agents).values({ + id: agentId, + companyId, + name: "Gateway Agent", + role: "engineer", + status: "idle", + adapterType: "openclaw_gateway", + adapterConfig: { + url: gateway.url, + headers: { + "x-openclaw-token": "gateway-token", + }, + payloadTemplate: { + message: "wake now", + }, + waitTimeoutMs: 2_000, + }, + runtimeConfig: {}, + permissions: {}, + }); + + await db.insert(issues).values({ + id: issueId, + companyId, + title: "Reopen after deferred comment", + status: "todo", + priority: "medium", + assigneeAgentId: agentId, + issueNumber: 1, + identifier: `${issuePrefix}-1`, + }); + + const comment1 = await db + .insert(issueComments) + .values({ + companyId, + issueId, + authorUserId: "user-1", + body: "First comment", + }) + .returning() + .then((rows) => rows[0]); + + const firstRun = await heartbeat.wakeup(agentId, { + source: "automation", + triggerDetail: "system", + reason: "issue_commented", + payload: { issueId, commentId: comment1.id }, + contextSnapshot: { + issueId, + taskId: issueId, + commentId: comment1.id, + wakeReason: "issue_commented", + }, + requestedByActorType: "user", + requestedByActorId: "user-1", + }); + + expect(firstRun).not.toBeNull(); + await waitFor(async () => { + const run = await db + .select({ status: heartbeatRuns.status }) + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, firstRun!.id)) + .then((rows) => rows[0] ?? null); + return run?.status === "running"; + }); + + const comment2 = await db + .insert(issueComments) + .values({ + companyId, + issueId, + authorUserId: "user-1", + body: "Please handle this follow-up after you finish", + }) + .returning() + .then((rows) => rows[0]); + + const deferredRun = await heartbeat.wakeup(agentId, { + source: "automation", + triggerDetail: "system", + reason: "issue_commented", + payload: { issueId, commentId: comment2.id }, + contextSnapshot: { + issueId, + taskId: issueId, + commentId: comment2.id, + wakeReason: "issue_commented", + }, + requestedByActorType: "user", + requestedByActorId: "user-1", + }); + + expect(deferredRun).toBeNull(); + + await waitFor(async () => { + const deferred = await db + .select() + .from(agentWakeupRequests) + .where( + and( + eq(agentWakeupRequests.companyId, companyId), + eq(agentWakeupRequests.agentId, agentId), + eq(agentWakeupRequests.status, "deferred_issue_execution"), + ), + ) + .then((rows) => rows[0] ?? null); + return Boolean(deferred); + }); + + await db + .update(issues) + .set({ + status: "done", + completedAt: new Date(), + executionRunId: null, + executionAgentNameKey: null, + executionLockedAt: null, + updatedAt: new Date(), + }) + .where(eq(issues.id, issueId)); + + gateway.releaseFirstWait(); + + await waitFor(() => gateway.getAgentPayloads().length === 2, 90_000); + await waitFor(async () => { + const runs = await db + .select() + .from(heartbeatRuns) + .where(eq(heartbeatRuns.agentId, agentId)); + return runs.length === 2 && runs.every((run) => run.status === "succeeded"); + }, 90_000); + + const reopenedIssue = await db + .select({ + status: issues.status, + completedAt: issues.completedAt, + }) + .from(issues) + .where(eq(issues.id, issueId)) + .then((rows) => rows[0] ?? null); + + expect(reopenedIssue).toMatchObject({ + status: "todo", + completedAt: null, + }); + + const secondPayload = gateway.getAgentPayloads()[1] ?? {}; + expect(secondPayload.paperclip).toMatchObject({ + wake: { + reason: "issue_commented", + commentIds: [comment2.id], + latestCommentId: comment2.id, + issue: { + id: issueId, + identifier: `${issuePrefix}-1`, + title: "Reopen after deferred comment", + status: "todo", + priority: "medium", + }, + }, + }); + expect(String(secondPayload.message ?? "")).toContain("Please handle this follow-up after you finish"); + } finally { + gateway.releaseFirstWait(); + await gateway.close(); + } + }, 120_000); + it("queues exactly one follow-up run when an issue-bound run exits without a comment", async () => { const gateway = await createControlledGatewayServer(); const companyId = randomUUID(); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index e0e0a317cec..8287c5a02f2 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -37,6 +37,7 @@ import { mergeHeartbeatRunResultJson, summarizeHeartbeatRunResultJson, } from "./heartbeat-run-summary.js"; +import { logActivity, type LogActivityInput } from "./activity-log.js"; import { buildWorkspaceReadyComment, cleanupExecutionWorkspaceArtifacts, @@ -3632,31 +3633,50 @@ export function heartbeatService(db: Db) { } async function releaseIssueExecutionAndPromote(run: typeof heartbeatRuns.$inferSelect) { - const promotedRun = await db.transaction(async (tx) => { - await tx.execute( - sql`select id from issues where company_id = ${run.companyId} and execution_run_id = ${run.id} for update`, - ); + const runContext = parseObject(run.contextSnapshot); + const contextIssueId = readNonEmptyString(runContext.issueId); + const promotionResult = await db.transaction(async (tx) => { + if (contextIssueId) { + await tx.execute( + sql`select id from issues where company_id = ${run.companyId} and id = ${contextIssueId} for update`, + ); + } else { + await tx.execute( + sql`select id from issues where company_id = ${run.companyId} and execution_run_id = ${run.id} for update`, + ); + } - const issue = await tx + let issue = await tx .select({ id: issues.id, companyId: issues.companyId, + identifier: issues.identifier, + status: issues.status, + executionRunId: issues.executionRunId, }) .from(issues) - .where(and(eq(issues.companyId, run.companyId), eq(issues.executionRunId, run.id))) + .where( + and( + eq(issues.companyId, run.companyId), + contextIssueId ? eq(issues.id, contextIssueId) : eq(issues.executionRunId, run.id), + ), + ) .then((rows) => rows[0] ?? null); - if (!issue) return; + if (!issue) return null; + if (issue.executionRunId && issue.executionRunId !== run.id) return null; - await tx - .update(issues) - .set({ - executionRunId: null, - executionAgentNameKey: null, - executionLockedAt: null, - updatedAt: new Date(), - }) - .where(eq(issues.id, issue.id)); + if (issue.executionRunId === run.id) { + await tx + .update(issues) + .set({ + executionRunId: null, + executionAgentNameKey: null, + executionLockedAt: null, + updatedAt: new Date(), + }) + .where(eq(issues.id, issue.id)); + } while (true) { const deferred = await tx @@ -3703,6 +3723,51 @@ export function heartbeatService(db: Db) { const deferredPayload = parseObject(deferred.payload); const deferredContextSeed = parseObject(deferredPayload[DEFERRED_WAKE_CONTEXT_KEY]); const promotedContextSeed: Record = { ...deferredContextSeed }; + const deferredCommentIds = extractWakeCommentIds(deferredContextSeed); + const shouldReopenDeferredCommentWake = + deferredCommentIds.length > 0 && (issue.status === "done" || issue.status === "cancelled"); + let reopenedActivity: LogActivityInput | null = null; + + if (shouldReopenDeferredCommentWake) { + const reopenedFromStatus = issue.status; + const reopenedIssue = await issuesSvc.update( + issue.id, + { + status: "todo", + executionState: null, + }, + tx, + ); + if (reopenedIssue) { + issue = { + ...issue, + identifier: reopenedIssue.identifier, + status: reopenedIssue.status, + executionRunId: reopenedIssue.executionRunId, + }; + if (!readNonEmptyString(promotedContextSeed.reopenedFrom)) { + promotedContextSeed.reopenedFrom = reopenedFromStatus; + } + reopenedActivity = { + companyId: issue.companyId, + actorType: "system", + actorId: "heartbeat", + agentId: deferred.agentId, + runId: run.id, + action: "issue.updated", + entityType: "issue", + entityId: issue.id, + details: { + status: "todo", + reopened: true, + reopenedFrom: reopenedFromStatus, + source: "deferred_comment_wake", + identifier: issue.identifier, + }, + }; + } + } + const promotedReason = readNonEmptyString(deferred.reason) ?? "issue_execution_promoted"; const promotedSource = (readNonEmptyString(deferred.source) as WakeupOptions["source"]) ?? "automation"; @@ -3764,12 +3829,20 @@ export function heartbeatService(db: Db) { }) .where(eq(issues.id, issue.id)); - return newRun; + return { + run: newRun, + reopenedActivity, + }; } }); + const promotedRun = promotionResult?.run ?? null; if (!promotedRun) return; + if (promotionResult?.reopenedActivity) { + await logActivity(db, promotionResult.reopenedActivity); + } + publishLiveEvent({ companyId: promotedRun.companyId, type: "heartbeat.run.queued",