Prevent stale conversation recovery and terminal task wakes

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: {

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",

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 =

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