Merge 6466257908 into c9e3bb7ca4

server/src/__tests__/heartbeat-process-recovery.test.ts

@@ -5859,6 +5859,128 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
     });
   });
 
+  it("blocks routine execution after successful missing-disposition recovery instead of starting productive continuation", async () => {
+    const { companyId, agentId, runId, issueId } =
+      await seedStrandedIssueFixture({
+        status: "in_progress",
+        runStatus: "succeeded",
+        livenessState: "advanced",
+      });
+    const sourceRunId = randomUUID();
+    await db
+      .update(issues)
+      .set({
+        originKind: "routine_execution",
+        originId: randomUUID(),
+      })
+      .where(eq(issues.id, issueId));
+    await db
+      .update(heartbeatRuns)
+      .set({
+        contextSnapshot: {
+          issueId,
+          taskId: issueId,
+          wakeReason: "source_scoped_recovery_action",
+          recoveryActionId: randomUUID(),
+          recoveryCause: SUCCESSFUL_RUN_MISSING_STATE_REASON,
+          sourceRunId,
+          recoveryIntent: "status_only",
+          allowDeliverableWork: false,
+          allowDocumentUpdates: false,
+          resumeRequiresNormalModel: true,
+        },
+      })
+      .where(eq(heartbeatRuns.id, runId));
+
+    const result =
+      await heartbeatService(db).reconcileStrandedAssignedIssues();
+
+    expect(result.continuationRequeued).toBe(0);
+    expect(result.successfulRunHandoffEscalated).toBe(1);
+    expect(result.issueIds).toEqual([issueId]);
+    expect(
+      await db
+        .select()
+        .from(issues)
+        .where(eq(issues.id, issueId))
+        .then((rows) => rows[0]?.status),
+    ).toBe("blocked");
+    await expectSourceScopedStrandedRecoveryAction({
+      companyId,
+      agentId,
+      issueId,
+      runId,
+      previousStatus: "in_progress",
+      retryReason: null,
+      cause: SUCCESSFUL_RUN_MISSING_STATE_REASON,
+      kind: "missing_disposition",
+    });
+    const runs = await db
+      .select()
+      .from(heartbeatRuns)
+      .where(eq(heartbeatRuns.agentId, agentId));
+    expect(runs).toHaveLength(1);
+  });
+
+  it("preserves productive continuation when fresh owner direction precedes its asynchronous wake", async () => {
+    const { companyId, agentId, runId, issueId } =
+      await seedStrandedIssueFixture({
+        status: "in_progress",
+        runStatus: "succeeded",
+        livenessState: "advanced",
+      });
+    const recoveryRunAt = new Date(Date.now() - 1_000);
+    await db
+      .update(issues)
+      .set({
+        originKind: "routine_execution",
+        originId: randomUUID(),
+      })
+      .where(eq(issues.id, issueId));
+    await db
+      .update(heartbeatRuns)
+      .set({
+        contextSnapshot: {
+          issueId,
+          taskId: issueId,
+          wakeReason: "source_scoped_recovery_action",
+          recoveryActionId: randomUUID(),
+          recoveryCause: SUCCESSFUL_RUN_MISSING_STATE_REASON,
+          recoveryIntent: "status_only",
+          allowDeliverableWork: false,
+          allowDocumentUpdates: false,
+          resumeRequiresNormalModel: true,
+        },
+        createdAt: recoveryRunAt,
+      })
+      .where(eq(heartbeatRuns.id, runId));
+    await db.insert(issueComments).values({
+      companyId,
+      issueId,
+      authorType: "user",
+      authorUserId: "local-board",
+      body: "Continue with this new owner instruction.",
+      createdAt: new Date(recoveryRunAt.getTime() + 500),
+    });
+
+    const result =
+      await heartbeatService(db).reconcileStrandedAssignedIssues();
+
+    expect(result.continuationRequeued).toBe(1);
+    expect(result.successfulRunHandoffEscalated).toBe(0);
+    const runs = await db
+      .select()
+      .from(heartbeatRuns)
+      .where(eq(heartbeatRuns.agentId, agentId));
+    expect(runs).toHaveLength(2);
+    expect(
+      runs.find((run) => run.id !== runId)?.contextSnapshot,
+    ).toMatchObject({
+      issueId,
+      source: "issue.productive_terminal_continuation_recovery",
+    });
+  });
+
   it("converts a continuation parked for review into a dependency wait on its open sub-tasks", async () => {
     const { companyId, agentId, issueId } = await seedStrandedIssueFixture({
       status: "in_progress",

server/src/services/recovery/service.ts

@@ -8,6 +8,7 @@ import {
   gt,
   gte,
   inArray,
+  isNotNull,
   isNull,
   not,
   notInArray,
@@ -754,6 +755,55 @@ function isExhaustedSuccessfulRunHandoff(latestRun: LatestIssueRun) {
   return { ...evidence, exhausted: true };
 }
 
+function routineMissingDispositionRecoveryEvidence(
+  issue: Pick<typeof issues.$inferSelect, "originKind">,
+  latestRun: LatestIssueRun,
+) {
+  // A status-only recovery may succeed without resolving the routine item.
+  // Treat that lineage as exhausted so it cannot become productive work.
+  if (
+    issue.originKind !== "routine_execution" ||
+    latestRun?.status !== "succeeded"
+  )
+    return null;
+
+  const context = parseObject(latestRun.contextSnapshot);
+  const paperclipWake = parseObject(context.paperclipWake);
+  const recovery = parseObject(paperclipWake.recovery);
+  const wakeReason =
+    readNonEmptyString(context.wakeReason) ??
+    readNonEmptyString(paperclipWake.reason);
+  const recoveryCause =
+    readNonEmptyString(context.recoveryCause) ??
+    readNonEmptyString(recovery.cause);
+  const isRecoveryActionRun =
+    wakeReason === "source_scoped_recovery_action" ||
+    readNonEmptyString(context.recoveryActionId) !== null;
+  const isMissingDispositionRecovery =
+    recoveryCause === SUCCESSFUL_RUN_MISSING_STATE_REASON ||
+    recoveryCause === "successful_run_missing_issue_disposition";
+  if (!isRecoveryActionRun || !isMissingDispositionRecovery) return null;
+
+  return {
+    sourceRunId:
+      readNonEmptyString(context.sourceRunId) ??
+      readNonEmptyString(context.resumeFromRunId) ??
+      readNonEmptyString(context.retryOfRunId),
+    correctiveRunId: latestRun.id,
+    missingDisposition:
+      readNonEmptyString(context.missingDisposition) ?? "clear_next_step",
+    handoffAttempt: Math.max(1, asNumber(context.handoffAttempt, 1)),
+    maxHandoffAttempts: Math.max(
+      1,
+      asNumber(
+        context.maxHandoffAttempts,
+        DEFAULT_MAX_SUCCESSFUL_RUN_HANDOFF_ATTEMPTS,
+      ),
+    ),
+    exhausted: true,
+  };
+}
+
 function issueIdFromRunContext(contextSnapshot: unknown) {
   const context = parseObject(contextSnapshot);
   return (
@@ -975,6 +1025,28 @@ export function recoveryService(
       .then((rows) => rows[0] ?? null);
   }
 
+  async function hasFreshUserDirectionAfterRun(
+    issue: Pick<typeof issues.$inferSelect, "companyId" | "id">,
+    latestRun: NonNullable<LatestIssueRun>,
+  ) {
+    return db
+      .select({ id: issueComments.id })
+      .from(issueComments)
+      .where(
+        and(
+          eq(issueComments.companyId, issue.companyId),
+          eq(issueComments.issueId, issue.id),
+          isNotNull(issueComments.authorUserId),
+          isNull(issueComments.authorAgentId),
+          isNull(issueComments.createdByRunId),
+          isNull(issueComments.deletedAt),
+          gt(issueComments.createdAt, latestRun.createdAt),
+        ),
+      )
+      .limit(1)
+      .then((rows) => Boolean(rows[0]));
+  }
+
   async function summarizeRecentContinuationRetries(
     companyId: string,
     issueId: string,
@@ -4966,7 +5038,17 @@ export function recoveryService(
         }
         continue;
       }
-      const handoffEvidence = isExhaustedSuccessfulRunHandoff(latestRun);
+      const exhaustedHandoffEvidence =
+        isExhaustedSuccessfulRunHandoff(latestRun);
+      const routineRecoveryEvidence =
+        routineMissingDispositionRecoveryEvidence(issue, latestRun);
+      const hasFreshUserDirection =
+        routineRecoveryEvidence && latestRun
+          ? await hasFreshUserDirectionAfterRun(issue, latestRun)
+          : false;
+      const handoffEvidence =
+        exhaustedHandoffEvidence ??
+        (hasFreshUserDirection ? null : routineRecoveryEvidence);
       if (handoffEvidence) {
         if (isPluginManagedIssueLifecycle(issue)) {
           result.skipped += 1;