mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-23 09:07:46 +02:00
Fix stale covered-blocked run detection
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
fbbf2e9c41
commit
616d6700dc
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",
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -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);
|
||||
|
||||
Reference in new issue
Block a user