mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-02 08:28:48 +01:00
fix(server): isolate blocker traversal budgets per root
Share queried graph facts across roots while retaining independent node/depth limits, expanded reachability and truncation for classification. Unrelated limited graphs must not change terminal subjects or delay archive deadlines. Prove depth/node overflow and overlapping-root liveness red/green, preserve the100-root SQL bound, and qualify existing owner assertions on isolated PostgreSQL. Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
66b71955cf
commit
efbe3fa67d
4 files changed
+101
-26
No files matched your search
@@ -15,7 +15,7 @@ import { decisionRetentionService, DEFAULT_DECISION_ARCHIVE_DAYS } from "../serv
|
||||
import { appendHeartbeatRunEvent } from "../services/heartbeat-run-events.js";
|
||||
import { issueThreadInteractionService } from "../services/issue-thread-interactions.js";
|
||||
import { createAgentLifecycle } from "../services/agent-lifecycle.js";
|
||||
import { issueService } from "../services/issues.js";
|
||||
import { BLOCKER_ATTENTION_MAX_DEPTH, BLOCKER_ATTENTION_MAX_NODES, issueService } from "../services/issues.js";
|
||||
import { notifyAttentionRetentionWork } from "../services/retention-work-notifications.js";
|
||||
import { idleWorkSnapshot } from "../services/task-admission.js";
|
||||
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
|
||||
@@ -256,6 +256,61 @@ const delay = DEFAULT_DECISION_ARCHIVE_DAYS * 86_400_000;
|
||||
expect(refreshed.rows!.some(entry => entry.sourceId === terminals[99].id)).toBe(false);
|
||||
});
|
||||
|
||||
it.each(["depth", "nodes"] as const)("preserves shallow terminal selection when an unrelated graph hits its %s limit", async limit => {
|
||||
const root = await insertIssue("blocked", { updatedAt: new Date() });
|
||||
const middle = await insertIssue("todo", { updatedAt: new Date() });
|
||||
const terminal = await insertIssue("backlog", { assigneeAgentId: agentId });
|
||||
const heavy = await insertIssue("blocked", { updatedAt: new Date() });
|
||||
const count = limit === "depth" ? BLOCKER_ATTENTION_MAX_DEPTH + 1 : BLOCKER_ATTENTION_MAX_NODES;
|
||||
const heavyNodes = await db.insert(issues).values(Array.from({ length: count }, () => ({ companyId,
|
||||
title: randomUUID(), status: "todo", updatedAt: new Date() }))).returning();
|
||||
await db.insert(issueRelations).values([
|
||||
{ companyId, issueId: middle.id, relatedIssueId: root.id, type: "blocks" },
|
||||
{ companyId, issueId: terminal.id, relatedIssueId: middle.id, type: "blocks" },
|
||||
...heavyNodes.map((node, index) => ({ companyId, issueId: node.id,
|
||||
relatedIssueId: limit === "nodes" || index === 0 ? heavy.id : heavyNodes[index - 1].id, type: "blocks" })),
|
||||
]);
|
||||
const owner = issueService(db);
|
||||
const expected = (await owner.listBlockerAttention(companyId, [root])).get(root.id)!;
|
||||
const limited = (await owner.listBlockerAttention(companyId, [heavy])).get(heavy.id)!;
|
||||
expect(expected.terminalBlockerIssueId).toBe(terminal.id);
|
||||
expect(limited.terminalBlockerIssueId).toBe(limited.directBlockerIssueId);
|
||||
for (const input of [[root, heavy], [heavy, root]]) {
|
||||
const batch = await owner.listBlockerAttention(companyId, input);
|
||||
expect(batch.get(root.id)).toEqual(expected); expect(batch.get(heavy.id)).toEqual(limited);
|
||||
}
|
||||
const page = await attentionRetentionService(db).load({ companyId, sourceKind: "blocker_attention" });
|
||||
const entry = page.rows!.find(item => item.rootId === root.id && item.variant === "terminal")!;
|
||||
expect(entry.sourceId).toBe(terminal.id); expect(entry.dueAt).toBeLessThan(Date.now());
|
||||
await attentionRetentionService(db).check(entry);
|
||||
expect((await state("blocker_attention", terminal.id))?.archivedAt).toBeInstanceOf(Date);
|
||||
expect(await db.select().from(decisionArchiveNotificationOutbox).where(eq(decisionArchiveNotificationOutbox.sourceId, terminal.id))).toHaveLength(1);
|
||||
});
|
||||
|
||||
it("keeps root-specific reachability and cycle budgets when another root explores farther", async () => {
|
||||
const root = await insertIssue("blocked");
|
||||
const chain = await db.insert(issues).values(Array.from({ length: BLOCKER_ATTENTION_MAX_DEPTH + 1 }, (_, index) => ({
|
||||
companyId, title: randomUUID(), status: index === BLOCKER_ATTENTION_MAX_DEPTH ? "in_progress"
|
||||
: index === BLOCKER_ATTENTION_MAX_DEPTH - 2 ? "blocked" : "todo", updatedAt: old() }))).returning();
|
||||
const overlapping = chain[BLOCKER_ATTENTION_MAX_DEPTH - 2];
|
||||
const cycle = await insertIssue("blocked");
|
||||
const leaf = await insertIssue("backlog", { assigneeAgentId: agentId });
|
||||
await db.insert(issueRelations).values([
|
||||
...chain.map((node, index) => ({ companyId, issueId: node.id, relatedIssueId: index ? chain[index - 1].id : root.id, type: "blocks" })),
|
||||
{ companyId, issueId: leaf.id, relatedIssueId: cycle.id, type: "blocks" },
|
||||
{ companyId, issueId: cycle.id, relatedIssueId: leaf.id, type: "blocks" },
|
||||
]);
|
||||
const owner = issueService(db), roots = [root, overlapping, cycle];
|
||||
const singles = new Map();
|
||||
for (const item of roots) singles.set(item.id, (await owner.listBlockerAttention(companyId, [item])).get(item.id));
|
||||
expect(singles.get(root.id).blockingTreeLive).toBe(false);
|
||||
expect(singles.get(overlapping.id).blockingTreeLive).toBe(true);
|
||||
for (const input of [roots, [...roots].reverse()]) {
|
||||
const batch = await owner.listBlockerAttention(companyId, input);
|
||||
for (const item of roots) expect(batch.get(item.id)).toEqual(singles.get(item.id));
|
||||
}
|
||||
});
|
||||
|
||||
it("scopes ordinary execution-control writes to their company", async () => {
|
||||
const otherCompany = randomUUID();
|
||||
await db.insert(companies).values({ id: otherCompany, name: "Other", issuePrefix: "OTHER" });
|
||||
|
||||
@@ -8,6 +8,7 @@ import {
|
||||
approvals,
|
||||
companies,
|
||||
createDb,
|
||||
applyPendingMigrations,
|
||||
heartbeatRuns,
|
||||
issueApprovals,
|
||||
issueRelations,
|
||||
@@ -21,7 +22,10 @@ import {
|
||||
import { issueService } from "../services/issues.js";
|
||||
import { buildIssueGraphLivenessIncidentKey } from "../services/recovery/origins.js";
|
||||
|
||||
const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport();
|
||||
import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
|
||||
|
||||
const externalUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_BLOCKER_ATTENTION_TEST_DATABASE_URL");
|
||||
const embeddedPostgresSupport = externalUrl ? { supported: true } : await getEmbeddedPostgresTestSupport();
|
||||
const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip;
|
||||
|
||||
if (!embeddedPostgresSupport.supported) {
|
||||
@@ -36,8 +40,9 @@ describeEmbeddedPostgres("issue blocker attention", () => {
|
||||
let tempDb: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | null = null;
|
||||
|
||||
beforeAll(async () => {
|
||||
tempDb = await startEmbeddedPostgresTestDatabase("paperclip-issue-blocker-attention-");
|
||||
db = createDb(tempDb.connectionString);
|
||||
if (externalUrl) await applyPendingMigrations(externalUrl);
|
||||
else tempDb = await startEmbeddedPostgresTestDatabase("paperclip-issue-blocker-attention-");
|
||||
db = createDb(externalUrl ?? tempDb!.connectionString, { maxConnections: 6 });
|
||||
svc = issueService(db);
|
||||
}, 20_000);
|
||||
|
||||
@@ -55,6 +60,7 @@ describeEmbeddedPostgres("issue blocker attention", () => {
|
||||
});
|
||||
|
||||
afterAll(async () => {
|
||||
await db?.$client.end({ timeout: 1 });
|
||||
await tempDb?.cleanup();
|
||||
});
|
||||
|
||||
|
||||
@@ -3840,18 +3840,19 @@ async function listIssueBlockerAttentionMap(
|
||||
const edgesByIssueId = new Map<string, IssueBlockerAttentionEdge[]>();
|
||||
for (const root of roots) nodesById.set(root.id, { ...root });
|
||||
|
||||
let frontier = roots.map((root) => root.id);
|
||||
let truncated = false;
|
||||
type Traversal = { nodes: Set<string>; expanded: Set<string>; frontier: string[]; truncated: boolean };
|
||||
const traversals = new Map<string, Traversal>(roots.map(root => [root.id, {
|
||||
nodes: new Set([root.id]), expanded: new Set<string>(), frontier: [root.id], truncated: false,
|
||||
}]));
|
||||
const expanded = new Set<string>();
|
||||
const pendingFinalizeBlockerIssueIds = new Set<string>();
|
||||
for (
|
||||
let depth = 0;
|
||||
frontier.length > 0 && depth < BLOCKER_ATTENTION_MAX_DEPTH;
|
||||
depth += 1
|
||||
) {
|
||||
const nextFrontier = new Set<string>();
|
||||
for (let depth = 0; depth < BLOCKER_ATTENTION_MAX_DEPTH; depth += 1) {
|
||||
const active = [...traversals.values()].filter(traversal => !traversal.truncated && traversal.frontier.length > 0);
|
||||
if (active.length === 0) break;
|
||||
const frontier = [...new Set(active.flatMap(traversal => traversal.frontier))].filter(id => !expanded.has(id));
|
||||
|
||||
for (const chunk of chunkList(
|
||||
[...new Set(frontier)],
|
||||
frontier,
|
||||
ISSUE_LIST_RELATED_QUERY_CHUNK_SIZE,
|
||||
)) {
|
||||
const readinessByIssueId = await listIssueDependencyReadinessMap(
|
||||
@@ -3959,17 +3960,27 @@ async function listIssueBlockerAttentionMap(
|
||||
assigneeAgentId: row.assigneeAgentId,
|
||||
assigneeUserId: row.assigneeUserId,
|
||||
});
|
||||
nextFrontier.add(row.blockerIssueId);
|
||||
}
|
||||
for (const id of chunk) expanded.add(id);
|
||||
}
|
||||
|
||||
if (nodesById.size > BLOCKER_ATTENTION_MAX_NODES) {
|
||||
truncated = true;
|
||||
break;
|
||||
for (const traversal of active) {
|
||||
const nextFrontier = new Set<string>();
|
||||
for (const id of traversal.frontier) {
|
||||
traversal.expanded.add(id);
|
||||
for (const edge of edgesByIssueId.get(id) ?? []) {
|
||||
if (traversal.nodes.has(edge.blockerIssueId) || !nodesById.has(edge.blockerIssueId)) continue;
|
||||
traversal.nodes.add(edge.blockerIssueId);
|
||||
nextFrontier.add(edge.blockerIssueId);
|
||||
}
|
||||
}
|
||||
traversal.truncated = traversal.nodes.size > BLOCKER_ATTENTION_MAX_NODES;
|
||||
traversal.frontier = traversal.truncated ? [] : [...nextFrontier];
|
||||
}
|
||||
frontier = [...nextFrontier];
|
||||
}
|
||||
if (frontier.length > 0) truncated = true;
|
||||
for (const traversal of traversals.values()) {
|
||||
if (traversal.frontier.length > 0) traversal.truncated = true;
|
||||
}
|
||||
|
||||
const nodeIds = [...nodesById.keys()];
|
||||
const activeIssueIds = new Set<string>();
|
||||
@@ -4215,6 +4226,7 @@ async function listIssueBlockerAttentionMap(
|
||||
const classifyPath = (
|
||||
nodeId: string,
|
||||
seen: Set<string>,
|
||||
truncated: boolean,
|
||||
): PathClassification => {
|
||||
const sample = blockerSampleIdentifier(nodesById.get(nodeId));
|
||||
if (truncated || seen.has(nodeId)) {
|
||||
@@ -4319,7 +4331,7 @@ async function listIssueBlockerAttentionMap(
|
||||
const nextSeen = new Set(seen);
|
||||
nextSeen.add(nodeId);
|
||||
const classified = downstream.map((edge) =>
|
||||
classifyPath(edge.blockerIssueId, nextSeen),
|
||||
classifyPath(edge.blockerIssueId, nextSeen, truncated),
|
||||
);
|
||||
const stalledChild = classified.find(
|
||||
(result) => result.stalled || result.sampleStalledBlockerIdentifier,
|
||||
@@ -4385,8 +4397,8 @@ async function listIssueBlockerAttentionMap(
|
||||
};
|
||||
};
|
||||
|
||||
const pathHasLiveWork = (nodeId: string, seen: Set<string>): boolean => {
|
||||
if (seen.has(nodeId)) return false;
|
||||
const pathHasLiveWork = (nodeId: string, seen: Set<string>, traversal: Traversal): boolean => {
|
||||
if (seen.has(nodeId) || !traversal.nodes.has(nodeId)) return false;
|
||||
const node = nodesById.get(nodeId);
|
||||
if (!node || node.companyId !== companyId) return false;
|
||||
if (
|
||||
@@ -4397,6 +4409,7 @@ async function listIssueBlockerAttentionMap(
|
||||
)
|
||||
return true;
|
||||
|
||||
if (!traversal.expanded.has(nodeId)) return false;
|
||||
const nextSeen = new Set(seen);
|
||||
nextSeen.add(nodeId);
|
||||
return (edgesByIssueId.get(node.id) ?? []).some((edge) => {
|
||||
@@ -4406,7 +4419,7 @@ async function listIssueBlockerAttentionMap(
|
||||
!pendingFinalizeBlockerIssueIds.has(edge.blockerIssueId)
|
||||
)
|
||||
return false;
|
||||
return pathHasLiveWork(edge.blockerIssueId, nextSeen);
|
||||
return pathHasLiveWork(edge.blockerIssueId, nextSeen, traversal);
|
||||
});
|
||||
};
|
||||
|
||||
@@ -4419,6 +4432,7 @@ async function listIssueBlockerAttentionMap(
|
||||
};
|
||||
|
||||
for (const root of roots) {
|
||||
const traversal = traversals.get(root.id)!;
|
||||
const topLevelEdges = (edgesByIssueId.get(root.id) ?? []).filter((edge) => {
|
||||
const blocker = nodesById.get(edge.blockerIssueId);
|
||||
return (
|
||||
@@ -4440,7 +4454,7 @@ async function listIssueBlockerAttentionMap(
|
||||
|
||||
const classified = topLevelEdges.map((edge) => ({
|
||||
edge,
|
||||
result: classifyPath(edge.blockerIssueId, new Set([root.id])),
|
||||
result: classifyPath(edge.blockerIssueId, new Set([root.id]), traversal.truncated),
|
||||
}));
|
||||
const coveredBlockerCount = classified.filter(
|
||||
(entry) => entry.result.covered,
|
||||
@@ -4523,7 +4537,7 @@ async function listIssueBlockerAttentionMap(
|
||||
sampleStalledFromChain ??
|
||||
null,
|
||||
blockingTreeLive: topLevelEdges.some((edge) =>
|
||||
pathHasLiveWork(edge.blockerIssueId, new Set([root.id])),
|
||||
pathHasLiveWork(edge.blockerIssueId, new Set([root.id]), traversal),
|
||||
),
|
||||
directBlockerIssueId: sampleEntry?.edge.blockerIssueId ?? null,
|
||||
terminalBlockerIssueId,
|
||||
|
||||
Reference in new issue
Block a user