Harden recovery wait boundaries

server/src/services/heartbeat.ts

@@ -80,6 +80,7 @@ import {
   sanitizeRuntimeServiceBaseEnv,
 } from "./workspace-runtime.js";
 import { issueService } from "./issues.js";
+import { issueThreadInteractionService } from "./issue-thread-interactions.js";
 import { parseIssueExecutionState } from "./issue-execution-policy.js";
 import {
   ISSUE_TREE_CONTROL_INTERACTION_WAKE_REASONS,
@@ -2774,6 +2775,11 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
   }
 
   async function hasExplicitIssueWait(companyId: string, issueId: string) {
+    await issueThreadInteractionService(db).expireInvalidPendingRequestConfirmationsForIssue(
+      { id: issueId, companyId },
+      { agentId: null, userId: null },
+    );
+
     const [interaction, approval] = await Promise.all([
       db
         .select({ id: issueThreadInteractions.id })

server/src/services/issue-thread-interactions.ts

@@ -1,5 +1,5 @@
 import { isDeepStrictEqual } from "node:util";
-import { and, asc, eq, inArray } from "drizzle-orm";
+import { and, asc, eq, gt, inArray, sql } from "drizzle-orm";
 import type { Db } from "@paperclipai/db";
 import {
   documents,
@@ -463,6 +463,55 @@ async function expireStaleRequestConfirmationTarget(db: Db | any, args: {
   return hydrateInteraction(updated);
 }
 
+async function expireRequestConfirmationSupersededByLatestUserComment(db: Db | any, args: {
+  row: IssueThreadInteractionRow;
+  actor: InteractionActor;
+}): Promise<IssueThreadInteraction | null> {
+  if (args.row.kind !== "request_confirmation" || args.row.status !== "pending") return null;
+  const interaction = hydrateInteraction(args.row) as RequestConfirmationInteraction;
+  if (interaction.payload.supersedeOnUserComment !== true) return null;
+
+  const [comment] = await db
+    .select({ id: issueComments.id })
+    .from(issueComments)
+    .where(and(
+      eq(issueComments.companyId, args.row.companyId),
+      eq(issueComments.issueId, args.row.issueId),
+      gt(issueComments.createdAt, args.row.createdAt),
+      sql`${issueComments.authorUserId} is not null`,
+    ))
+    .orderBy(asc(issueComments.createdAt))
+    .limit(1);
+  if (!comment) return null;
+
+  const now = new Date();
+  const [updated] = await db
+    .update(issueThreadInteractions)
+    .set({
+      status: "expired",
+      result: {
+        version: 1,
+        outcome: "superseded_by_comment",
+        commentId: comment.id,
+      },
+      resolvedByAgentId: args.actor.agentId ?? null,
+      resolvedByUserId: args.actor.userId ?? null,
+      resolvedAt: now,
+      updatedAt: now,
+    })
+    .where(and(
+      eq(issueThreadInteractions.id, args.row.id),
+      eq(issueThreadInteractions.status, "pending"),
+    ))
+    .returning();
+
+  if (!updated) {
+    throw conflict("Interaction has already been resolved");
+  }
+  await touchIssue(db, args.row.issueId);
+  return hydrateInteraction(updated);
+}
+
 export function issueThreadInteractionService(db: Db) {
   async function getIdempotentInteraction(args: {
     issueId: string;
@@ -1133,6 +1182,33 @@ export function issueThreadInteractionService(db: Db) {
       return expired;
     },
 
+    expireInvalidPendingRequestConfirmationsForIssue: async (
+      issue: { id: string; companyId: string },
+      actor: InteractionActor,
+    ) => {
+      const rows = await db
+        .select()
+        .from(issueThreadInteractions)
+        .where(and(
+          eq(issueThreadInteractions.companyId, issue.companyId),
+          eq(issueThreadInteractions.issueId, issue.id),
+          eq(issueThreadInteractions.kind, "request_confirmation"),
+          eq(issueThreadInteractions.status, "pending"),
+        ));
+
+      const expired: IssueThreadInteraction[] = [];
+      for (const row of rows) {
+        const staleTarget = await expireStaleRequestConfirmationTarget(db, { row, actor });
+        if (staleTarget) {
+          expired.push(staleTarget);
+          continue;
+        }
+        const superseded = await expireRequestConfirmationSupersededByLatestUserComment(db, { row, actor });
+        if (superseded) expired.push(superseded);
+      }
+      return expired;
+    },
+
     answerQuestions: async (
       issue: { id: string; companyId: string },
       interactionId: string,

server/src/services/recovery/service.ts

@@ -32,6 +32,7 @@ import { budgetService } from "../budgets.js";
 import { instanceSettingsService } from "../instance-settings.js";
 import { issueTreeControlService } from "../issue-tree-control.js";
 import { issueService } from "../issues.js";
+import { issueThreadInteractionService } from "../issue-thread-interactions.js";
 import { getRunLogStore } from "../run-log-store.js";
 import {
   RECOVERY_ORIGIN_KINDS,
@@ -398,6 +399,10 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
     if (issue.status === "done" || issue.status === "cancelled") return true;
     if (issue.status === "blocked" || issue.status === "in_review") return true;
     if (issue.executionState) return true;
+    await issueThreadInteractionService(db).expireInvalidPendingRequestConfirmationsForIssue(
+      { id: issue.id, companyId: issue.companyId },
+      { agentId: null, userId: null },
+    );
 
     const [interaction, approval, recoveryIssue] = await Promise.all([
       db
@@ -1793,6 +1798,11 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
         continue;
       }
 
+      if (await hasExplicitWaitingPath(issue)) {
+        result.skipped += 1;
+        continue;
+      }
+
       const latestRun = await getLatestIssueRun(issue.companyId, issue.id);
       if (isStrandedIssueRecoveryIssue(issue) && isUnsuccessfulTerminalIssueRun(latestRun)) {
         const updated = await escalateStrandedRecoveryIssueInPlace({