From e4146f272baa9a7e544ad42d85482e66c01de84b Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 11 Sep 2026 15:52:29 -0500 Subject: [PATCH] Prevent stale conversation recovery and terminal task wakes --- .../heartbeat-comment-wake-batching.test.ts | 48 +++++++++++++++---- .../heartbeat-process-recovery.test.ts | 33 +++++++++++++ server/src/services/heartbeat.ts | 21 ++++++++ server/src/services/recovery/service.ts | 12 +++++ 4 files changed, 104 insertions(+), 10 deletions(-) diff --git a/server/src/__tests__/heartbeat-comment-wake-batching.test.ts b/server/src/__tests__/heartbeat-comment-wake-batching.test.ts index 9bc0e48011..df11681291 100644 --- a/server/src/__tests__/heartbeat-comment-wake-batching.test.ts +++ b/server/src/__tests__/heartbeat-comment-wake-batching.test.ts @@ -1105,7 +1105,11 @@ describeEmbeddedPostgres("heartbeat comment wake batching", () => { } }, 120_000); - it("does not reopen a finished issue when the deferred comment wake came from another agent", async () => { + it.each([ + { caseName: "allows a non-assignee mention on completed work", targetAssignee: false, terminalStatus: "done" }, + { caseName: "cancels an assignee continuation on completed work", targetAssignee: true, terminalStatus: "done" }, + { caseName: "cancels an assignee continuation on cancelled work", targetAssignee: true, terminalStatus: "cancelled" }, + ] as const)("$caseName without reopening an agent-commented task", async ({ targetAssignee, terminalStatus }) => { const gateway = await createControlledGatewayServer(); const companyId = randomUUID(); const assigneeAgentId = randomUUID(); @@ -1113,6 +1117,9 @@ describeEmbeddedPostgres("heartbeat comment wake batching", () => { const issueId = randomUUID(); const issuePrefix = `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`; const heartbeat = heartbeatService(db); + const targetAgentId = targetAssignee ? assigneeAgentId : mentionedAgentId; + const commentingAgentId = targetAssignee ? mentionedAgentId : assigneeAgentId; + const wakeReason = targetAssignee ? "issue_commented" : "issue_comment_mentioned"; try { await db.insert(companies).values({ @@ -1208,28 +1215,29 @@ describeEmbeddedPostgres("heartbeat comment wake batching", () => { .values({ companyId, issueId, - authorAgentId: assigneeAgentId, - createdByRunId: firstRun?.id ?? null, + authorAgentId: commentingAgentId, + createdByRunId: targetAssignee ? null : firstRun?.id ?? null, body: "@Mentioned Agent please review after I finish", }) .returning() .then((rows) => rows[0]); - const deferredRun = await heartbeat.wakeup(mentionedAgentId, { + const deferredRun = await heartbeat.wakeup(targetAgentId, { source: "automation", triggerDetail: "system", - reason: "issue_comment_mentioned", + reason: wakeReason, payload: { issueId, commentId: comment.id }, contextSnapshot: { issueId, taskId: issueId, commentId: comment.id, wakeCommentId: comment.id, - wakeReason: "issue_comment_mentioned", + wakeReason, + ...(targetAssignee ? { resumeIntent: true, followUpRequested: true } : {}), source: "comment.mention", }, requestedByActorType: "agent", - requestedByActorId: assigneeAgentId, + requestedByActorId: commentingAgentId, }); expect(deferredRun).toBeNull(); @@ -1241,7 +1249,7 @@ describeEmbeddedPostgres("heartbeat comment wake batching", () => { .where( and( eq(agentWakeupRequests.companyId, companyId), - eq(agentWakeupRequests.agentId, mentionedAgentId), + eq(agentWakeupRequests.agentId, targetAgentId), eq(agentWakeupRequests.status, "deferred_issue_execution"), ), ) @@ -1256,7 +1264,7 @@ describeEmbeddedPostgres("heartbeat comment wake batching", () => { await db .update(issues) .set({ - status: "done", + status: terminalStatus, completedAt: new Date(), executionRunId: null, executionAgentNameKey: null, @@ -1267,6 +1275,26 @@ describeEmbeddedPostgres("heartbeat comment wake batching", () => { gateway.releaseFirstWait(); + if (targetAssignee) { + await waitFor(async () => { + const cancelled = await db.select().from(agentWakeupRequests).where(and( + eq(agentWakeupRequests.companyId, companyId), + eq(agentWakeupRequests.agentId, targetAgentId), + eq(agentWakeupRequests.status, "cancelled"), + )); + return cancelled.some((wake) => wake.error === "Deferred execution wake no longer applies to a terminal task"); + }); + const runs = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, companyId)); + expect(runs).toEqual([expect.objectContaining({ id: firstRun!.id, status: "succeeded" })]); + expect(gateway.getAgentPayloads()).toHaveLength(1); + const [closedIssue] = await db.select().from(issues).where(eq(issues.id, issueId)); + expect(closedIssue).toMatchObject({ status: terminalStatus, executionRunId: null }); + expect(closedIssue.completedAt).not.toBeNull(); + const [retainedComment] = await db.select().from(issueComments).where(eq(issueComments.id, comment.id)); + expect(retainedComment.body).toContain("please review after I finish"); + return; + } + await waitFor(() => gateway.getAgentPayloads().length === 2, 90_000); await waitFor(async () => { const runs = await db @@ -1294,7 +1322,7 @@ describeEmbeddedPostgres("heartbeat comment wake batching", () => { expect(secondPayload.paperclip).toBeUndefined(); const secondWake = parseWakePayloadFromMessage(secondPayload.message); expect(secondWake).toMatchObject({ - reason: "issue_comment_mentioned", + reason: wakeReason, commentIds: [comment.id], latestCommentId: comment.id, issue: { diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index e63a7564de..ab223f7d56 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -1,3 +1,4 @@ +import { instanceSettingsService } from "../services/instance-settings.js"; import { randomUUID } from "node:crypto"; import { spawn, type ChildProcess } from "node:child_process"; import fs from "node:fs/promises"; @@ -8571,6 +8572,38 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { } }); + it("does not recover a finished native chat while its response publication is pending", async () => { + const { agentId, issueId, runId } = await seedStrandedIssueFixture({ + status: "in_progress", + runStatus: "succeeded", + livenessState: "advanced", + resultJson: { finalizationReasonCode: "conversation_turn_finished" }, + }); + await instanceSettingsService(db).updateExperimental({ enableAgentChat: true }); + try { + await db.update(issues).set({ + conversationAgentId: agentId, + conversationUserId: "responsible-user", + conversationState: "active", + }).where(eq(issues.id, issueId)); + const result = await heartbeatService(db).reconcileStrandedAssignedIssues(); + expect(result.continuationRequeued).toBe(0); + expect(result.escalated).toBe(0); + const runs = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.agentId, agentId)); + expect(runs.map((run) => run.id)).toEqual([runId]); + const wakes = await db.select().from(agentWakeupRequests) + .where(eq(agentWakeupRequests.agentId, agentId)); + expect(wakes).toHaveLength(1); + expect(wakes[0].reason).toBe("issue_assigned"); + const [issue] = await db.select().from(issues).where(eq(issues.id, issueId)); + // No fabricated idle state: durable response publication still settles it. + expect(issue.status).toBe("in_progress"); + expect(issue.conversationState).toBe("active"); + } finally { + await instanceSettingsService(db).updateExperimental({ enableAgentChat: false }); + } + }); + it("leaves the productive-but-stranded continuation path unchanged under the new classifier", async () => { const { agentId, issueId, runId } = await seedStrandedIssueFixture({ status: "in_progress", diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index f721a09eb6..f4cb769a15 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -23177,6 +23177,27 @@ export function heartbeatService( } } + // A queued agent comment can outlive the assignment it addressed. The + // human reopen path above may revive it; otherwise do not dispatch an + // assignee continuation against work that has already ended. Mentions + // to other agents remain notifications and can inspect the closed task. + if ( + (issue.status === "done" || issue.status === "cancelled") && + deferred.agentId === issue.assigneeAgentId + ) { + const now = new Date(); + await tx + .update(agentWakeupRequests) + .set({ + status: "cancelled", + finishedAt: now, + error: "Deferred execution wake no longer applies to a terminal task", + updatedAt: now, + }) + .where(eq(agentWakeupRequests.id, deferred.id)); + continue; + } + const promotedReason = readNonEmptyString(deferred.reason) ?? "issue_execution_promoted"; const promotedSource = diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index bd51c83ffb..5fce2ab8d6 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -3023,6 +3023,18 @@ export function recoveryService( } let latestRun = await getLatestIssueRun(issue.companyId, issue.id); + // A native chat can finish between the earlier settlement read and this + // fresh run read, before its response is materialized. Its trusted + // finalizer owns that settlement; generic productive-work recovery must + // not invent another conversation turn during the publication window. + if ( + issue.conversationAgentId && + latestRun?.status === "succeeded" && + parseObject(latestRun.resultJson).finalizationReasonCode === "conversation_turn_finished" + ) { + result.skipped += 1; + continue; + } const agent = await getAgent(agentId); const agentInvokable = agent && agent.companyId === issue.companyId