From 3ea2ff41874890c9a62b177f390cfe5fb0c81dc1 Mon Sep 17 00:00:00 2001 From: Dotta Date: Tue, 28 Apr 2026 15:59:03 -0500 Subject: [PATCH] Surface invalid-leaf recovery_needed in blockerAttention read model A blocked parent like PAP-2089 currently reports a generic needs_attention chain even when its leaf is the productive-run-stopped shape from PAP-2674's plan. Extend listIssueBlockerAttentionMap to fetch each candidate leaf's latest issue-linked succeeded run plus its productivity-review state, classify the leaf as recovery_needed when it matches the shape, and propagate the leaf identifier, next-action owner, and CTA hint up to the root. The leaf itself also surfaces recovery_needed on its own attention so the leaf-level notice can render when an inbox wake lands directly on it. Adds two integration cases (productive_run_stopped and continuation_exhausted) to the embedded-Postgres regression suite. Co-Authored-By: Paperclip --- .../__tests__/issue-blocker-attention.test.ts | 77 +++++ server/src/services/issues.ts | 320 +++++++++++++++++- 2 files changed, 379 insertions(+), 18 deletions(-) diff --git a/server/src/__tests__/issue-blocker-attention.test.ts b/server/src/__tests__/issue-blocker-attention.test.ts index 611921911ad..134c3f5f432 100644 --- a/server/src/__tests__/issue-blocker-attention.test.ts +++ b/server/src/__tests__/issue-blocker-attention.test.ts @@ -356,6 +356,83 @@ describeEmbeddedPostgres("issue blocker attention", () => { }); }); + async function recordedRun(input: { + companyId: string; + agentId: string; + issueId: string; + livenessState?: string | null; + continuationAttempt?: number; + }) { + const runId = randomUUID(); + await db.insert(heartbeatRuns).values({ + id: runId, + companyId: input.companyId, + agentId: input.agentId, + status: "succeeded", + contextSnapshot: { issueId: input.issueId }, + livenessState: (input.livenessState ?? "advanced") as never, + continuationAttempt: input.continuationAttempt ?? 0, + }); + return runId; + } + + it("classifies a chain whose leaf had a productive run that exited without a continuation as recovery_needed/productive_run_stopped", async () => { + const { companyId, agentId } = await createCompany("PRR"); + const parentId = await insertIssue({ companyId, identifier: "PRR-1", title: "Parent", status: "blocked" }); + const leafId = await insertIssue({ + companyId, + identifier: "PRR-2", + title: "Recovery leaf", + status: "in_progress", + assigneeAgentId: agentId, + }); + await block({ companyId, blockerIssueId: leafId, blockedIssueId: parentId }); + await recordedRun({ companyId, agentId, issueId: leafId, livenessState: "advanced" }); + + const issuesList = await svc.list(companyId, { status: "blocked,in_progress" }); + const parent = issuesList.find((issue) => issue.id === parentId); + const leaf = issuesList.find((issue) => issue.id === leafId); + + expect(parent?.blockerAttention).toMatchObject({ + state: "recovery_needed", + reason: "productive_run_stopped", + sampleBlockerIdentifier: "PRR-2", + nextActionHint: "wake_to_continue", + }); + expect(parent?.blockerAttention?.nextActionOwner).toMatchObject({ type: "agent", agentId }); + + // Leaf surface gets the same recovery_needed classification on its own + // blockerAttention so the leaf-perspective notice can render. + expect(leaf?.blockerAttention).toMatchObject({ + state: "recovery_needed", + reason: "productive_run_stopped", + nextActionHint: "wake_to_continue", + }); + expect(leaf?.blockerAttention?.nextActionOwner).toMatchObject({ type: "agent", agentId }); + }); + + it("escalates to continuation_exhausted when the latest run already used its bounded continuations", async () => { + const { companyId, agentId } = await createCompany("PRX"); + const parentId = await insertIssue({ companyId, identifier: "PRX-1", title: "Parent", status: "blocked" }); + const leafId = await insertIssue({ + companyId, + identifier: "PRX-2", + title: "Exhausted leaf", + status: "in_progress", + assigneeAgentId: agentId, + }); + await block({ companyId, blockerIssueId: leafId, blockedIssueId: parentId }); + await recordedRun({ companyId, agentId, issueId: leafId, livenessState: "advanced", continuationAttempt: 2 }); + + const parent = (await svc.list(companyId, { status: "blocked" })).find((issue) => issue.id === parentId); + expect(parent?.blockerAttention).toMatchObject({ + state: "recovery_needed", + reason: "continuation_exhausted", + sampleBlockerIdentifier: "PRX-2", + nextActionHint: "create_recovery_issue", + }); + }); + it("does not treat a scheduled retry as actively covered work", async () => { const { companyId, agentId } = await createCompany("PBY"); const parentId = await insertIssue({ companyId, identifier: "PBY-1", title: "Parent", status: "blocked" }); diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index 741d8ce302c..1420cf4068d 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -29,6 +29,10 @@ import { } from "@paperclipai/db"; import type { IssueBlockerAttention, + IssueBlockerAttentionReason, + IssueBlockerAttentionState, + IssueBlockerNextActionHint, + IssueBlockerNextActionOwner, IssueProductivityReview, IssueProductivityReviewTrigger, IssueRelationIssueSummary, @@ -687,6 +691,18 @@ const BLOCKER_ATTENTION_OPEN_RECOVERY_TERMINAL_STATUSES = ["done", "cancelled"]; const BLOCKER_ATTENTION_MAX_DEPTH = 8; const BLOCKER_ATTENTION_MAX_NODES = 2000; const BLOCKER_ATTENTION_INVOKABLE_AGENT_STATUSES = new Set(["active", "idle", "running", "error"]); +// Leaves whose latest issue-linked run looks "productive but stranded" — the +// run produced concrete progress or described future work, but the issue is +// non-terminal and no continuation/wake/interaction took ownership of the next +// step. These leaves drive the recovery_needed surface. +const BLOCKER_ATTENTION_RECOVERY_LIVENESS_STATES = new Set([ + "advanced", + "needs_followup", + "plan_only", + "empty_response", +]); +const BLOCKER_ATTENTION_RECOVERY_LEAF_STATUSES = new Set(["todo", "in_progress"]); +const BLOCKER_ATTENTION_DEFAULT_MAX_CONTINUATION_ATTEMPTS = 2; type IssueBlockerAttentionNode = { id: string; @@ -770,9 +786,96 @@ function createIssueBlockerAttention(input: Partial = {}) attentionBlockerCount: input.attentionBlockerCount ?? 0, sampleBlockerIdentifier: input.sampleBlockerIdentifier ?? null, sampleStalledBlockerIdentifier: input.sampleStalledBlockerIdentifier ?? null, + nextActionOwner: input.nextActionOwner ?? null, + nextActionHint: input.nextActionHint ?? null, }; } +type LeafRecoveryClassification = { + reason: Extract< + IssueBlockerAttentionReason, + "productive_run_stopped" | "continuation_exhausted" | "continuation_suppressed" + >; + hint: IssueBlockerNextActionHint; + owner: IssueBlockerNextActionOwner; +}; + +function leafRecoveryHintForReason( + reason: LeafRecoveryClassification["reason"], + owner: IssueBlockerNextActionOwner, +): IssueBlockerNextActionHint { + if (owner.type === "user") return "reassign"; + if (owner.type === "none") return "reassign"; + if (reason === "productive_run_stopped") return "wake_to_continue"; + if (reason === "continuation_exhausted") return "create_recovery_issue"; + return "needs_human_review"; +} + +function ownerForLeafRecovery(node: IssueBlockerAttentionNode): IssueBlockerNextActionOwner { + if (node.assigneeAgentId) { + return { type: "agent", agentId: node.assigneeAgentId, userId: null }; + } + if (node.assigneeUserId) { + return { type: "user", agentId: null, userId: node.assigneeUserId }; + } + return { type: "none", agentId: null, userId: null }; +} + +type IssueBlockerAttentionLatestRunRow = { + issueId: string | null; + status: string; + livenessState: string | null; + continuationAttempt: number | null; + finishedAt: Date | null; +}; + +type IssueBlockerAttentionLatestRun = { + status: string; + livenessState: string | null; + continuationAttempt: number | null; + finishedAt: Date | null; +}; + +function classifyLeafRecovery(input: { + node: IssueBlockerAttentionNode; + latestRun: IssueBlockerAttentionLatestRun | null; + hasActivePath: boolean; + hasWaitingPath: boolean; + hasProductivityReview: boolean; + agentInvokable: boolean; + maxContinuationAttempts: number; +}): LeafRecoveryClassification | null { + const { node, latestRun, hasActivePath, hasWaitingPath, hasProductivityReview, agentInvokable, maxContinuationAttempts } = input; + if (!BLOCKER_ATTENTION_RECOVERY_LEAF_STATUSES.has(node.status)) return null; + if (hasActivePath || hasWaitingPath) return null; + if (!latestRun) return null; + if (latestRun.status !== "succeeded") return null; + if (!latestRun.livenessState || !BLOCKER_ATTENTION_RECOVERY_LIVENESS_STATES.has(latestRun.livenessState)) return null; + + const owner = ownerForLeafRecovery(node); + // If the leaf has no agent owner at all, fall through to needs_attention so + // the existing "no owner" surface continues to apply. + if (owner.type === "none") return null; + + let reason: LeafRecoveryClassification["reason"]; + if (hasProductivityReview) { + reason = "continuation_suppressed"; + } else if ((latestRun.continuationAttempt ?? 0) >= maxContinuationAttempts) { + reason = "continuation_exhausted"; + } else { + reason = "productive_run_stopped"; + } + + // An invokable agent who can be woken keeps the wake_to_continue affordance; + // otherwise the user needs to reassign or escalate. + let hint = leafRecoveryHintForReason(reason, owner); + if (owner.type === "agent" && !agentInvokable) { + hint = reason === "continuation_exhausted" ? "create_recovery_issue" : "needs_human_review"; + } + + return { reason, hint, owner }; +} + function blockerSampleIdentifier(node: IssueBlockerAttentionNode | null | undefined) { return node?.identifier ?? node?.id ?? null; } @@ -1006,18 +1109,27 @@ async function listIssueBlockerAttentionMap( companyId: string, issueRows: IssueBlockerAttentionInputNode[], ): Promise> { - const roots = issueRows.filter((row) => row.companyId === companyId && row.status === "blocked"); + const sameCompanyRows = issueRows.filter((row) => row.companyId === companyId); + const roots = sameCompanyRows.filter((row) => row.status === "blocked"); + // Non-blocked, non-terminal input issues with an agent owner are eligible to + // surface as invalid leaves on their own detail/inbox surfaces. + const recoverySeedRows = sameCompanyRows.filter((row) => + BLOCKER_ATTENTION_RECOVERY_LEAF_STATUSES.has(row.status) && Boolean(row.assigneeAgentId), + ); const attentionMap = new Map(); for (const row of issueRows) { if (row.status !== "blocked") { attentionMap.set(row.id, createIssueBlockerAttention()); } } - if (roots.length === 0) return attentionMap; + if (roots.length === 0 && recoverySeedRows.length === 0) return attentionMap; const nodesById = new Map(); const edgesByIssueId = new Map(); for (const root of roots) nodesById.set(root.id, { ...root }); + for (const seed of recoverySeedRows) { + if (!nodesById.has(seed.id)) nodesById.set(seed.id, { ...seed }); + } let frontier = roots.map((root) => root.id); let truncated = false; @@ -1221,40 +1333,150 @@ async function listIssueBlockerAttentionMap( : []; const agentsById = new Map(agentRows.map((agent) => [agent.id, agent])); + // Recovery-leaf inputs: latest issue-linked run + productivity-review hold. + // We only need this for non-blocked agent-owned non-terminal nodes; gate the + // queries on that subset so we do not pull runs for every blocker chain. + const recoveryCandidateIds = [...nodesById.values()] + .filter((node) => + BLOCKER_ATTENTION_RECOVERY_LEAF_STATUSES.has(node.status) && + Boolean(node.assigneeAgentId) && + !activeIssueIds.has(node.id) && + !explicitWaitingIssueIds.has(node.id), + ) + .map((node) => node.id); + + const latestRunByIssueId = new Map(); + if (recoveryCandidateIds.length > 0) { + for (const chunk of chunkList(recoveryCandidateIds, ISSUE_LIST_RELATED_QUERY_CHUNK_SIZE)) { + const runRows: IssueBlockerAttentionLatestRunRow[] = await dbOrTx + .select({ + issueId: sql`${heartbeatRuns.contextSnapshot} ->> 'issueId'`, + status: heartbeatRuns.status, + livenessState: heartbeatRuns.livenessState, + continuationAttempt: heartbeatRuns.continuationAttempt, + finishedAt: heartbeatRuns.finishedAt, + }) + .from(heartbeatRuns) + .where( + and( + eq(heartbeatRuns.companyId, companyId), + inArray(sql`${heartbeatRuns.contextSnapshot} ->> 'issueId'`, chunk), + ), + ) + .orderBy(desc(heartbeatRuns.createdAt), desc(heartbeatRuns.id)); + for (const row of runRows) { + if (!row.issueId) continue; + if (latestRunByIssueId.has(row.issueId)) continue; + latestRunByIssueId.set(row.issueId, { + status: row.status, + livenessState: row.livenessState, + continuationAttempt: row.continuationAttempt, + finishedAt: row.finishedAt, + }); + } + } + } + + const productivityReviewIssueIds = new Set(); + if (recoveryCandidateIds.length > 0) { + for (const chunk of chunkList(recoveryCandidateIds, ISSUE_LIST_RELATED_QUERY_CHUNK_SIZE)) { + const productivityRows: Array<{ originId: string | null }> = await dbOrTx + .select({ originId: issues.originId }) + .from(issues) + .where( + and( + eq(issues.companyId, companyId), + eq(issues.originKind, PRODUCTIVITY_REVIEW_ORIGIN_KIND), + isNull(issues.hiddenAt), + inArray(issues.originId, chunk), + notInArray(issues.status, PRODUCTIVITY_REVIEW_TERMINAL_STATUSES), + ), + ); + for (const row of productivityRows) { + if (row.originId) productivityReviewIssueIds.add(row.originId); + } + } + } + + const leafRecoveryByNodeId = new Map(); + for (const nodeId of recoveryCandidateIds) { + const node = nodesById.get(nodeId); + if (!node) continue; + const assignee = node.assigneeAgentId ? agentsById.get(node.assigneeAgentId) : null; + const agentInvokable = Boolean( + assignee && assignee.companyId === companyId && BLOCKER_ATTENTION_INVOKABLE_AGENT_STATUSES.has(assignee.status), + ); + const classification = classifyLeafRecovery({ + node, + latestRun: latestRunByIssueId.get(nodeId) ?? null, + hasActivePath: activeIssueIds.has(nodeId), + hasWaitingPath: explicitWaitingIssueIds.has(nodeId), + hasProductivityReview: productivityReviewIssueIds.has(nodeId), + agentInvokable, + maxContinuationAttempts: BLOCKER_ATTENTION_DEFAULT_MAX_CONTINUATION_ATTEMPTS, + }); + if (classification) leafRecoveryByNodeId.set(nodeId, classification); + } + type PathClassification = { covered: boolean; stalled: boolean; sampleBlockerIdentifier: string | null; sampleStalledBlockerIdentifier: string | null; + recoveryLeafIdentifier: string | null; + recovery: LeafRecoveryClassification | null; }; + const emptyPath = (sample: string | null): PathClassification => ({ + covered: false, + stalled: false, + sampleBlockerIdentifier: sample, + sampleStalledBlockerIdentifier: null, + recoveryLeafIdentifier: null, + recovery: null, + }); + const coveredPath = (sample: string | null): PathClassification => ({ + covered: true, + stalled: false, + sampleBlockerIdentifier: sample, + sampleStalledBlockerIdentifier: null, + recoveryLeafIdentifier: null, + recovery: null, + }); const classifyPath = ( nodeId: string, seen: Set, ): PathClassification => { const sample = blockerSampleIdentifier(nodesById.get(nodeId)); if (truncated || seen.has(nodeId)) { - return { covered: false, stalled: false, sampleBlockerIdentifier: sample, sampleStalledBlockerIdentifier: null }; + return emptyPath(sample); } const node = nodesById.get(nodeId); if (!node || node.companyId !== companyId) { - return { covered: false, stalled: false, sampleBlockerIdentifier: nodeId, sampleStalledBlockerIdentifier: null }; + return emptyPath(nodeId); } const nodeSample = blockerSampleIdentifier(node); if (node.status === "done") { - return { covered: true, stalled: false, sampleBlockerIdentifier: nodeSample, sampleStalledBlockerIdentifier: null }; + return coveredPath(nodeSample); } if (node.status === "in_review") { const hasWaitingPath = activeIssueIds.has(node.id) || Boolean(node.assigneeUserId) || explicitWaitingIssueIds.has(node.id); if (hasWaitingPath) { - return { covered: true, stalled: false, sampleBlockerIdentifier: nodeSample, sampleStalledBlockerIdentifier: null }; + return coveredPath(nodeSample); } - return { covered: false, stalled: true, sampleBlockerIdentifier: nodeSample, sampleStalledBlockerIdentifier: nodeSample }; + return { + covered: false, + stalled: true, + sampleBlockerIdentifier: nodeSample, + sampleStalledBlockerIdentifier: nodeSample, + recoveryLeafIdentifier: null, + recovery: null, + }; } if (activeIssueIds.has(node.id)) { - return { covered: true, stalled: false, sampleBlockerIdentifier: nodeSample, sampleStalledBlockerIdentifier: null }; + return coveredPath(nodeSample); } if (node.status === "cancelled") { - return { covered: false, stalled: false, sampleBlockerIdentifier: nodeSample, sampleStalledBlockerIdentifier: null }; + return emptyPath(nodeSample); } const downstream = (edgesByIssueId.get(node.id) ?? []).filter((edge) => nodesById.get(edge.blockerIssueId)?.status !== "done"); @@ -1264,13 +1486,28 @@ async function listIssueBlockerAttentionMap( const classified = downstream.map((edge) => classifyPath(edge.blockerIssueId, nextSeen)); const stalledChild = classified.find((result) => result.stalled || result.sampleStalledBlockerIdentifier); const sampleStalled = stalledChild?.sampleStalledBlockerIdentifier ?? null; - const hardAttention = classified.find((result) => !result.covered && !result.stalled); + const recoveryChild = classified.find((result) => result.recovery && result.recoveryLeafIdentifier); + const recoveryLeafIdentifier = recoveryChild?.recoveryLeafIdentifier ?? null; + const recovery = recoveryChild?.recovery ?? null; + const hardAttention = classified.find((result) => !result.covered && !result.stalled && !result.recovery); if (hardAttention) { return { covered: false, stalled: false, sampleBlockerIdentifier: hardAttention.sampleBlockerIdentifier, sampleStalledBlockerIdentifier: sampleStalled, + recoveryLeafIdentifier, + recovery, + }; + } + if (recoveryChild) { + return { + covered: false, + stalled: false, + sampleBlockerIdentifier: recoveryLeafIdentifier ?? recoveryChild.sampleBlockerIdentifier, + sampleStalledBlockerIdentifier: sampleStalled, + recoveryLeafIdentifier, + recovery, }; } const stalledEntry = classified.find((result) => result.stalled); @@ -1280,6 +1517,8 @@ async function listIssueBlockerAttentionMap( stalled: true, sampleBlockerIdentifier: stalledEntry.sampleBlockerIdentifier, sampleStalledBlockerIdentifier: sampleStalled, + recoveryLeafIdentifier: null, + recovery: null, }; } return { @@ -1287,17 +1526,31 @@ async function listIssueBlockerAttentionMap( stalled: false, sampleBlockerIdentifier: classified[0]?.sampleBlockerIdentifier ?? nodeSample, sampleStalledBlockerIdentifier: null, + recoveryLeafIdentifier: null, + recovery: null, + }; + } + + const recovery = leafRecoveryByNodeId.get(node.id) ?? null; + if (recovery) { + return { + covered: false, + stalled: false, + sampleBlockerIdentifier: nodeSample, + sampleStalledBlockerIdentifier: null, + recoveryLeafIdentifier: nodeSample, + recovery, }; } if (node.assigneeAgentId) { const assignee = agentsById.get(node.assigneeAgentId); if (!assignee || assignee.companyId !== companyId || !BLOCKER_ATTENTION_INVOKABLE_AGENT_STATUSES.has(assignee.status)) { - return { covered: false, stalled: false, sampleBlockerIdentifier: nodeSample, sampleStalledBlockerIdentifier: null }; + return emptyPath(nodeSample); } } - return { covered: false, stalled: false, sampleBlockerIdentifier: nodeSample, sampleStalledBlockerIdentifier: null }; + return emptyPath(nodeSample); }; for (const root of roots) { @@ -1316,20 +1569,34 @@ async function listIssueBlockerAttentionMap( })); const coveredBlockerCount = classified.filter((entry) => entry.result.covered).length; const stalledBlockerCount = classified.filter((entry) => entry.result.stalled).length; - const attentionBlockerCount = classified.length - coveredBlockerCount - stalledBlockerCount; - const hardAttentionEntry = classified.find((entry) => !entry.result.covered && !entry.result.stalled); + const recoveryEntry = classified.find((entry) => entry.result.recovery && entry.result.recoveryLeafIdentifier); + const recoveryBlockerCount = classified.filter((entry) => entry.result.recovery).length; + const attentionBlockerCount = classified.length - coveredBlockerCount - stalledBlockerCount - recoveryBlockerCount; + const hardAttentionEntry = classified.find((entry) => !entry.result.covered && !entry.result.stalled && !entry.result.recovery); const stalledEntry = classified.find((entry) => entry.result.stalled); - const sampleEntry = hardAttentionEntry ?? stalledEntry ?? classified[0] ?? null; + const sampleEntry = hardAttentionEntry ?? recoveryEntry ?? stalledEntry ?? classified[0] ?? null; const sampleNode = sampleEntry ? nodesById.get(sampleEntry.edge.blockerIssueId) : null; const sampleStalledFromChain = classified .map((entry) => entry.result.sampleStalledBlockerIdentifier) .find((value) => value); + const recoveryLeafIdentifier = recoveryEntry?.result.recoveryLeafIdentifier ?? null; + const recoveryClassification = recoveryEntry?.result.recovery ?? null; - let state: IssueBlockerAttention["state"]; - let reason: IssueBlockerAttention["reason"]; + let state: IssueBlockerAttentionState; + let reason: IssueBlockerAttentionReason; + let nextActionOwner: IssueBlockerNextActionOwner | null = null; + let nextActionHint: IssueBlockerNextActionHint = null; + let sampleBlockerIdentifier = + sampleEntry?.result.sampleBlockerIdentifier ?? blockerSampleIdentifier(sampleNode); if (attentionBlockerCount > 0) { state = "needs_attention"; reason = "attention_required"; + } else if (recoveryClassification) { + state = "recovery_needed"; + reason = recoveryClassification.reason; + nextActionOwner = recoveryClassification.owner; + nextActionHint = recoveryClassification.hint; + sampleBlockerIdentifier = recoveryLeafIdentifier ?? sampleBlockerIdentifier; } else if (stalledBlockerCount > 0) { state = "stalled"; reason = "stalled_review"; @@ -1347,9 +1614,26 @@ async function listIssueBlockerAttentionMap( coveredBlockerCount, stalledBlockerCount, attentionBlockerCount, - sampleBlockerIdentifier: sampleEntry?.result.sampleBlockerIdentifier ?? blockerSampleIdentifier(sampleNode), + sampleBlockerIdentifier, sampleStalledBlockerIdentifier: stalledEntry?.result.sampleStalledBlockerIdentifier ?? sampleStalledFromChain ?? null, + nextActionOwner, + nextActionHint, + })); + } + + // Non-blocked recovery seeds: a leaf that is itself the invalid endpoint + // gets its own recovery_needed surface (e.g. opening the leaf directly from + // an inbox wake). + for (const seed of recoverySeedRows) { + const recovery = leafRecoveryByNodeId.get(seed.id); + if (!recovery) continue; + attentionMap.set(seed.id, createIssueBlockerAttention({ + state: "recovery_needed", + reason: recovery.reason, + sampleBlockerIdentifier: blockerSampleIdentifier(nodesById.get(seed.id)) ?? seed.identifier ?? seed.id, + nextActionOwner: recovery.owner, + nextActionHint: recovery.hint, })); }