mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-27 02:44:50 +01:00
fix(server): restore blocked recovery visibility
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
3e348b96b9
commit
fe2a243b9c
7 files changed
+171
-4
No files matched your search
@@ -1303,6 +1303,80 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
|
||||
expect(checkoutReleasedIssue?.checkoutRunId).toBeNull();
|
||||
});
|
||||
|
||||
it("restores one lost monitor dispatch before escalating a second process loss", async () => {
|
||||
const { companyId, agentId, runId, issueId } = await seedRunFixture({
|
||||
adapterType: "openclaw_gateway",
|
||||
agentStatus: "idle",
|
||||
processPid: null,
|
||||
processGroupId: null,
|
||||
contextSnapshot: {
|
||||
wakeReason: "issue_monitor_due",
|
||||
nextCheckAt: "2026-03-19T00:00:00.000Z",
|
||||
},
|
||||
});
|
||||
const heartbeat = heartbeatService(db);
|
||||
|
||||
const firstLoss = await heartbeat.reapOrphanedRuns();
|
||||
expect(firstLoss).toEqual({ reaped: 1, runIds: [runId] });
|
||||
|
||||
const firstRetry = await db
|
||||
.select()
|
||||
.from(heartbeatRuns)
|
||||
.where(and(eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.retryOfRunId, runId)))
|
||||
.then((rows) => rows[0] ?? null);
|
||||
expect(firstRetry).toMatchObject({ processLossRetryCount: 1 });
|
||||
expect(firstRetry?.contextSnapshot).toMatchObject({
|
||||
wakeReason: "process_lost_retry",
|
||||
retryReason: "issue_continuation_needed",
|
||||
retryOfRunId: runId,
|
||||
});
|
||||
|
||||
const secondAttempt = await seedRunFixture({
|
||||
adapterType: "openclaw_gateway",
|
||||
agentStatus: "idle",
|
||||
processPid: null,
|
||||
processGroupId: null,
|
||||
processLossRetryCount: 1,
|
||||
contextSnapshot: {
|
||||
wakeReason: "process_lost_retry",
|
||||
retryReason: "issue_continuation_needed",
|
||||
retryOfRunId: runId,
|
||||
},
|
||||
});
|
||||
|
||||
const secondLoss = await heartbeat.reapOrphanedRuns();
|
||||
expect(secondLoss).toEqual({ reaped: 1, runIds: [secondAttempt.runId] });
|
||||
|
||||
const secondAttemptRuns = await db
|
||||
.select()
|
||||
.from(heartbeatRuns)
|
||||
.where(eq(heartbeatRuns.agentId, secondAttempt.agentId));
|
||||
expect(secondAttemptRuns.find((run) => run.id === secondAttempt.runId)).toMatchObject({
|
||||
id: secondAttempt.runId,
|
||||
status: "failed",
|
||||
errorCode: "process_lost",
|
||||
processLossRetryCount: 1,
|
||||
});
|
||||
expect(secondAttemptRuns.some((run) => run.processLossRetryCount > 1)).toBe(false);
|
||||
|
||||
const issue = await waitForValue(async () =>
|
||||
db.select().from(issues).where(eq(issues.id, secondAttempt.issueId)).then((rows) => {
|
||||
const row = rows[0] ?? null;
|
||||
return row?.status === "blocked" ? row : null;
|
||||
})
|
||||
);
|
||||
expect(issue?.monitorNextCheckAt).toBeNull();
|
||||
|
||||
await expectSourceScopedStrandedRecoveryAction({
|
||||
companyId: secondAttempt.companyId,
|
||||
agentId: secondAttempt.agentId,
|
||||
issueId: secondAttempt.issueId,
|
||||
runId: secondAttempt.runId,
|
||||
previousStatus: "in_progress",
|
||||
retryReason: "issue_continuation_needed",
|
||||
});
|
||||
});
|
||||
|
||||
it("interrupts running runs on graceful shutdown and queues restart recovery without recording a failure", async () => {
|
||||
const { agentId, runId, issueId, wakeupRequestId } = await seedRunFixture({
|
||||
agentStatus: "running",
|
||||
|
||||
@@ -581,6 +581,37 @@ describeEmbeddedPostgres("issue blocker attention", () => {
|
||||
await expect(svc.count(companyId, { attention: "blocked" })).resolves.toBe(1);
|
||||
});
|
||||
|
||||
it("surfaces cancelled-blocker attention on an assigned todo source", async () => {
|
||||
const { companyId, agentId } = await createCompany("BICX");
|
||||
const sourceId = await insertIssue({
|
||||
companyId,
|
||||
identifier: "BICX-1",
|
||||
title: "Dispatch-suppressed source",
|
||||
status: "todo",
|
||||
assigneeAgentId: agentId,
|
||||
});
|
||||
const blockerId = await insertIssue({
|
||||
companyId,
|
||||
identifier: "BICX-2",
|
||||
title: "Cancelled dependency",
|
||||
status: "cancelled",
|
||||
assigneeAgentId: agentId,
|
||||
});
|
||||
await block({ companyId, blockerIssueId: blockerId, blockedIssueId: sourceId });
|
||||
|
||||
const rows = await svc.list(companyId, { attention: "blocked" });
|
||||
const source = rows.find((issue) => issue.id === sourceId);
|
||||
|
||||
expect(source?.blockedInboxAttention).toMatchObject({
|
||||
kind: "blocked",
|
||||
state: "needs_attention",
|
||||
reason: "blocked_by_cancelled_issue",
|
||||
owner: { type: "agent", agentId },
|
||||
action: { label: "Replace blocker" },
|
||||
leafIssue: { id: blockerId, identifier: "BICX-2" },
|
||||
});
|
||||
});
|
||||
|
||||
it("redacts external wait details from blocked inbox payloads and search", async () => {
|
||||
const { companyId } = await createCompany("BIX");
|
||||
const owner = "Private Vendor Security Team";
|
||||
|
||||
@@ -273,6 +273,30 @@ describe("issue graph liveness classifier", () => {
|
||||
expect(paused[0]?.state).toBe("blocked_by_uninvokable_assignee");
|
||||
});
|
||||
|
||||
it("detects a cancelled blocker on an assigned todo source", () => {
|
||||
const findings = classifyIssueGraphLiveness({
|
||||
issues: [
|
||||
issue({ status: "todo" }),
|
||||
issue({
|
||||
id: blockerId,
|
||||
identifier: "PAP-1704",
|
||||
title: "Cancelled unblock work",
|
||||
status: "cancelled",
|
||||
assigneeAgentId: "blocker-agent",
|
||||
}),
|
||||
],
|
||||
relations: blocks,
|
||||
agents: [agent(), manager, agent({ id: "blocker-agent", name: "Cancelled owner" })],
|
||||
});
|
||||
|
||||
expect(findings).toHaveLength(1);
|
||||
expect(findings[0]).toMatchObject({
|
||||
issueId: blockedId,
|
||||
state: "blocked_by_cancelled_issue",
|
||||
recoveryIssueId: blockerId,
|
||||
});
|
||||
});
|
||||
|
||||
it("detects blocker assignees under terminated org ancestors as uninvokable", () => {
|
||||
const findings = classifyIssueGraphLiveness({
|
||||
issues: [
|
||||
|
||||
@@ -8509,13 +8509,16 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
|
||||
const contextSnapshot = parseObject(run.contextSnapshot);
|
||||
const issueId = readNonEmptyString(contextSnapshot.issueId);
|
||||
const retryReason = readNonEmptyString(contextSnapshot.wakeReason) === "issue_monitor_due"
|
||||
? "issue_continuation_needed"
|
||||
: "process_lost";
|
||||
const taskKey = deriveTaskKeyWithHeartbeatFallback(contextSnapshot, null);
|
||||
const sessionBefore = await resolveSessionBeforeForWakeup(agent, taskKey);
|
||||
const retryContextSnapshot = withRecoveryModelProfileHint({
|
||||
...contextSnapshot,
|
||||
retryOfRunId: run.id,
|
||||
wakeReason: "process_lost_retry",
|
||||
retryReason: "process_lost",
|
||||
retryReason,
|
||||
}, "normal_model");
|
||||
const responsibleUserId = await resolveResponsibleUserIdForRunContext(run, retryContextSnapshot);
|
||||
|
||||
@@ -10965,7 +10968,26 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
});
|
||||
}
|
||||
|
||||
const shouldRetry = tracksLocalChild && (!!run.processPid || !!run.processGroupId) && (run.processLossRetryCount ?? 0) < 1;
|
||||
const runContext = parseObject(run.contextSnapshot);
|
||||
const monitorIssueId = readNonEmptyString(runContext.issueId);
|
||||
const monitorDispatchLostWithoutFutureWake =
|
||||
readNonEmptyString(runContext.wakeReason) === "issue_monitor_due" &&
|
||||
Boolean(monitorIssueId) &&
|
||||
await db
|
||||
.select({ monitorNextCheckAt: issues.monitorNextCheckAt })
|
||||
.from(issues)
|
||||
.where(and(eq(issues.id, monitorIssueId!), eq(issues.companyId, run.companyId)))
|
||||
.limit(1)
|
||||
.then((rows) => {
|
||||
const issue = rows[0];
|
||||
if (!issue) return false;
|
||||
const nextCheckAt = issue.monitorNextCheckAt;
|
||||
return !nextCheckAt || nextCheckAt.getTime() <= now.getTime();
|
||||
});
|
||||
const shouldRetry = (run.processLossRetryCount ?? 0) < 1 && (
|
||||
(tracksLocalChild && (!!run.processPid || !!run.processGroupId)) ||
|
||||
monitorDispatchLostWithoutFutureWake
|
||||
);
|
||||
const baseMessage = buildProcessLossMessage(run, descendantOnlyCleanup ? { descendantOnly: true } : undefined);
|
||||
const unmanagedBackgroundTaskEvidence = descendantOnlyCleanup
|
||||
? {
|
||||
|
||||
@@ -2985,7 +2985,7 @@ async function listIssueBlockedInboxAttentionMap(
|
||||
.where(and(
|
||||
eq(issues.companyId, companyId),
|
||||
visibleIssueCondition(),
|
||||
notInArray(issues.status, [...BLOCKED_INBOX_TERMINAL_STATUSES]),
|
||||
ne(issues.status, "done"),
|
||||
)),
|
||||
dbOrTx
|
||||
.select({
|
||||
|
||||
@@ -590,7 +590,19 @@ export function classifyIssueGraphLiveness(input: IssueGraphLivenessInput): Issu
|
||||
}
|
||||
|
||||
for (const issue of input.issues) {
|
||||
if (issue.status === "blocked") {
|
||||
const hasUnresolvedBlockerEdge = (blockersByBlockedIssueId.get(issue.id) ?? []).some((relation) => {
|
||||
if (relation.companyId !== issue.companyId) return false;
|
||||
const blocker = issuesById.get(relation.blockerIssueId);
|
||||
return Boolean(blocker && blocker.companyId === issue.companyId && blocker.status !== "done");
|
||||
});
|
||||
const shouldInspectBlockedChain = issue.status === "blocked" || (
|
||||
issue.status !== "done" &&
|
||||
issue.status !== "cancelled" &&
|
||||
Boolean(issue.assigneeAgentId) &&
|
||||
hasUnresolvedBlockerEdge
|
||||
);
|
||||
|
||||
if (shouldInspectBlockedChain) {
|
||||
if (unresolvedBlockers.has(issue.id)) continue;
|
||||
const chainFinding = firstBlockedChainFinding(issue, issue, [issue], new Set());
|
||||
if (chainFinding) findings.push(chainFinding);
|
||||
|
||||
Reference in new issue
Block a user