mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-30 21:49:18 +01:00
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 <devin@paperclip.ing> Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
11 files changed
+412
-7
No files matched your search
@@ -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.
|
||||
@@ -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" });
|
||||
|
||||
@@ -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);
|
||||
});
|
||||
});
|
||||
@@ -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);
|
||||
});
|
||||
});
|
||||
@@ -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<HeartbeatRunRow, "status" | "errorCode" | "resultJson">): 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,
|
||||
|
||||
@@ -23,6 +23,12 @@ export type RunSnapshot = {
|
||||
configurationIncompletePayload: Record<string, unknown> | 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 = {
|
||||
|
||||
@@ -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) });
|
||||
|
||||
@@ -493,7 +493,13 @@ async function runReleaseRecoveryTail(
|
||||
input: ReleaseIssueExecutionInput,
|
||||
postCommitEffects: PostCommitEffect[],
|
||||
): Promise<ReleaseTransactionResult> {
|
||||
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"
|
||||
);
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)) ||
|
||||
|
||||
Reference in new issue
Block a user