Fix stale covered-blocked run detection

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-04-24 13:32:07 -05:00
1 parent 3e4dd2a988
commit d4c70cdd18
2 files changed
+91 -17

No files matched your search

@@ -1,4 +1,5 @@
import { randomUUID } from "node:crypto";
import { eq } from "drizzle-orm";
import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest";
import {
agents,
@@ -99,13 +100,19 @@ describeEmbeddedPostgres("issue blocker attention", () => {
});
}
async function activeRun(input: { companyId: string; agentId: string; issueId: string; status?: string }) {
async function activeRun(input: { companyId: string; agentId: string; issueId: string; status?: string; current?: boolean }) {
const runId = randomUUID();
await db.insert(heartbeatRuns).values({
id: runId,
companyId: input.companyId,
agentId: input.agentId,
status: input.status ?? "running",
contextSnapshot: { issueId: input.issueId },
});
if (input.current !== false) {
await db.update(issues).set({ executionRunId: runId }).where(eq(issues.id, input.issueId));
}
return runId;
}
it("classifies a blocked parent as covered when its child has a running execution path", async () => {
@@ -122,7 +129,7 @@ describeEmbeddedPostgres("issue blocker attention", () => {
await block({ companyId, blockerIssueId: childId, blockedIssueId: parentId });
await activeRun({ companyId, agentId, issueId: childId });
const [parent] = await svc.list(companyId, { status: "blocked" });
const parent = (await svc.list(companyId, { status: "blocked" })).find((issue) => issue.id === parentId);
expect(parent?.blockerAttention).toMatchObject({
state: "covered",
@@ -156,7 +163,7 @@ describeEmbeddedPostgres("issue blocker attention", () => {
await block({ companyId, blockerIssueId: idleBlockerId, blockedIssueId: parentId });
await activeRun({ companyId, agentId, issueId: activeChildId });
const [parent] = await svc.list(companyId, { status: "blocked" });
const parent = (await svc.list(companyId, { status: "blocked" })).find((issue) => issue.id === parentId);
expect(parent?.blockerAttention).toMatchObject({
state: "needs_attention",
@@ -183,7 +190,7 @@ describeEmbeddedPostgres("issue blocker attention", () => {
await block({ companyId, blockerIssueId: leafId, blockedIssueId: blockerId });
await activeRun({ companyId, agentId, issueId: leafId });
const [parent] = await svc.list(companyId, { status: "blocked" });
const parent = (await svc.list(companyId, { status: "blocked" })).find((issue) => issue.id === parentId);
expect(parent?.blockerAttention).toMatchObject({
state: "covered",
@@ -209,7 +216,7 @@ describeEmbeddedPostgres("issue blocker attention", () => {
await block({ companyId, blockerIssueId: blockerId, blockedIssueId: parentId });
await activeRun({ companyId: other.companyId, agentId: other.agentId, issueId: blockerId });
const [parent] = await svc.list(companyId, { status: "blocked" });
const parent = (await svc.list(companyId, { status: "blocked" })).find((issue) => issue.id === parentId);
expect(parent?.blockerAttention).toMatchObject({
state: "needs_attention",
@@ -220,4 +227,54 @@ describeEmbeddedPostgres("issue blocker attention", () => {
sampleBlockerIdentifier: "PBS-2",
});
});
it("does not cover a blocker from a stale run the issue no longer owns", async () => {
const { companyId, agentId } = await createCompany("PBX");
const parentId = await insertIssue({ companyId, identifier: "PBX-1", title: "Parent", status: "blocked" });
const blockerId = await insertIssue({
companyId,
identifier: "PBX-2",
title: "Previously running blocker",
status: "blocked",
assigneeAgentId: agentId,
});
await block({ companyId, blockerIssueId: blockerId, blockedIssueId: parentId });
await activeRun({ companyId, agentId, issueId: blockerId, current: false });
const parent = (await svc.list(companyId, { status: "blocked" })).find((issue) => issue.id === parentId);
expect(parent?.blockerAttention).toMatchObject({
state: "needs_attention",
reason: "attention_required",
unresolvedBlockerCount: 1,
coveredBlockerCount: 0,
attentionBlockerCount: 1,
sampleBlockerIdentifier: "PBX-2",
});
});
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" });
const blockerId = await insertIssue({
companyId,
identifier: "PBY-2",
title: "Retrying blocker",
status: "blocked",
assigneeAgentId: agentId,
});
await block({ companyId, blockerIssueId: blockerId, blockedIssueId: parentId });
await activeRun({ companyId, agentId, issueId: blockerId, status: "scheduled_retry" });
const parent = (await svc.list(companyId, { status: "blocked" })).find((issue) => issue.id === parentId);
expect(parent?.blockerAttention).toMatchObject({
state: "needs_attention",
reason: "attention_required",
unresolvedBlockerCount: 1,
coveredBlockerCount: 0,
attentionBlockerCount: 1,
sampleBlockerIdentifier: "PBY-2",
});
});
});
+29 -12
View File
@@ -657,7 +657,7 @@ async function withIssueLabels(dbOrTx: any, rows: IssueRow[]): Promise<IssueWith
}
const ACTIVE_RUN_STATUSES = ["queued", "running"];
const BLOCKER_ATTENTION_ACTIVE_RUN_STATUSES = ["queued", "running", "scheduled_retry"];
const BLOCKER_ATTENTION_ACTIVE_RUN_STATUSES = ["queued", "running"];
const BLOCKER_ATTENTION_ACTIVE_WAKE_STATUSES = ["queued", "deferred_issue_execution"];
const BLOCKER_ATTENTION_MAX_DEPTH = 8;
const BLOCKER_ATTENTION_MAX_NODES = 2000;
@@ -670,9 +670,16 @@ type IssueBlockerAttentionNode = {
identifier: string | null;
title: string;
status: string;
executionRunId?: string | null;
assigneeAgentId: string | null;
assigneeUserId: string | null;
};
type IssueBlockerAttentionInputNode =
Pick<
IssueBlockerAttentionNode,
"id" | "companyId" | "parentId" | "identifier" | "title" | "status" | "assigneeAgentId" | "assigneeUserId"
>
& { executionRunId?: string | null };
type IssueBlockerAttentionEdge = {
issueId: string;
@@ -759,7 +766,7 @@ function appendBlockerAttentionEdges(
async function listIssueBlockerAttentionMap(
dbOrTx: any,
companyId: string,
issueRows: Array<Pick<IssueBlockerAttentionNode, "id" | "companyId" | "parentId" | "identifier" | "title" | "status" | "assigneeAgentId" | "assigneeUserId">>,
issueRows: IssueBlockerAttentionInputNode[],
): Promise<Map<string, IssueBlockerAttention>> {
const roots = issueRows.filter((row) => row.companyId === companyId && row.status === "blocked");
const attentionMap = new Map<string, IssueBlockerAttention>();
@@ -790,6 +797,7 @@ async function listIssueBlockerAttentionMap(
identifier: issues.identifier,
title: issues.title,
status: issues.status,
executionRunId: issues.executionRunId,
assigneeAgentId: issues.assigneeAgentId,
assigneeUserId: issues.assigneeUserId,
})
@@ -814,6 +822,7 @@ async function listIssueBlockerAttentionMap(
identifier: issues.identifier,
title: issues.title,
status: issues.status,
executionRunId: issues.executionRunId,
assigneeAgentId: issues.assigneeAgentId,
assigneeUserId: issues.assigneeUserId,
})
@@ -848,6 +857,7 @@ async function listIssueBlockerAttentionMap(
identifier: row.identifier,
title: row.title,
status: row.status,
executionRunId: row.executionRunId,
assigneeAgentId: row.assigneeAgentId,
assigneeUserId: row.assigneeUserId,
});
@@ -866,23 +876,33 @@ async function listIssueBlockerAttentionMap(
const nodeIds = [...nodesById.keys()];
const activeIssueIds = new Set<string>();
const agentIds = new Set<string>();
const issueIdByExecutionRunId = new Map<string, string>();
for (const node of nodesById.values()) {
if (node.assigneeAgentId) agentIds.add(node.assigneeAgentId);
if (node.executionRunId) issueIdByExecutionRunId.set(node.executionRunId, node.id);
}
for (const chunk of chunkList(nodeIds, ISSUE_LIST_RELATED_QUERY_CHUNK_SIZE)) {
const runRowsPromise: Promise<IssueBlockerAttentionActivePathRow[]> = dbOrTx
for (const chunk of chunkList([...issueIdByExecutionRunId.keys()], ISSUE_LIST_RELATED_QUERY_CHUNK_SIZE)) {
const runRows: Array<{ id: string }> = await dbOrTx
.select({
issueId: sql<string | null>`${heartbeatRuns.contextSnapshot} ->> 'issueId'`,
id: heartbeatRuns.id,
})
.from(heartbeatRuns)
.where(
and(
eq(heartbeatRuns.companyId, companyId),
inArray(heartbeatRuns.status, BLOCKER_ATTENTION_ACTIVE_RUN_STATUSES),
inArray(sql<string>`${heartbeatRuns.contextSnapshot} ->> 'issueId'`, chunk),
inArray(heartbeatRuns.id, chunk),
),
);
for (const row of runRows) {
const issueId = issueIdByExecutionRunId.get(row.id);
if (issueId) activeIssueIds.add(issueId);
}
}
for (const chunk of chunkList(nodeIds, ISSUE_LIST_RELATED_QUERY_CHUNK_SIZE)) {
const wakeRowsPromise: Promise<IssueBlockerAttentionActivePathRow[]> = dbOrTx
.select({
issueId: sql<string | null>`${agentWakeupRequests.payload} ->> 'issueId'`,
@@ -896,11 +916,8 @@ async function listIssueBlockerAttentionMap(
inArray(sql<string>`${agentWakeupRequests.payload} ->> 'issueId'`, chunk),
),
);
const [runRows, wakeRows] = await Promise.all([
runRowsPromise,
wakeRowsPromise,
]);
for (const row of [...runRows, ...wakeRows]) {
const wakeRows = await wakeRowsPromise;
for (const row of wakeRows) {
if (row.issueId) activeIssueIds.add(row.issueId);
}
}
@@ -1985,7 +2002,7 @@ export function issueService(db: Db) {
listBlockerAttention: async (
companyId: string,
issueRows: Array<Pick<IssueBlockerAttentionNode, "id" | "companyId" | "parentId" | "identifier" | "title" | "status" | "assigneeAgentId" | "assigneeUserId">>,
issueRows: IssueBlockerAttentionInputNode[],
dbOrTx: any = db,
) => {
return listIssueBlockerAttentionMap(dbOrTx, companyId, issueRows);