diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index b8d05bbb32..973d35c0e7 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -552,6 +552,8 @@ Recovery rule: This is an active-work continuity recovery. +After a productive successful run, recovery checks that the issue is still `in_progress` and assigned to the same agent under the enqueue transaction's issue lock. The sweep's earlier snapshot cannot authorize a continuation after completion, cancellation, reassignment, or a move to another status. A mismatch records a skipped wake receipt without creating a run. An empty queued continuation cancelled because the issue became terminal is omitted from task chat; its cancellation remains in the run log. Runs that actually started still show their stop state. + The same bounded rule applies when the previous heartbeat reported waiting on a local/background watcher and that watcher was killed, disappeared, or was never represented by a durable Paperclip primitive. Paperclip queues at most one continuation for the same recovery fingerprint. If the continuation also leaves only local watcher evidence, Paperclip must surface a real blocker or explicit recovery action instead of repeating continuation recovery. A new monitor, scheduled wake, healthy delegated blocker issue, or other durable source mutation resolves that recovery fingerprint normally. #### Deliberate wait is not a lost run diff --git a/server/src/__tests__/heartbeat-stale-queue-invalidation.test.ts b/server/src/__tests__/heartbeat-stale-queue-invalidation.test.ts index ba9fad4ecf..faf7f719c9 100644 --- a/server/src/__tests__/heartbeat-stale-queue-invalidation.test.ts +++ b/server/src/__tests__/heartbeat-stale-queue-invalidation.test.ts @@ -25,6 +25,7 @@ import { heartbeatService, } from "../services/heartbeat.ts"; import { runningProcesses } from "../adapters/index.ts"; +import { recoveryService } from "../services/recovery/service.ts"; const mockAdapterExecute = vi.hoisted(() => vi.fn(async () => ({ @@ -395,6 +396,93 @@ describeEmbeddedPostgres("heartbeat stale queued-run invalidation", () => { ]); }); + it.each([ + { runtimeMode: "native", status: "done", reassigned: false }, + { runtimeMode: "legacy", status: "done", reassigned: false }, + { runtimeMode: "native", status: "cancelled", reassigned: false }, + { runtimeMode: "native", status: "backlog", reassigned: false }, + { runtimeMode: "native", status: "in_review", reassigned: false }, + { runtimeMode: "native", status: "blocked", reassigned: false }, + { runtimeMode: "native", status: "in_progress", reassigned: true }, + ] as const)("skips stale $runtimeMode productive recovery after status=$status reassigned=$reassigned commits under the enqueue lock", async ({ runtimeMode, status, reassigned }) => { + const { companyId, agentId } = await seedCompanyAndAgent(); + const issueId = randomUUID(); + const runId = randomUUID(); + await db.insert(issues).values({ + id: issueId, + companyId, + title: "Completion racing with productive recovery", + status: "in_progress", + assigneeAgentId: agentId, + }); + await db.insert(heartbeatRuns).values({ + id: runId, + companyId, + agentId, + invocationSource: "assignment", + runtimeMode, + status: "succeeded", + livenessState: "completed", + contextSnapshot: { issueId, wakeReason: "issue_assigned" }, + startedAt: new Date(), + finishedAt: new Date(), + }); + + // Let the real sweep select its in-progress snapshot, then hold the issue + // lock until the real enqueue transaction is waiting on the newer state. + const enqueueWakeup = vi.fn(async (...[targetAgentId, options]: Parameters) => { + let pendingWake!: ReturnType; + await db.transaction(async (tx) => { + await tx.update(issues).set({ + status, + ...(reassigned ? { assigneeAgentId: null, assigneeUserId: "responsible-user" } : {}), + }).where(eq(issues.id, issueId)); + const [{ pid }] = await tx.execute<{ pid: number }>(sql`select pg_backend_pid() as pid`); + pendingWake = heartbeat.wakeup(targetAgentId, options); + try { + expect(await waitForCondition(async () => { + const [{ waiting }] = await db.execute<{ waiting: boolean }>(sql` + select exists ( + select 1 from pg_stat_activity + where ${pid} = any(pg_blocking_pids(pid)) + ) as waiting + `); + return waiting; + })).toBe(true); + } catch (error) { + // Observe a pending rejection even if the lock assertion fails. + void pendingWake.catch(() => {}); + throw error; + } + }); + return pendingWake; + }); + const recovery = recoveryService(db, { enqueueWakeup }); + const result = await recovery.reconcileStrandedAssignedIssues(); + + expect(enqueueWakeup).toHaveBeenCalledOnce(); + expect(result).toMatchObject({ continuationRequeued: 0, escalated: 0, skipped: 1, issueIds: [] }); + expect(await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns)).toEqual([{ id: runId }]); + expect(await db.select().from(issueComments)).toHaveLength(0); + expect(mockAdapterExecute).not.toHaveBeenCalled(); + const [wakeup] = await db.select().from(agentWakeupRequests); + expect(wakeup).toMatchObject({ + status: "skipped", + reason: "issue_state_guard_mismatch", + runId: null, + payload: { + heartbeatSkip: { + expectedStatuses: ["in_progress"], + actualStatus: status, + expectedAssigneeAgentId: agentId, + actualAssigneeAgentId: reassigned ? null : agentId, + }, + }, + }); + const [issue] = await db.select().from(issues).where(eq(issues.id, issueId)); + expect(issue).toMatchObject({ status, assigneeAgentId: reassigned ? null : agentId }); + }); + it("cancels a resolved connection-intent wake parked before queued-run claim", async () => { const { companyId, agentId } = await seedCompanyAndAgent(); const issueId = randomUUID(); diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index be14888d3b..0361621ec1 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -182,6 +182,10 @@ type RecoveryWakeupOptions = { requestedByActorType?: "user" | "agent" | "system"; requestedByActorId?: string | null; contextSnapshot?: Record; + issueStateGuard?: { + statuses: string[]; + assigneeAgentId: string; + }; }; type RecoveryWakeup = ( @@ -1921,6 +1925,17 @@ export function recoveryService( source: "automation", triggerDetail: "system", reason: input.reason, + // The sweep can combine an old in-progress issue snapshot with a newer + // successful run. Validate eligibility under the enqueue issue lock so + // completion or reassignment cannot create a redundant continuation. + ...(input.source === "issue.productive_terminal_continuation_recovery" + ? { + issueStateGuard: { + statuses: ["in_progress"], + assigneeAgentId: input.agentId, + }, + } + : {}), payload: withRecoveryContext( { issueId: input.issueId, diff --git a/ui/src/components/TaskChatThread.test.tsx b/ui/src/components/TaskChatThread.test.tsx index 533ef704d1..a5199d77af 100644 --- a/ui/src/components/TaskChatThread.test.tsx +++ b/ui/src/components/TaskChatThread.test.tsx @@ -1229,31 +1229,72 @@ describe("TaskChatThread runtime transcript selection", () => { }, ); - it("does not show a completed-response notice for a redundant cancelled continuation", () => { - render( - {}} - linkedRuns={[ - { - runId: "connection-continuation-skipped", - status: "cancelled", - errorCode: "issue_not_in_progress", - startedAt: null, - agentId: "agent-1", - agentName: "Runner", - adapterType: "paperclip_runner", - createdAt: "2026-09-07T18:00:00.000Z", - finishedAt: "2026-09-07T18:00:01.000Z", - }, - ]} - />, - ); - expect(container.textContent).not.toContain( - "The runner returned no user-facing response.", - ); - expect(container.textContent).not.toContain("Run completed"); - }); + it.each([ + ["legacy", "issue_not_in_progress"], + ["native", "issue_not_in_progress"], + ["legacy", "issue_terminal_status"], + ["native", "issue_terminal_status"], + ] as const)( + "hides a redundant cancelled continuation (%s, %s)", + (runtimeMode, errorCode) => { + render( + {}} + linkedRuns={[ + { + runId: "connection-continuation-skipped", + runtimeMode, + status: "cancelled", + errorCode, + startedAt: null, + agentId: "agent-1", + agentName: "Runner", + adapterType: "paperclip_runner", + createdAt: "2026-09-07T18:00:00.000Z", + finishedAt: "2026-09-07T18:00:01.000Z", + }, + ]} + />, + ); + expect(container.textContent).not.toContain( + "The runner returned no user-facing response.", + ); + expect(container.textContent).not.toContain("Run completed"); + expect(container.textContent).not.toContain("Couldn't start"); + expect(container.textContent).not.toContain("Run cancelled"); + expect(container.textContent).not.toContain("before returning an answer"); + }, + ); + + it.each(["legacy", "native"] as const)( + "keeps a cancellation visible when the %s run had already started", + (runtimeMode) => { + render( + {}} + linkedRuns={[ + { + runId: "started-cancellation", + runtimeMode, + status: "cancelled", + errorCode: "issue_terminal_status", + agentId: "agent-1", + agentName: "Runner", + adapterType: "paperclip_runner", + createdAt: "2026-09-07T18:00:00.000Z", + startedAt: "2026-09-07T18:00:00.500Z", + finishedAt: "2026-09-07T18:00:01.000Z", + }, + ]} + />, + ); + expect(container.textContent).toContain( + runtimeMode === "native" ? "Run cancelled" : "Stopped", + ); + }, + ); it("does not treat a progress comment as the final response of a failed native run", () => { nativeTranscriptState.transcriptByRun.set("native-progress-failed", [ diff --git a/ui/src/components/TaskChatThread.tsx b/ui/src/components/TaskChatThread.tsx index 5335b102ab..1fff76d2db 100644 --- a/ui/src/components/TaskChatThread.tsx +++ b/ui/src/components/TaskChatThread.tsx @@ -1384,6 +1384,18 @@ export function TaskChatThread(props: TaskChatThreadProps) { if (liveRun && source.id === liveRun.id) continue; const entries = transcriptByRun.get(source.id) ?? []; const meta = linkedRunMetaById.get(source.id); + // A queued continuation can become unnecessary while another turn finishes + // the task. Keep that cancellation in the run log, not the conversation. + // Apply this before native stop markers are assembled as well. + if ( + source.status === "cancelled" && + entries.length === 0 && + (meta?.errorCode === "issue_not_in_progress" || + (meta?.errorCode === "issue_terminal_status" && !meta.startedAt)) + ) { + settledRunIds.add(source.id); + continue; + } const acceptedSummary = acceptedSemanticResultSummary(meta?.resultJson); const parsedSource = transcriptToTaskChatItems(entries, { runId: source.id, @@ -1516,16 +1528,6 @@ export function TaskChatThread(props: TaskChatThreadProps) { }); } if (entries.length === 0) { - // A queued continuation cancelled after the task was completed or parked - // never produced a provider turn. Keep its record in the run log without - // presenting it as a completed chat response. - if ( - source.status === "cancelled" && - meta?.errorCode === "issue_not_in_progress" - ) { - settledRunIds.add(source.id); - continue; - } if (sourceIsPaperclipRunner && sourceYielded) { settledRunIds.add(source.id); continue;