From 8d9d2f9ddfa74d6d873e74cd6baec58112694108 Mon Sep 17 00:00:00 2001 From: Young Jin Kim Date: Thu, 15 Oct 2026 03:18:23 -0400 Subject: [PATCH] fix(recovery): preserve deliberate stops and dependency waits Preserve deliberate assignee blocks and operator stops during dependency and run-release recovery. Attribute native waits to their bound owner. Ignore unrelated edits and shared-workspace exports from another issue. Continue after genuine dependency changes while preserving independently authorized user input. Keep the original contributor's implementation and review history. Add native ownership, unrelated-edit, later readiness, and shared-workspace regressions. The shared-workspace loop was reproduced before the fix. Co-Authored-By: Devin Foley Co-Authored-By: Paperclip --- doc/dependency-wake-recovery.md | 19 +++ ...eartbeat-issue-liveness-escalation.test.ts | 143 ++++++++++++++++++ ...sue-dependency-restore-wake-policy.test.ts | 45 ++++++ .../wake-queue/adapters/operator-stop.test.ts | 22 +++ .../modules/wake-queue/adapters/postgres.ts | 14 ++ .../modules/wake-queue/application/types.ts | 6 + .../wake-queue/application/use-cases.test.ts | 31 ++++ .../wake-queue/application/use-cases.ts | 8 +- server/src/routes/issues.ts | 16 +- .../src/services/issue-dependency-wakeups.ts | 32 ++++ server/src/services/recovery/service.ts | 83 ++++++++++ 11 files changed, 412 insertions(+), 7 deletions(-) create mode 100644 doc/dependency-wake-recovery.md create mode 100644 server/src/__tests__/issue-dependency-restore-wake-policy.test.ts create mode 100644 server/src/modules/wake-queue/adapters/operator-stop.test.ts diff --git a/doc/dependency-wake-recovery.md b/doc/dependency-wake-recovery.md new file mode 100644 index 00000000000..0187b99c6f3 --- /dev/null +++ b/doc/dependency-wake-recovery.md @@ -0,0 +1,19 @@ +# Dependency wakes and deliberate stops + +A task can remain blocked after all recorded dependencies finish. Its assignee +may be waiting for an external action. Blocking its own task does not request +another dependency wake. The periodic recovery sweep preserves this decision, +including native status decisions attributed to the assignee's bound run. + +Title, monitor, and other unrelated edits do not change this wait. A later +assignment, status, or blocker-set update permits reconciliation again. A blocker +that completes or finishes workspace export after the wait also permits a wake. +The export must name the blocker issue. An export from the waiting task or an +unattributed operation in the same shared workspace does not remove its wait. +Normal readiness, duplicate-wake, interaction, pause, and execution gates still +apply. A board transition into blocked keeps the existing recovery behavior. + +Cancelling a run or pausing its agent suppresses immediate stranded-work recovery. +A queued user message remains a separate continuation request. The release path +still preserves independently authorized deferred input; stopping the old run +does not discard that input or grant permission to replay its old tools. diff --git a/server/src/__tests__/heartbeat-issue-liveness-escalation.test.ts b/server/src/__tests__/heartbeat-issue-liveness-escalation.test.ts index 25916217a7d..c70afd6bb67 100644 --- a/server/src/__tests__/heartbeat-issue-liveness-escalation.test.ts +++ b/server/src/__tests__/heartbeat-issue-liveness-escalation.test.ts @@ -1173,6 +1173,149 @@ describeEmbeddedPostgres("heartbeat resolved dependency wake reconciliation", () expect(result.existingWakeSkipped).toBe(1); }); + it("does not re-wake an assignee agent that opened the current blocked cycle itself", async () => { + const { companyId, agentId, blockedIssueId } = + await seedResolvedDependencyBackstopFixture({ workspaceState: "none" }); + const blockedTransitionAt = new Date("2026-08-01T12:00:00.000Z"); + await db + .update(issues) + .set({ blockedTransitionAt, updatedAt: blockedTransitionAt }) + .where(eq(issues.id, blockedIssueId)); + await db.insert(activityLog).values({ + companyId, + actorType: "agent", + actorId: agentId, + agentId, + action: "issue.updated", + entityType: "issue", + entityId: blockedIssueId, + details: { status: "blocked", changes: { status: { from: "in_progress", to: "blocked" } } }, + createdAt: blockedTransitionAt, + }); + + const result = await heartbeatService(db).reconcileResolvedDependencyWakes(); + + expect(result.healed).toBe(0); + expect(result.selfBlockedSkipped).toBe(1); + }); + + it("heals a self-blocked issue after a later board edit in the same blocked cycle", async () => { + const { companyId, agentId, blockedIssueId } = + await seedResolvedDependencyBackstopFixture({ workspaceState: "none" }); + const blockedTransitionAt = new Date("2026-08-01T12:00:00.000Z"); + await db + .update(issues) + .set({ blockedTransitionAt, updatedAt: blockedTransitionAt }) + .where(eq(issues.id, blockedIssueId)); + await db.insert(activityLog).values([ + { + companyId, actorType: "agent", actorId: agentId, agentId, action: "issue.updated", + entityType: "issue", entityId: blockedIssueId, details: { status: "blocked" }, createdAt: blockedTransitionAt, + }, + { + companyId, actorType: "user", actorId: "board-user", action: "issue.updated", + entityType: "issue", entityId: blockedIssueId, details: { blockedByIssueIds: [] }, + createdAt: new Date("2026-08-01T12:05:00.000Z"), + }, + ]); + + const result = await heartbeatService(db).reconcileResolvedDependencyWakes(); + + expect(result.selfBlockedSkipped).toBe(0); + expect(result.healed).toBe(1); + }); + + it.each(["title", "monitorNotes"])("preserves an assignee block after an unrelated %s edit", async (field) => { + const { companyId, agentId, blockedIssueId } = + await seedResolvedDependencyBackstopFixture({ workspaceState: "none" }); + const blockedTransitionAt = new Date("2026-08-01T12:00:00.000Z"); + await db.update(issues).set({ blockedTransitionAt }).where(eq(issues.id, blockedIssueId)); + await db.insert(activityLog).values([ + { companyId, actorType: "agent", actorId: agentId, agentId, action: "issue.updated", + entityType: "issue", entityId: blockedIssueId, details: { status: "blocked" }, createdAt: blockedTransitionAt }, + { companyId, actorType: "user", actorId: "board-user", action: "issue.updated", + entityType: "issue", entityId: blockedIssueId, details: { changes: { [field]: { from: "old", to: "new" } } }, + createdAt: new Date("2026-08-01T12:05:00.000Z") }, + ]); + const result = await heartbeatService(db).reconcileResolvedDependencyWakes(); + expect(result.healed).toBe(0); + expect(result.selfBlockedSkipped).toBe(1); + }); + + it.each([true, false])("checks the native blocked decision run owner (assignee: %s)", async (sameOwner) => { + const { companyId, agentId, blockedIssueId } = + await seedResolvedDependencyBackstopFixture({ workspaceState: "none" }); + const blockedTransitionAt = new Date("2026-08-01T12:00:00.000Z"); + await db.update(issues).set({ blockedTransitionAt }).where(eq(issues.id, blockedIssueId)); + const otherAgentId = randomUUID(); + if (!sameOwner) await db.insert(agents).values({ id: otherAgentId, companyId, name: "Other agent", role: "engineer", status: "idle", adapterType: "process" }); + const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId: sameOwner ? agentId : otherAgentId, + issueId: blockedIssueId, status: "succeeded", runtimeMode: "native", contextSnapshot: { issueId: blockedIssueId } }).returning(); + await db.insert(activityLog).values({ companyId, actorType: "system", actorId: "native-status-committer", + runId: run.id, action: "issue.updated", entityType: "issue", entityId: blockedIssueId, + details: { source: "native_status_decision", fromStatus: "in_progress", toStatus: "blocked" }, createdAt: blockedTransitionAt }); + const result = await heartbeatService(db).reconcileResolvedDependencyWakes(); + expect(result.healed).toBe(sameOwner ? 0 : 1); + expect(result.selfBlockedSkipped).toBe(sameOwner ? 1 : 0); + }); + + it.each(["completion", "finalize"] as const)("heals a self-block after later blocker %s", async (event) => { + const { companyId, agentId, blockedIssueId, blockerIssueId, executionWorkspaceId } = + await seedResolvedDependencyBackstopFixture({ workspaceState: event === "finalize" ? "not_finalized" : "none" }); + const blockedTransitionAt = new Date(Date.now() - 120_000); + const completedAt = new Date(); + await db.update(issues).set({ blockedTransitionAt }).where(eq(issues.id, blockedIssueId)); + await db.insert(activityLog).values({ companyId, actorType: "agent", actorId: agentId, agentId, + action: "issue.updated", entityType: "issue", entityId: blockedIssueId, details: { status: "blocked" }, createdAt: blockedTransitionAt }); + if (event === "completion") await db.update(issues).set({ completedAt }).where(eq(issues.id, blockerIssueId)); + else await db.insert(workspaceOperations).values({ companyId, executionWorkspaceId, issueId: blockerIssueId, + phase: "workspace_finalize", status: "succeeded", startedAt: completedAt, finishedAt: completedAt }); + const result = await heartbeatService(db).reconcileResolvedDependencyWakes(); + expect(result.healed).toBe(1); + expect(result.selfBlockedSkipped).toBe(0); + }); + + it.each(["dependent", "blocker", "unattributed"] as const)("attributes shared-workspace export to its %s issue", async (exporter) => { + const { companyId, agentId, blockedIssueId, blockerIssueId, executionWorkspaceId } = + await seedResolvedDependencyBackstopFixture({ workspaceState: "finalized" }); + const blockedTransitionAt = new Date(Date.now() - 30_000); + await db.update(issues).set({ blockedTransitionAt, executionWorkspaceId }).where(eq(issues.id, blockedIssueId)); + await db.insert(activityLog).values({ companyId, actorType: "agent", actorId: agentId, agentId, + action: "issue.updated", entityType: "issue", entityId: blockedIssueId, details: { status: "blocked" }, createdAt: blockedTransitionAt }); + await db.insert(workspaceOperations).values({ companyId, executionWorkspaceId, + issueId: exporter === "dependent" ? blockedIssueId : exporter === "blocker" ? blockerIssueId : null, phase: "workspace_finalize", + status: "succeeded", startedAt: new Date(), finishedAt: new Date() }); + const result = await heartbeatService(db).reconcileResolvedDependencyWakes(); + expect(result.healed).toBe(exporter === "blocker" ? 1 : 0); + expect(result.selfBlockedSkipped).toBe(exporter === "blocker" ? 0 : 1); + }); + + it("still heals when someone other than the assignee opened the blocked cycle", async () => { + const { companyId, agentId, blockedIssueId } = + await seedResolvedDependencyBackstopFixture({ workspaceState: "none" }); + const blockedTransitionAt = new Date("2026-08-01T12:00:00.000Z"); + await db + .update(issues) + .set({ blockedTransitionAt, updatedAt: blockedTransitionAt }) + .where(eq(issues.id, blockedIssueId)); + await db.insert(activityLog).values({ + companyId, + actorType: "user", + actorId: "board-user", + action: "issue.updated", + entityType: "issue", + entityId: blockedIssueId, + details: { status: "blocked" }, + createdAt: blockedTransitionAt, + }); + + const result = await heartbeatService(db).reconcileResolvedDependencyWakes(); + + expect(result.selfBlockedSkipped).toBe(0); + expect(result.healed).toBe(1); + void agentId; + }); + it("counts null dependency wake returns as deferred instead of enqueue failures", async () => { const { companyId, agentId } = await seedResolvedDependencyBackstopFixture({ workspaceState: "none" }); diff --git a/server/src/__tests__/issue-dependency-restore-wake-policy.test.ts b/server/src/__tests__/issue-dependency-restore-wake-policy.test.ts new file mode 100644 index 00000000000..62a32534c42 --- /dev/null +++ b/server/src/__tests__/issue-dependency-restore-wake-policy.test.ts @@ -0,0 +1,45 @@ +import { describe, expect, it } from "vitest"; +import { shouldWakeOnRestoredBlockedDependency } from "../services/issue-dependency-wakeups.js"; + +const base = { + previousStatus: "in_progress", + nextStatus: "blocked", + previousAssigneeAgentId: "agent-1", + nextAssigneeAgentId: "agent-1", + blockerSetEdited: false, + actorType: "user", + actorAgentId: null, +}; + +describe("shouldWakeOnRestoredBlockedDependency", () => { + it("wakes the assignee when the board blocks an issue whose blockers are already resolved", () => { + expect(shouldWakeOnRestoredBlockedDependency(base)).toBe(true); + }); + + it("does not wake the assignee agent that blocked its own issue (re-block loop regression)", () => { + expect(shouldWakeOnRestoredBlockedDependency({ ...base, actorType: "agent", actorAgentId: "agent-1" })).toBe(false); + expect( + shouldWakeOnRestoredBlockedDependency({ ...base, actorType: "agent", actorAgentId: "agent-1", blockerSetEdited: true }), + ).toBe(false); + }); + + it("still wakes when another agent blocks the issue or the assignee changes", () => { + expect(shouldWakeOnRestoredBlockedDependency({ ...base, actorType: "agent", actorAgentId: "agent-2" })).toBe(true); + expect( + shouldWakeOnRestoredBlockedDependency({ + ...base, + actorType: "agent", + actorAgentId: "agent-2", + previousAssigneeAgentId: "agent-1", + nextAssigneeAgentId: "agent-2", + }), + ).toBe(true); + }); + + it("keeps the existing gates: already blocked with no edit, not blocked, or unassigned", () => { + expect(shouldWakeOnRestoredBlockedDependency({ ...base, previousStatus: "blocked" })).toBe(false); + expect(shouldWakeOnRestoredBlockedDependency({ ...base, previousStatus: "blocked", blockerSetEdited: true })).toBe(true); + expect(shouldWakeOnRestoredBlockedDependency({ ...base, nextStatus: "todo" })).toBe(false); + expect(shouldWakeOnRestoredBlockedDependency({ ...base, nextAssigneeAgentId: null })).toBe(false); + }); +}); diff --git a/server/src/modules/wake-queue/adapters/operator-stop.test.ts b/server/src/modules/wake-queue/adapters/operator-stop.test.ts new file mode 100644 index 00000000000..6fa25207e1f --- /dev/null +++ b/server/src/modules/wake-queue/adapters/operator-stop.test.ts @@ -0,0 +1,22 @@ +import { describe, expect, it } from "vitest"; +import { isOperatorStoppedRunRow } from "./postgres.js"; + +describe("isOperatorStoppedRunRow", () => { + it.each([ + [{ status: "cancelled", errorCode: "cancelled", resultJson: { cancelledByActorType: "user" } }, true], + [{ status: "cancelled", errorCode: "cancelled", resultJson: { cancelledByActorType: "board" } }, true], + [{ status: "cancelled", errorCode: "agent_paused", resultJson: null }, true], + [{ status: "cancelled", errorCode: "operator_interrupted", resultJson: {} }, true], + [{ status: "cancelled", errorCode: "cancelled", resultJson: { stopReason: "cancelled" } }, false], + [{ status: "cancelled", errorCode: "issue_terminal_status", resultJson: null }, false], + [{ status: "cancelled", errorCode: "cancelled", resultJson: { cancellation: { + source: "operator", expected: true, initiator: { type: "user" }, reason: "Stop", recordedAt: "2026-10-15T06:00:00Z", + } } }, true], + [{ status: "cancelled", errorCode: "operator_interrupted", resultJson: { cancelledByActorType: "user", cancellation: { + source: "queued_message", expected: true, initiator: { type: "user" }, reason: "New message", recordedAt: "2026-10-15T06:00:00Z", + } } }, false], + [{ status: "failed", errorCode: "agent_paused", resultJson: { cancelledByActorType: "user" } }, false], + ])("classifies %j as %s", (row, expected) => { + expect(isOperatorStoppedRunRow(row as never)).toBe(expected); + }); +}); diff --git a/server/src/modules/wake-queue/adapters/postgres.ts b/server/src/modules/wake-queue/adapters/postgres.ts index caf6372b2f5..480d020088d 100644 --- a/server/src/modules/wake-queue/adapters/postgres.ts +++ b/server/src/modules/wake-queue/adapters/postgres.ts @@ -98,9 +98,23 @@ function toRunSnapshot(row: HeartbeatRunRow): RunSnapshot { responsibleUserId: row.responsibleUserId, contextSnapshot: parseObject(row.contextSnapshot), configurationIncompletePayload: Object.keys(configurationIncompletePayload).length > 0 ? configurationIncompletePayload : null, + operatorStopped: isOperatorStoppedRunRow(row), }; } +const OPERATOR_STOP_ERROR_CODES = new Set(["operator_interrupted", "agent_paused"]); + +/** Mirrors recovery's `isOperatorCancelledRun`, plus agent pause, for the release-time recovery tail. */ +export function isOperatorStoppedRunRow(row: Pick): boolean { + if (row.status !== "cancelled") return false; + const cancellation = readRunCancellation(row.resultJson); + if (cancellation?.source === "queued_message") return false; + if (cancellation?.source === "operator" && cancellation.expected) return true; + if (row.errorCode && OPERATOR_STOP_ERROR_CODES.has(row.errorCode)) return true; + const actorType = parseObject(row.resultJson).cancelledByActorType; + return actorType === "user" || actorType === "board"; +} + function toIssueSnapshot(row: IssueRow): IssueSnapshot { return { conversationAgentId: row.conversationAgentId, diff --git a/server/src/modules/wake-queue/application/types.ts b/server/src/modules/wake-queue/application/types.ts index d3b620f0bd8..080a705bc80 100644 --- a/server/src/modules/wake-queue/application/types.ts +++ b/server/src/modules/wake-queue/application/types.ts @@ -23,6 +23,12 @@ export type RunSnapshot = { configurationIncompletePayload: Record | null; /** The host schedules failed conversation turns with its durable retry budget. */ conversationContinuation?: boolean; + /** + * True when a person deliberately stopped this run: the board cancel route + * (`resultJson.cancelledByActorType` of `user`/`board`), an operator interrupt, + * or an agent pause. Release must not immediately re-queue such a run. + */ + operatorStopped?: boolean; }; export type IssueSnapshot = { diff --git a/server/src/modules/wake-queue/application/use-cases.test.ts b/server/src/modules/wake-queue/application/use-cases.test.ts index f17fad36b3b..8f2cff02ab0 100644 --- a/server/src/modules/wake-queue/application/use-cases.test.ts +++ b/server/src/modules/wake-queue/application/use-cases.test.ts @@ -698,6 +698,37 @@ describe("releaseIssueExecution", () => { expect(queueCall.contextSnapshot).toBe(resolveCall.contextSnapshot); }); + it.each([ + ["board cancel", { errorCode: "cancelled", operatorStopped: true }], + ["agent pause", { errorCode: "agent_paused", operatorStopped: true }], + ] as const)("does not re-queue or escalate a run a person stopped (%s)", async (_label, stop) => { + // Regression: a board cancel of a queued wake re-created it as + // issue_continuation_needed / issue_assignment_recovery within a second. + for (const status of ["in_progress", "todo"]) { + const transaction = createFakeTransaction(); + const recovery = createFakeRecovery(); + const release = createReleaseIssueExecution({ + issueLock: createFakeIssueLock(createFakeHost(), transaction, { ...ISSUE, status }, { ...RUN, status: "cancelled", ...stop }), + recovery, + }); + const result = await release({ companyId: RUN.companyId, runId: RUN.id, now: new Date() }); + expect(result.outcome.kind).toBe("released"); + expect(transaction.queueImmediateRecoveryRun).not.toHaveBeenCalled(); + expect(recovery.escalateStrandedAssignedIssue).not.toHaveBeenCalled(); + } + }); + + it("still recovers a control-plane cancelled run that no person stopped", async () => { + const transaction = createFakeTransaction(); + const release = createReleaseIssueExecution({ + issueLock: createFakeIssueLock(createFakeHost(), transaction, ISSUE, { ...RUN, status: "cancelled", errorCode: "cancelled", operatorStopped: false }), + recovery: createFakeRecovery(), + }); + const result = await release({ companyId: RUN.companyId, runId: RUN.id, now: new Date() }); + expect(result.outcome.kind).toBe("queued_recovery"); + expect(transaction.queueImmediateRecoveryRun).toHaveBeenCalledTimes(1); + }); + it("throws WakeQueueApplicationError with code responsible_user_unresolved for a recovery run, without queuing it", async () => { const transaction = createFakeTransaction(); const host = createFakeHost({ resolveResponsibleUserId: vi.fn(async () => null) }); diff --git a/server/src/modules/wake-queue/application/use-cases.ts b/server/src/modules/wake-queue/application/use-cases.ts index 7be80c89457..0cbceec1b63 100644 --- a/server/src/modules/wake-queue/application/use-cases.ts +++ b/server/src/modules/wake-queue/application/use-cases.ts @@ -493,7 +493,13 @@ async function runReleaseRecoveryTail( input: ReleaseIssueExecutionInput, postCommitEffects: PostCommitEffect[], ): Promise { - const suppressImmediateRecovery = input.suppressImmediateRecovery === true || Boolean( + // A run a person deliberately stopped (board cancel, operator interrupt, agent + // pause) is not stranded work. Re-queuing it here re-created the cancelled wake + // within a second of the cancel (issue_continuation_needed / assignment_recovery), + // and for a paused agent escalated the issue to `blocked`. The periodic sweep + // already stands down for operator cancels; this keeps release consistent. + const operatorStopped = run.status === "cancelled" && run.operatorStopped === true; + const suppressImmediateRecovery = input.suppressImmediateRecovery === true || operatorStopped || Boolean( issue.conversationAgentId && issue.conversationUserId && issue.conversationState === "waiting" && issue.status === "in_review" ); diff --git a/server/src/routes/issues.ts b/server/src/routes/issues.ts index 0265eb18db3..b89f2d6c139 100644 --- a/server/src/routes/issues.ts +++ b/server/src/routes/issues.ts @@ -283,6 +283,7 @@ import { ISSUE_BLOCKERS_RESOLVED_WAKE_REASON, buildIssueBlockersResolvedWakeStateKey, findExistingIssueBlockersResolvedWakeForReadyState, + shouldWakeOnRestoredBlockedDependency, } from "../services/issue-dependency-wakeups.js"; import { assertEnvironmentSelectionForCompany } from "./environment-selection.js"; import { @@ -15648,12 +15649,15 @@ export function issueRoutes( } } - const restoredBlockedReadyDependency = - issue.status === "blocked" && - issue.assigneeAgentId && - (existing.status !== "blocked" || - Array.isArray(req.body.blockedByIssueIds) || - existing.assigneeAgentId !== issue.assigneeAgentId); + const restoredBlockedReadyDependency = shouldWakeOnRestoredBlockedDependency({ + previousStatus: existing.status, + nextStatus: issue.status, + previousAssigneeAgentId: existing.assigneeAgentId ?? null, + nextAssigneeAgentId: issue.assigneeAgentId ?? null, + blockerSetEdited: Array.isArray(req.body.blockedByIssueIds), + actorType: actor.actorType, + actorAgentId: actor.agentId ?? null, + }); if ( restoredBlockedReadyDependency && typeof dependencyReadinessSvc.getDependencyReadiness === "function" diff --git a/server/src/services/issue-dependency-wakeups.ts b/server/src/services/issue-dependency-wakeups.ts index c40d7ff0fde..ee09f3e55ff 100644 --- a/server/src/services/issue-dependency-wakeups.ts +++ b/server/src/services/issue-dependency-wakeups.ts @@ -5,6 +5,38 @@ import { agentWakeupRequests } from "@paperclipai/db"; export const ISSUE_BLOCKERS_RESOLVED_WAKE_REASON = "issue_blockers_resolved"; +/** + * Whether moving an issue into `blocked` should immediately wake its assignee + * because every recorded blocker is already resolved ("blockers restored"). + * + * The assignee agent blocking its own issue is declaring a wait on something + * outside the blocker graph (a pending confirmation, a human action). Waking it + * straight back made it re-block in a loop: every re-block stamps a new + * `blockedTransitionAt`, so the ready-state dedup key never matched and the + * agent was woken again on each run (five wakes in four minutes observed). + * Board or other-actor transitions and assignee changes still wake, and the + * dependency backstop still covers a genuinely stranded issue. + */ +export function shouldWakeOnRestoredBlockedDependency(input: { + previousStatus: string; + nextStatus: string; + previousAssigneeAgentId: string | null; + nextAssigneeAgentId: string | null; + blockerSetEdited: boolean; + actorType: string; + actorAgentId: string | null; +}): boolean { + if (input.nextStatus !== "blocked" || !input.nextAssigneeAgentId) return false; + const assigneeChanged = input.previousAssigneeAgentId !== input.nextAssigneeAgentId; + if (input.previousStatus === "blocked" && !input.blockerSetEdited && !assigneeChanged) return false; + const selfBlockedByAssignee = + input.actorType === "agent" && + input.actorAgentId !== null && + input.actorAgentId === input.nextAssigneeAgentId && + !assigneeChanged; + return !selfBlockedByAssignee; +} + // A wake counts as "already delivered or in flight for the current ready state" // for these statuses. The level-triggered state key uses this full set so that // one wake for a ready state suppresses further wakes for the SAME state. This diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index 1248f6cfb9a..9d2d0696eeb 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -69,6 +69,7 @@ import { statusDecisions, routines, workAssessments, + workspaceOperations, } from "@paperclipai/db"; import { parseObject, asBoolean, asNumber } from "../../adapters/utils.js"; import { runningProcesses } from "../../adapters/index.js"; @@ -1143,6 +1144,66 @@ export function recoveryService( return Boolean(run || deferredWake || nativeRecovery); } + async function wasBlockedByAssigneeAgent( + companyId: string, + issueId: string, + agentId: string, + blockedTransitionAt: Date | null, + blockerIssueIds: string[], + ) { + if (!blockedTransitionAt) return false; + const row = await db + .select({ + actorType: activityLog.actorType, + actorId: activityLog.actorId, + agentId: activityLog.agentId, + runAgentId: heartbeatRuns.agentId, + details: activityLog.details, + }) + .from(activityLog) + .leftJoin(heartbeatRuns, and( + eq(heartbeatRuns.id, activityLog.runId), + eq(heartbeatRuns.companyId, companyId), + or(eq(heartbeatRuns.issueId, issueId), sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueId}`), + )) + .where( + and( + eq(activityLog.companyId, companyId), + eq(activityLog.entityType, "issue"), + eq(activityLog.entityId, issueId), + eq(activityLog.action, "issue.updated"), + gte(activityLog.createdAt, new Date(blockedTransitionAt.getTime() - 5_000)), + sql`( + ${activityLog.details} ?| array['status', 'toStatus', 'assigneeAgentId', 'assigneeUserId', 'blockedByIssueIds'] + or (${activityLog.details}->'changes') ?| array['status', 'assigneeAgentId', 'assigneeUserId', 'blockedByIssueIds'] + )`, + ), + ) + .orderBy(desc(activityLog.createdAt), desc(activityLog.id)) + .limit(1) + .then((rows) => rows[0] ?? null); + const details = row?.details ?? {}; + const selfBlocked = row?.actorType === "agent" && row.agentId === agentId && details.status === "blocked"; + const nativeSelfBlocked = row?.actorType === "system" && + row.actorId === "native-status-committer" && row.runAgentId === agentId && + details.source === "native_status_decision" && details.toStatus === "blocked"; + if (!selfBlocked && !nativeSelfBlocked) return false; + const newlyCompletedBlocker = await db.select({ id: issues.id }).from(issues).where(and( + eq(issues.companyId, companyId), + inArray(issues.id, blockerIssueIds), + gt(issues.completedAt, blockedTransitionAt), + )).limit(1); + if (newlyCompletedBlocker.length > 0) return false; + const newlyFinalizedBlocker = await db.select({ id: workspaceOperations.id }).from(workspaceOperations) + .innerJoin(issues, and(eq(issues.companyId, companyId), inArray(issues.id, blockerIssueIds), + eq(issues.executionWorkspaceId, workspaceOperations.executionWorkspaceId), + eq(workspaceOperations.issueId, issues.id))) + .where(and(eq(workspaceOperations.companyId, companyId), eq(workspaceOperations.phase, "workspace_finalize"), + eq(workspaceOperations.status, "succeeded"), gt(workspaceOperations.finishedAt, blockedTransitionAt))) + .limit(1); + return newlyFinalizedBlocker.length === 0; + } + async function hasPendingWakeInteraction(companyId: string, issueId: string) { return db .select({ id: issueThreadInteractions.id }) @@ -4514,6 +4575,19 @@ export function recoveryService( result.skipped += 1; continue; } + // A person paused this agent (or stopped its run) on purpose. Escalating its + // assigned work to `blocked` while the pause lasts fights that decision and + // strands the issue after resume; leave it for the resumed agent. + if ( + agent?.status === "paused" && + agent.companyId === issue.companyId && + latestRun?.agentId === agentId && + latestRun.status === "cancelled" && + (latestRun.errorCode === "agent_paused" || isOperatorCancelledRun(latestRun, agentId)) + ) { + result.skipped += 1; + continue; + } if (issue.status !== "in_review" && !agentInvokable) { const classification = classifyContinuationFailure(latestRun); if ( @@ -5567,6 +5641,7 @@ export function recoveryService( checked: 0, healed: 0, existingWakeSkipped: 0, + selfBlockedSkipped: 0, livePathSkipped: 0, interactionSkipped: 0, pauseHoldSkipped: 0, @@ -5733,6 +5808,14 @@ export function recoveryService( result.existingWakeSkipped += 1; continue; } + // The assignee agent opened this blocked cycle itself, knowing its recorded + // blockers were resolved: it is waiting on something outside the blocker + // graph. Re-waking it here made it re-block on every sweep (each re-block + // stamps a new cycle, so the ready-state key never repeats). + if (await wasBlockedByAssigneeAgent(companyId, candidate.id, agentId, candidate.blockedTransitionAt, readiness.blockerIssueIds)) { + result.selfBlockedSkipped += 1; + continue; + } if ( (await hasActiveExecutionPath(companyId, candidate.id, agentId)) ||