From 1e68cb9be0f83f01d0ce5dd86f927a83d964ae5c Mon Sep 17 00:00:00 2001 From: Dotta Date: Tue, 28 Apr 2026 18:06:29 -0500 Subject: [PATCH] Handle explicit waits in liveness read models Co-Authored-By: Paperclip --- .../heartbeat-process-recovery.test.ts | 46 ++++++ .../__tests__/issue-blocker-attention.test.ts | 94 ++++++++++++ server/src/services/heartbeat.ts | 46 ++++++ server/src/services/issues.ts | 141 +++++++++++++++--- server/src/services/recovery/service.ts | 9 +- 5 files changed, 314 insertions(+), 22 deletions(-) diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index 173a13e8e3b..c5c2b9e676e 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -19,6 +19,7 @@ import { issueComments, issueDocuments, issueRelations, + issueThreadInteractions, issueTreeHoldMembers, issueTreeHolds, issues, @@ -310,6 +311,7 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { await db.delete(companySkills); await db.delete(costEvents); await db.delete(issueComments); + await db.delete(issueThreadInteractions); await db.delete(issueDocuments); await db.delete(documentRevisions); await db.delete(documents); @@ -318,6 +320,7 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { await db.delete(issueTreeHolds); for (let attempt = 0; attempt < 5; attempt += 1) { await db.delete(issueComments); + await db.delete(issueThreadInteractions); await db.delete(issueDocuments); try { await db.delete(issues); @@ -2385,6 +2388,49 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { expect(refreshedIssue?.assigneeAgentId).toBe(agentId); }); + it("suppresses bounded liveness continuation wakes when the issue has a pending confirmation", async () => { + const { companyId, agentId, issueId, runId } = await seedStrandedIssueFixture({ + status: "in_progress", + runStatus: "succeeded", + livenessState: "plan_only", + livenessReason: "Run produced a plan and needs approval", + nextAction: "Wait for the board to confirm the plan.", + }); + await db.insert(issueThreadInteractions).values({ + companyId, + issueId, + kind: "request_confirmation", + status: "pending", + payload: {}, + createdByAgentId: agentId, + sourceRunId: runId, + }); + const heartbeat = heartbeatService(db); + + const sourceRun = await db + .select() + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, runId)) + .then((rows) => rows[0] ?? null); + if (!sourceRun) throw new Error("Expected source run"); + + await heartbeat.handleRunLivenessContinuationForTest(sourceRun); + + const wakeups = await db + .select() + .from(agentWakeupRequests) + .where(and(eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.reason, "run_liveness_continuation"))); + expect(wakeups).toHaveLength(0); + + const refreshedRun = await db + .select() + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, runId)) + .then((rows) => rows[0] ?? null); + expect(refreshedRun?.continuationAttempt).toBe(0); + expect(refreshedRun?.livenessReason).toContain("continuation suppressed by explicit interaction or approval wait"); + }); + it("queues a bounded liveness continuation when no pause hold suppresses it", async () => { const { companyId, runId } = await seedStrandedIssueFixture({ status: "in_progress", diff --git a/server/src/__tests__/issue-blocker-attention.test.ts b/server/src/__tests__/issue-blocker-attention.test.ts index 134c3f5f432..6a2d3fba30d 100644 --- a/server/src/__tests__/issue-blocker-attention.test.ts +++ b/server/src/__tests__/issue-blocker-attention.test.ts @@ -4,10 +4,13 @@ import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; import { agents, agentWakeupRequests, + approvals, companies, createDb, heartbeatRuns, + issueApprovals, issueRelations, + issueThreadInteractions, issues, } from "@paperclipai/db"; import { @@ -39,6 +42,9 @@ describeEmbeddedPostgres("issue blocker attention", () => { afterEach(async () => { await db.delete(heartbeatRuns); await db.delete(agentWakeupRequests); + await db.delete(issueThreadInteractions); + await db.delete(issueApprovals); + await db.delete(approvals); await db.delete(issueRelations); await db.delete(issues); await db.delete(agents); @@ -115,6 +121,34 @@ describeEmbeddedPostgres("issue blocker attention", () => { return runId; } + async function pendingInteraction(input: { companyId: string; issueId: string; kind?: "request_confirmation" | "ask_user_questions" }) { + await db.insert(issueThreadInteractions).values({ + companyId: input.companyId, + issueId: input.issueId, + kind: input.kind ?? "request_confirmation", + status: "pending", + payload: {}, + }); + } + + async function linkedPendingApproval(input: { companyId: string; issueId: string; agentId: string }) { + const approvalId = randomUUID(); + await db.insert(approvals).values({ + id: approvalId, + companyId: input.companyId, + type: "request_board_approval", + requestedByAgentId: input.agentId, + status: "pending", + payload: {}, + }); + await db.insert(issueApprovals).values({ + companyId: input.companyId, + issueId: input.issueId, + approvalId, + linkedByAgentId: input.agentId, + }); + } + it("classifies a blocked parent as covered when its child has a running execution path", async () => { const { companyId, agentId } = await createCompany("PBC"); const parentId = await insertIssue({ companyId, identifier: "PBC-1", title: "Parent", status: "blocked" }); @@ -411,6 +445,66 @@ describeEmbeddedPostgres("issue blocker attention", () => { expect(leaf?.blockerAttention?.nextActionOwner).toMatchObject({ type: "agent", agentId }); }); + it("treats a pending plan confirmation on an in-progress leaf as explicit waiting instead of recovery", async () => { + const { companyId, agentId } = await createCompany("PRW"); + const parentId = await insertIssue({ companyId, identifier: "PRW-1", title: "Parent", status: "blocked" }); + const leafId = await insertIssue({ + companyId, + identifier: "PRW-2", + title: "Plan confirmation leaf", + status: "in_progress", + assigneeAgentId: agentId, + }); + await block({ companyId, blockerIssueId: leafId, blockedIssueId: parentId }); + await recordedRun({ companyId, agentId, issueId: leafId, livenessState: "advanced" }); + await pendingInteraction({ companyId, issueId: leafId, kind: "request_confirmation" }); + + 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: "covered", + reason: "explicit_waiting", + coveredBlockerCount: 1, + sampleBlockerIdentifier: "PRW-2", + nextActionHint: "needs_human_review", + }); + expect(parent?.blockerAttention?.nextActionOwner).toMatchObject({ type: "user" }); + expect(leaf?.blockerAttention).toMatchObject({ + state: "covered", + reason: "explicit_waiting", + sampleBlockerIdentifier: "PRW-2", + nextActionHint: "needs_human_review", + }); + expect(leaf?.blockerAttention?.nextActionOwner).toMatchObject({ type: "user" }); + }); + + it("treats linked pending approvals on blocked leaves as explicit waiting", async () => { + const { companyId, agentId } = await createCompany("PRA"); + const parentId = await insertIssue({ companyId, identifier: "PRA-1", title: "Parent", status: "blocked" }); + const leafId = await insertIssue({ + companyId, + identifier: "PRA-2", + title: "Approval leaf", + status: "blocked", + assigneeAgentId: agentId, + }); + await block({ companyId, blockerIssueId: leafId, blockedIssueId: parentId }); + await linkedPendingApproval({ companyId, issueId: leafId, agentId }); + + const parent = (await svc.list(companyId, { status: "blocked" })).find((issue) => issue.id === parentId); + + expect(parent?.blockerAttention).toMatchObject({ + state: "covered", + reason: "explicit_waiting", + coveredBlockerCount: 1, + sampleBlockerIdentifier: "PRA-2", + nextActionHint: "needs_human_review", + }); + expect(parent?.blockerAttention?.nextActionOwner).toMatchObject({ type: "user" }); + }); + 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" }); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index f04bd090caa..3faacafc388 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -26,8 +26,11 @@ import { issueDocuments, heartbeatRunEvents, heartbeatRuns, + approvals, + issueApprovals, issueComments, issueRelations, + issueThreadInteractions, issues, issueWorkProducts, projects, @@ -128,6 +131,8 @@ import { environmentRuntimeService } from "./environment-runtime.js"; import { environmentRunOrchestrator } from "./environment-run-orchestrator.js"; import type { PluginWorkerManager } from "./plugin-worker-manager.js"; +const EXPLICIT_WAITING_INTERACTION_KINDS = ["request_confirmation", "ask_user_questions"]; + const MAX_LIVE_LOG_CHUNK_BYTES = 8 * 1024; const MAX_PERSISTED_LOG_CHUNK_CHARS = 64 * 1024; const MAX_RUN_EVENT_PAYLOAD_STRING_CHARS = 16 * 1024; @@ -2768,6 +2773,39 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) }); } + async function hasExplicitIssueWait(companyId: string, issueId: string) { + const [interaction, approval] = await Promise.all([ + db + .select({ id: issueThreadInteractions.id }) + .from(issueThreadInteractions) + .where( + and( + eq(issueThreadInteractions.companyId, companyId), + eq(issueThreadInteractions.issueId, issueId), + eq(issueThreadInteractions.status, "pending"), + inArray(issueThreadInteractions.kind, EXPLICIT_WAITING_INTERACTION_KINDS), + ), + ) + .limit(1) + .then((rows) => rows[0] ?? null), + db + .select({ id: issueApprovals.issueId }) + .from(issueApprovals) + .innerJoin(approvals, eq(issueApprovals.approvalId, approvals.id)) + .where( + and( + eq(issueApprovals.companyId, companyId), + eq(issueApprovals.issueId, issueId), + inArray(approvals.status, ["pending", "revision_requested"]), + ), + ) + .limit(1) + .then((rows) => rows[0] ?? null), + ]); + + return Boolean(interaction || approval); + } + async function handleRunLivenessContinuation(run: typeof heartbeatRuns.$inferSelect) { const livenessState = run.livenessState as RunLivenessState | null; if (!livenessState) return; @@ -2812,6 +2850,14 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) }) : null; if (issue) { + if (await hasExplicitIssueWait(issue.companyId, issue.id)) { + await setRunStatus(run.id, run.status, { + livenessReason: + `${run.livenessReason ?? "Run ended without concrete progress"}; continuation suppressed by explicit interaction or approval wait`, + }); + return; + } + const productivityHold = await productivityReviews.isProductivityReviewContinuationHoldActive({ companyId: issue.companyId, issueId: issue.id, diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index 1420cf4068d..e367393c395 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -674,6 +674,7 @@ const ACTIVE_RUN_STATUSES = ["queued", "running"]; const BLOCKER_ATTENTION_ACTIVE_RUN_STATUSES = ["queued", "running"]; const BLOCKER_ATTENTION_ACTIVE_WAKE_STATUSES = ["queued", "deferred_issue_execution"]; const BLOCKER_ATTENTION_PENDING_INTERACTION_STATUSES = ["pending"]; +const BLOCKER_ATTENTION_WAITING_INTERACTION_KINDS = ["request_confirmation", "ask_user_questions"]; const BLOCKER_ATTENTION_PENDING_APPROVAL_STATUSES = ["pending", "revision_requested"]; const BLOCKER_ATTENTION_OPEN_RECOVERY_ORIGIN_KIND = "harness_liveness_escalation"; const PRODUCTIVITY_REVIEW_ORIGIN_KIND = "issue_productivity_review"; @@ -738,6 +739,10 @@ type IssueBlockerAttentionAgentRow = { companyId: string; status: string; }; +type IssueBlockerExplicitWaitingPath = { + owner: IssueBlockerNextActionOwner; + hint: IssueBlockerNextActionHint; +}; async function activeRunMapForIssues( dbOrTx: any, @@ -821,6 +826,23 @@ function ownerForLeafRecovery(node: IssueBlockerAttentionNode): IssueBlockerNext return { type: "none", agentId: null, userId: null }; } +function ownerForExplicitWait(input: { assigneeAgentId?: string | null; assigneeUserId?: string | null }): IssueBlockerNextActionOwner { + if (input.assigneeUserId) { + return { type: "user", agentId: null, userId: input.assigneeUserId }; + } + if (input.assigneeAgentId) { + return { type: "agent", agentId: input.assigneeAgentId, userId: null }; + } + return { type: "user", agentId: null, userId: null }; +} + +function explicitWaitingPathForOwner(owner: IssueBlockerNextActionOwner): IssueBlockerExplicitWaitingPath { + return { + owner, + hint: "needs_human_review", + }; +} + type IssueBlockerAttentionLatestRunRow = { issueId: string | null; status: string; @@ -1272,12 +1294,22 @@ async function listIssueBlockerAttentionMap( } } - const reviewNodeIds = [...nodesById.values()] - .filter((node) => node.status === "in_review") + const nonTerminalNodeIds = [...nodesById.values()] + .filter((node) => node.status !== "done" && node.status !== "cancelled") .map((node) => node.id); - const explicitWaitingIssueIds = new Set(); - if (reviewNodeIds.length > 0) { - for (const chunk of chunkList(reviewNodeIds, ISSUE_LIST_RELATED_QUERY_CHUNK_SIZE)) { + const explicitWaitingByIssueId = new Map(); + const setExplicitWaitingPath = (issueId: string | null | undefined, path: IssueBlockerExplicitWaitingPath) => { + if (!issueId || explicitWaitingByIssueId.has(issueId)) return; + explicitWaitingByIssueId.set(issueId, path); + }; + for (const node of nodesById.values()) { + if (node.status === "done" || node.status === "cancelled") continue; + if (node.assigneeUserId) { + setExplicitWaitingPath(node.id, explicitWaitingPathForOwner(ownerForExplicitWait(node))); + } + } + if (nonTerminalNodeIds.length > 0) { + for (const chunk of chunkList(nonTerminalNodeIds, ISSUE_LIST_RELATED_QUERY_CHUNK_SIZE)) { const interactionRows: Array<{ issueId: string }> = await dbOrTx .select({ issueId: issueThreadInteractions.issueId }) .from(issueThreadInteractions) @@ -1285,10 +1317,16 @@ async function listIssueBlockerAttentionMap( and( eq(issueThreadInteractions.companyId, companyId), inArray(issueThreadInteractions.status, BLOCKER_ATTENTION_PENDING_INTERACTION_STATUSES), + inArray(issueThreadInteractions.kind, BLOCKER_ATTENTION_WAITING_INTERACTION_KINDS), inArray(issueThreadInteractions.issueId, chunk), ), ); - for (const row of interactionRows) explicitWaitingIssueIds.add(row.issueId); + for (const row of interactionRows) { + const node = nodesById.get(row.issueId); + setExplicitWaitingPath(row.issueId, explicitWaitingPathForOwner(ownerForExplicitWait({ + assigneeUserId: node?.assigneeUserId ?? null, + }))); + } const approvalRows: Array<{ issueId: string }> = await dbOrTx .select({ issueId: issueApprovals.issueId }) @@ -1301,10 +1339,23 @@ async function listIssueBlockerAttentionMap( inArray(issueApprovals.issueId, chunk), ), ); - for (const row of approvalRows) explicitWaitingIssueIds.add(row.issueId); + for (const row of approvalRows) { + const node = nodesById.get(row.issueId); + setExplicitWaitingPath(row.issueId, explicitWaitingPathForOwner(ownerForExplicitWait({ + assigneeUserId: node?.assigneeUserId ?? null, + }))); + } - const recoveryRows: Array<{ originId: string | null }> = await dbOrTx - .select({ originId: issues.originId }) + const recoveryRows: Array<{ + originId: string | null; + assigneeAgentId: string | null; + assigneeUserId: string | null; + }> = await dbOrTx + .select({ + originId: issues.originId, + assigneeAgentId: issues.assigneeAgentId, + assigneeUserId: issues.assigneeUserId, + }) .from(issues) .where( and( @@ -1316,7 +1367,7 @@ async function listIssueBlockerAttentionMap( ), ); for (const row of recoveryRows) { - if (row.originId) explicitWaitingIssueIds.add(row.originId); + setExplicitWaitingPath(row.originId, explicitWaitingPathForOwner(ownerForExplicitWait(row))); } } } @@ -1341,7 +1392,7 @@ async function listIssueBlockerAttentionMap( BLOCKER_ATTENTION_RECOVERY_LEAF_STATUSES.has(node.status) && Boolean(node.assigneeAgentId) && !activeIssueIds.has(node.id) && - !explicitWaitingIssueIds.has(node.id), + !explicitWaitingByIssueId.has(node.id), ) .map((node) => node.id); @@ -1410,7 +1461,7 @@ async function listIssueBlockerAttentionMap( node, latestRun: latestRunByIssueId.get(nodeId) ?? null, hasActivePath: activeIssueIds.has(nodeId), - hasWaitingPath: explicitWaitingIssueIds.has(nodeId), + hasWaitingPath: explicitWaitingByIssueId.has(nodeId), hasProductivityReview: productivityReviewIssueIds.has(nodeId), agentInvokable, maxContinuationAttempts: BLOCKER_ATTENTION_DEFAULT_MAX_CONTINUATION_ATTEMPTS, @@ -1420,27 +1471,37 @@ async function listIssueBlockerAttentionMap( type PathClassification = { covered: boolean; + coveredReason: IssueBlockerAttentionReason | null; stalled: boolean; sampleBlockerIdentifier: string | null; sampleStalledBlockerIdentifier: string | null; recoveryLeafIdentifier: string | null; recovery: LeafRecoveryClassification | null; + waitingPath: IssueBlockerExplicitWaitingPath | null; }; const emptyPath = (sample: string | null): PathClassification => ({ covered: false, + coveredReason: null, stalled: false, sampleBlockerIdentifier: sample, sampleStalledBlockerIdentifier: null, recoveryLeafIdentifier: null, recovery: null, + waitingPath: null, }); - const coveredPath = (sample: string | null): PathClassification => ({ + const coveredPath = ( + sample: string | null, + coveredReason: IssueBlockerAttentionReason = null, + waitingPath: IssueBlockerExplicitWaitingPath | null = null, + ): PathClassification => ({ covered: true, + coveredReason, stalled: false, sampleBlockerIdentifier: sample, sampleStalledBlockerIdentifier: null, recoveryLeafIdentifier: null, recovery: null, + waitingPath, }); const classifyPath = ( nodeId: string, @@ -1458,18 +1519,23 @@ async function listIssueBlockerAttentionMap( if (node.status === "done") { return coveredPath(nodeSample); } + const explicitWaitingPath = explicitWaitingByIssueId.get(node.id) ?? null; + if (explicitWaitingPath) { + return coveredPath(nodeSample, "explicit_waiting", explicitWaitingPath); + } if (node.status === "in_review") { - const hasWaitingPath = activeIssueIds.has(node.id) || Boolean(node.assigneeUserId) || explicitWaitingIssueIds.has(node.id); - if (hasWaitingPath) { + if (activeIssueIds.has(node.id)) { return coveredPath(nodeSample); } return { covered: false, + coveredReason: null, stalled: true, sampleBlockerIdentifier: nodeSample, sampleStalledBlockerIdentifier: nodeSample, recoveryLeafIdentifier: null, recovery: null, + waitingPath: null, }; } if (activeIssueIds.has(node.id)) { @@ -1489,45 +1555,55 @@ async function listIssueBlockerAttentionMap( const recoveryChild = classified.find((result) => result.recovery && result.recoveryLeafIdentifier); const recoveryLeafIdentifier = recoveryChild?.recoveryLeafIdentifier ?? null; const recovery = recoveryChild?.recovery ?? null; + const waitingEntry = classified.find((result) => result.coveredReason === "explicit_waiting" && result.waitingPath); + const waitingPath = waitingEntry?.waitingPath ?? null; const hardAttention = classified.find((result) => !result.covered && !result.stalled && !result.recovery); if (hardAttention) { return { covered: false, + coveredReason: null, stalled: false, sampleBlockerIdentifier: hardAttention.sampleBlockerIdentifier, sampleStalledBlockerIdentifier: sampleStalled, recoveryLeafIdentifier, recovery, + waitingPath, }; } if (recoveryChild) { return { covered: false, + coveredReason: null, stalled: false, sampleBlockerIdentifier: recoveryLeafIdentifier ?? recoveryChild.sampleBlockerIdentifier, sampleStalledBlockerIdentifier: sampleStalled, recoveryLeafIdentifier, recovery, + waitingPath, }; } const stalledEntry = classified.find((result) => result.stalled); if (stalledEntry) { return { covered: false, + coveredReason: null, stalled: true, sampleBlockerIdentifier: stalledEntry.sampleBlockerIdentifier, sampleStalledBlockerIdentifier: sampleStalled, recoveryLeafIdentifier: null, recovery: null, + waitingPath, }; } return { covered: true, + coveredReason: waitingPath ? "explicit_waiting" : null, stalled: false, - sampleBlockerIdentifier: classified[0]?.sampleBlockerIdentifier ?? nodeSample, + sampleBlockerIdentifier: waitingEntry?.sampleBlockerIdentifier ?? classified[0]?.sampleBlockerIdentifier ?? nodeSample, sampleStalledBlockerIdentifier: null, recoveryLeafIdentifier: null, recovery: null, + waitingPath, }; } @@ -1535,11 +1611,13 @@ async function listIssueBlockerAttentionMap( if (recovery) { return { covered: false, + coveredReason: null, stalled: false, sampleBlockerIdentifier: nodeSample, sampleStalledBlockerIdentifier: null, recoveryLeafIdentifier: nodeSample, recovery, + waitingPath: null, }; } @@ -1556,9 +1634,13 @@ async function listIssueBlockerAttentionMap( for (const root of roots) { const topLevelEdges = (edgesByIssueId.get(root.id) ?? []).filter((edge) => nodesById.get(edge.blockerIssueId)?.status !== "done"); if (topLevelEdges.length === 0) { + const waitingPath = explicitWaitingByIssueId.get(root.id) ?? null; attentionMap.set(root.id, createIssueBlockerAttention({ - state: "needs_attention", - reason: "attention_required", + state: waitingPath ? "covered" : "needs_attention", + reason: waitingPath ? "explicit_waiting" : "attention_required", + sampleBlockerIdentifier: waitingPath ? blockerSampleIdentifier(nodesById.get(root.id)) : null, + nextActionOwner: waitingPath?.owner ?? null, + nextActionHint: waitingPath?.hint ?? null, })); continue; } @@ -1574,6 +1656,7 @@ async function listIssueBlockerAttentionMap( 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 waitingEntry = classified.find((entry) => entry.result.coveredReason === "explicit_waiting" && entry.result.waitingPath); const sampleEntry = hardAttentionEntry ?? recoveryEntry ?? stalledEntry ?? classified[0] ?? null; const sampleNode = sampleEntry ? nodesById.get(sampleEntry.edge.blockerIssueId) : null; const sampleStalledFromChain = classified @@ -1602,9 +1685,14 @@ async function listIssueBlockerAttentionMap( reason = "stalled_review"; } else { state = "covered"; - reason = topLevelEdges.every((edge) => nodesById.get(edge.blockerIssueId)?.parentId === root.id) - ? "active_child" - : "active_dependency"; + reason = waitingEntry + ? "explicit_waiting" + : topLevelEdges.every((edge) => nodesById.get(edge.blockerIssueId)?.parentId === root.id) + ? "active_child" + : "active_dependency"; + nextActionOwner = waitingEntry?.result.waitingPath?.owner ?? null; + nextActionHint = waitingEntry?.result.waitingPath?.hint ?? null; + sampleBlockerIdentifier = waitingEntry?.result.sampleBlockerIdentifier ?? sampleBlockerIdentifier; } attentionMap.set(root.id, createIssueBlockerAttention({ @@ -1626,6 +1714,17 @@ async function listIssueBlockerAttentionMap( // gets its own recovery_needed surface (e.g. opening the leaf directly from // an inbox wake). for (const seed of recoverySeedRows) { + const waitingPath = explicitWaitingByIssueId.get(seed.id) ?? null; + if (waitingPath) { + attentionMap.set(seed.id, createIssueBlockerAttention({ + state: "covered", + reason: "explicit_waiting", + sampleBlockerIdentifier: blockerSampleIdentifier(nodesById.get(seed.id)) ?? seed.identifier ?? seed.id, + nextActionOwner: waitingPath.owner, + nextActionHint: waitingPath.hint, + })); + continue; + } const recovery = leafRecoveryByNodeId.get(seed.id); if (!recovery) continue; attentionMap.set(seed.id, createIssueBlockerAttention({ diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index 38cf03eff12..c1051346479 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -61,6 +61,7 @@ const ACTIVE_RUN_OUTPUT_EVIDENCE_TAIL_BYTES = 8 * 1024; const STRANDED_ISSUE_RECOVERY_ORIGIN_KIND = RECOVERY_ORIGIN_KINDS.strandedIssueRecovery; const STALE_ACTIVE_RUN_EVALUATION_ORIGIN_KIND = RECOVERY_ORIGIN_KINDS.staleActiveRunEvaluation; const DEFERRED_WAKE_CONTEXT_KEY = "_paperclipWakeContext"; +const EXPLICIT_WAITING_INTERACTION_KINDS = ["request_confirmation", "ask_user_questions"]; type RecoveryWakeupOptions = { source?: "timer" | "assignment" | "on_demand" | "automation"; @@ -407,6 +408,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) eq(issueThreadInteractions.companyId, issue.companyId), eq(issueThreadInteractions.issueId, issue.id), eq(issueThreadInteractions.status, "pending"), + inArray(issueThreadInteractions.kind, EXPLICIT_WAITING_INTERACTION_KINDS), ), ) .limit(1) @@ -2075,7 +2077,12 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) status: issueThreadInteractions.status, }) .from(issueThreadInteractions) - .where(eq(issueThreadInteractions.status, "pending")), + .where( + and( + eq(issueThreadInteractions.status, "pending"), + inArray(issueThreadInteractions.kind, EXPLICIT_WAITING_INTERACTION_KINDS), + ), + ), db .select({ companyId: issueApprovals.companyId,