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 <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-04-28 15:59:03 -05:00
1 parent 7d475f85bf
commit 3ea2ff4187
2 files changed
+379 -18

No files matched your search

@@ -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" });
+302 -18
View File
@@ -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<IssueBlockerAttention> = {})
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<Map<string, IssueBlockerAttention>> {
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<string, IssueBlockerAttention>();
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<string, IssueBlockerAttentionNode>();
const edgesByIssueId = new Map<string, IssueBlockerAttentionEdge[]>();
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<string, IssueBlockerAttentionLatestRun>();
if (recoveryCandidateIds.length > 0) {
for (const chunk of chunkList(recoveryCandidateIds, ISSUE_LIST_RELATED_QUERY_CHUNK_SIZE)) {
const runRows: IssueBlockerAttentionLatestRunRow[] = await dbOrTx
.select({
issueId: sql<string | null>`${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<string>`${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<string>();
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<string, LeafRecoveryClassification>();
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<string>,
): 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,
}));
}