diff --git a/server/src/__tests__/agent-conversations.test.ts b/server/src/__tests__/agent-conversations.test.ts index 049e2f8062..edc861e41b 100644 --- a/server/src/__tests__/agent-conversations.test.ts +++ b/server/src/__tests__/agent-conversations.test.ts @@ -34,6 +34,8 @@ import { import { issueService } from "../services/issues.js"; import { documentService } from "../services/documents.js"; import { getTaskPlanContext } from "../services/task-plan-context.js"; +import { terminalizeLegacyExecution, LEGACY_RECOVERY_CAUSE } from "../services/legacy-execution-recovery.js"; +import { settleUnrecoverableExecutions } from "../services/execution-recovery-resolution.js"; import { renderPaperclipWakePrompt } from "@paperclipai/adapter-utils/server-utils"; import { instanceSettingsService } from "../services/instance-settings.js"; import { @@ -777,6 +779,49 @@ const support = await getEmbeddedPostgresTestSupport(); expect((await issueService(db).getDependencyReadiness(chat.id)).blockerIssueIds).toEqual([blocker.id]); }); + it.each([ + { name: "reset idle chat", generation: 1, sourceGeneration: 0, waiting: true, ordinary: false, superseded: true }, + { name: "reset active chat", generation: 1, sourceGeneration: 0, waiting: false, ordinary: false, superseded: true }, + { name: "newer reply in the same session", generation: 1, sourceGeneration: 1, waiting: true, ordinary: false, superseded: true }, + { name: "current unanswered chat turn", generation: 1, sourceGeneration: 1, waiting: false, ordinary: false, superseded: false }, + { name: "unprepared failure without a session generation", generation: 1, sourceGeneration: undefined, waiting: true, ordinary: false, superseded: false }, + { name: "ordinary review task", generation: 0, sourceGeneration: 0, waiting: true, ordinary: true, superseded: false }, + ])("guards delayed cancelled-run recovery for $name", async (scenario) => { + const task = scenario.ordinary + ? await issueService(db).create(companyId, { title: "Ordinary review", status: "in_review", assigneeAgentId: agentId }) + : await create(); + const status = scenario.waiting ? "in_review" : "in_progress"; + await db.update(issues).set({ + status, + ...(scenario.ordinary ? {} : { + conversationSessionGeneration: scenario.generation, + conversationState: scenario.waiting ? "waiting" : "active", + }), + }).where(eq(issues.id, task.id)); + const run = await runFor(task.id, randomUUID(), { + conversationSessionGeneration: scenario.sourceGeneration, + }); + await terminalizeLegacyExecution({ db, run, status: "cancelled", patch: { finishedAt: new Date() } }); + let actions = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, task.id)); + expect(actions).toHaveLength(scenario.superseded ? 0 : 1); + // Also exercise an action queued before /new or the newer reply settled. + if (!actions.length) { + actions = await db.insert(issueRecoveryActions).values({ + companyId, sourceIssueId: task.id, kind: "active_run_watchdog", + ownerType: "board", returnOwnerAgentId: agentId, + cause: LEGACY_RECOVERY_CAUSE, fingerprint: `legacy-execution:${run.id}`, + evidence: { runId: run.id }, nextAction: "Reconcile stopped work", + }).returning(); + } + await settleUnrecoverableExecutions(db); + const [after] = await db.select().from(issues).where(eq(issues.id, task.id)); + const [action] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, actions[0]!.id)); + expect(after.status).toBe(scenario.superseded ? status : "blocked"); + expect(action).toMatchObject({ status: "resolved", outcome: scenario.superseded ? "cancelled" : "blocked" }); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)))[0].status).toBe("cancelled"); + if (!scenario.ordinary) expect(after.conversationSessionGeneration).toBe(scenario.generation); + }); + it("only parks answered turns and preserves idle across recovery classification", async () => { const issue = await create(); const message = await issueService(db).addComment( diff --git a/server/src/services/agent-conversations.ts b/server/src/services/agent-conversations.ts index 0ebdb1b6f4..e0efab65ef 100644 --- a/server/src/services/agent-conversations.ts +++ b/server/src/services/agent-conversations.ts @@ -38,6 +38,26 @@ export function isWaitingConversation( issue.status === "in_review" ); } + +/** Recovery for an older turn must not replace a reset or an answered chat. */ +export function isSupersededConversationRun( + issue: ConversationIdentity & { + conversationSessionGeneration?: number; + executionRunId?: string | null; + }, + run: { id: string; contextSnapshot: Record | null }, +): boolean { + if (!isConversation(issue)) return false; + const generation = run.contextSnapshot?.conversationSessionGeneration; + return ( + (typeof generation === "number" && + typeof issue.conversationSessionGeneration === "number" && + generation !== issue.conversationSessionGeneration) || + (typeof generation === "number" && + isWaitingConversation(issue) && + issue.executionRunId !== run.id) + ); +} /** Execution tasks may link to a conversation, but never drive its turns. * Apply before enqueue, including while a reply is still running: waiting until * finalization is too late to prevent a deferred dependency follow-up. diff --git a/server/src/services/execution-recovery-resolution.ts b/server/src/services/execution-recovery-resolution.ts index a605c7b43b..6d802b005a 100644 --- a/server/src/services/execution-recovery-resolution.ts +++ b/server/src/services/execution-recovery-resolution.ts @@ -18,6 +18,7 @@ import { type ExecutionReconciliation, } from "@paperclipai/shared"; import { parseIssueExecutionState } from "./issue-execution-policy.js"; +import { isSupersededConversationRun } from "./agent-conversations.js"; /** An operator records observed outcomes; this is not permission to blindly retry. */ export async function validateExecutionReconciliation(input: { @@ -397,6 +398,7 @@ export async function settleUnrecoverableExecutions( ) return; const current = + !isSupersededConversationRun(task, run) && action.returnOwnerAgentId !== null && task.assigneeAgentId === action.returnOwnerAgentId && !["done", "cancelled"].includes(task.status) && diff --git a/server/src/services/legacy-execution-recovery.ts b/server/src/services/legacy-execution-recovery.ts index cc1a2c0b94..6a7d5c06bf 100644 --- a/server/src/services/legacy-execution-recovery.ts +++ b/server/src/services/legacy-execution-recovery.ts @@ -5,6 +5,7 @@ import { heartbeatRuns, issueRecoveryActions, issues, type Db } from "@paperclip import { issueRecoveryActionService } from "./issue-recovery-actions.js"; import { parseIssueExecutionState } from "./issue-execution-policy.js"; import { executionFailureRetryCount } from "./execution-recovery-attempt.js"; +import { isSupersededConversationRun } from "./agent-conversations.js"; type Run = typeof heartbeatRuns.$inferSelect; export const LEGACY_RECOVERY_CAUSE = "legacy_execution_requires_reconciliation"; @@ -95,6 +96,7 @@ export async function terminalizeLegacyExecution(input: { review.currentParticipant?.type === "agent" && review.currentParticipant.agentId === run.agentId; if ( task && + !isSupersededConversationRun(task, updated) && (task.assigneeAgentId === run.agentId || isCurrentReviewer) && !["done", "cancelled"].includes(task.status) ) {