mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-23 00:53:38 +02:00
Handle explicit waits in liveness read models
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
b72ab70a4f
commit
1e68cb9be0
5 files changed
+314
-22
No files matched your search
@@ -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",
|
||||
|
||||
@@ -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" });
|
||||
|
||||
@@ -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,
|
||||
|
||||
+120
-21
@@ -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<string>();
|
||||
if (reviewNodeIds.length > 0) {
|
||||
for (const chunk of chunkList(reviewNodeIds, ISSUE_LIST_RELATED_QUERY_CHUNK_SIZE)) {
|
||||
const explicitWaitingByIssueId = new Map<string, IssueBlockerExplicitWaitingPath>();
|
||||
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({
|
||||
|
||||
@@ -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,
|
||||
|
||||
Reference in new issue
Block a user