fix(attention): address Greptile concurrency findings

Serialize partial item verdict submissions with a row lock so concurrent\nresolutions merge against the latest stored result. Batch failed-run supersession\nlookups to avoid issuing a query for every failed run.\n\nCo-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
Dotta committed 2026-07-10 20:31:14 +00:00
1 parent 8db9fa6a14
commit f53d9bb0b7
3 files changed
+165 -87

No files matched your search

@@ -608,6 +608,57 @@ describeEmbeddedPostgres("attention service", () => {
});
});
it("suppresses failed-run attention after a newer run for the same issue", async () => {
const { companyId, workerId } = await seedCompany("ATN");
const issueId = await insertIssue({
companyId,
identifier: "ATN-1",
title: "Recoverable task",
status: "in_progress",
});
const failedRunId = randomUUID();
const failedAt = new Date("2026-07-09T12:00:00.000Z");
await db.insert(heartbeatRuns).values([
{
id: failedRunId,
companyId,
agentId: workerId,
invocationSource: "automation",
status: "failed",
error: "adapter failed",
contextSnapshot: { issueId },
createdAt: failedAt,
updatedAt: failedAt,
finishedAt: failedAt,
},
{
id: randomUUID(),
companyId,
agentId: workerId,
invocationSource: "automation",
status: "succeeded",
contextSnapshot: { issueId },
createdAt: new Date("2026-07-09T12:01:00.000Z"),
updatedAt: new Date("2026-07-09T12:01:00.000Z"),
finishedAt: new Date("2026-07-09T12:01:00.000Z"),
},
]);
await db.insert(heartbeatRunEvents).values({
companyId,
runId: failedRunId,
agentId: workerId,
seq: 1,
eventType: "lifecycle",
message: "Bounded retry exhausted after 4 scheduled attempts; no further automatic retry will be queued",
createdAt: new Date("2026-07-09T12:00:01.000Z"),
});
const feed = await attentionService(db).list(companyId, { userId: "board-user" });
expect(feed.items.filter((item) => item.sourceKind === "failed_run")).toEqual([]);
});
it("enriches interaction details with project, workspace, plan metadata, and images", async () => {
const { companyId, workerId } = await seedCompany("ATE");
const projectId = randomUUID();
+29 -10
View File
@@ -1042,19 +1042,38 @@ export function attentionService(db: Db) {
failedIssueIds,
);
const failedImageMap = await issueImageMap(db, companyId, failedIssueIds);
for (const run of failedRows) {
const issueId = readRunIssueId(run.contextSnapshot);
const newer = await db
.select({ id: heartbeatRuns.id })
const failedAgentIds = [...new Set(failedRows.map((row) => row.agentId))];
const oldestFailedRunCreatedAt = failedRows.reduce<Date | null>((oldest, row) => {
if (!oldest || row.createdAt < oldest) return row.createdAt;
return oldest;
}, null);
const latestRunCreatedAtByKey = new Map<string, Date>();
if (oldestFailedRunCreatedAt && failedAgentIds.length > 0) {
const newerRuns = await db
.select({
agentId: heartbeatRuns.agentId,
createdAt: heartbeatRuns.createdAt,
contextSnapshot: heartbeatRuns.contextSnapshot,
})
.from(heartbeatRuns)
.where(and(
eq(heartbeatRuns.companyId, companyId),
eq(heartbeatRuns.agentId, run.agentId),
gt(heartbeatRuns.createdAt, run.createdAt),
sql`coalesce(${heartbeatRuns.contextSnapshot} ->> 'issueId', ${heartbeatRuns.contextSnapshot} ->> 'taskId', '') = ${issueId ?? ""}`,
))
.limit(1);
if (newer.length > 0) continue;
inArray(heartbeatRuns.agentId, failedAgentIds),
gt(heartbeatRuns.createdAt, oldestFailedRunCreatedAt),
));
for (const newerRun of newerRuns) {
const newerRunKey = `${newerRun.agentId}:${readRunIssueId(newerRun.contextSnapshot) ?? ""}`;
const latestCreatedAt = latestRunCreatedAtByKey.get(newerRunKey);
if (!latestCreatedAt || newerRun.createdAt > latestCreatedAt) {
latestRunCreatedAtByKey.set(newerRunKey, newerRun.createdAt);
}
}
}
for (const run of failedRows) {
const issueId = readRunIssueId(run.contextSnapshot);
const runKey = `${run.agentId}:${issueId ?? ""}`;
const hasNewerRun = (latestRunCreatedAtByKey.get(runKey)?.getTime() ?? 0) > run.createdAt.getTime();
if (hasNewerRun) continue;
const issue = issueId ? failedIssueMap.get(issueId) ?? null : null;
const dedupKey = `run:${run.id}`;
@@ -1431,89 +1431,97 @@ export function issueThreadInteractionService(db: Db) {
actor: InteractionActor,
): Promise<{ interaction: IssueThreadInteraction; newlyResolvedItemIds: string[] }> => {
const data = submitIssueThreadInteractionVerdictsSchema.parse(input);
const current = await db
.select()
.from(issueThreadInteractions)
.where(eq(issueThreadInteractions.id, interactionId))
.then((rows) => rows[0] ?? null);
const submission = await db.transaction(async (tx) => {
const current = await tx
.select()
.from(issueThreadInteractions)
.where(eq(issueThreadInteractions.id, interactionId))
.for("update")
.then((rows) => rows[0] ?? null);
if (!current) throw notFound("Interaction not found");
if (current.companyId !== issue.companyId || current.issueId !== issue.id) {
throw notFound("Interaction not found");
}
if (current.kind !== "request_item_verdicts") {
throw unprocessable("Only request_item_verdicts interactions can receive item verdicts");
}
const interaction = hydrateInteraction(current) as RequestItemVerdictsInteraction;
if (current.status !== "pending") {
if (current.status === "answered") {
const resolvedIds = new Set(interaction.result?.items.map((item) => item.id) ?? []);
const payloadIds = new Set(interaction.payload.items.map((item) => item.id));
for (const submitted of data.verdicts) {
if (!payloadIds.has(submitted.id)) {
throw unprocessable(`Unknown item verdict id: ${submitted.id}`);
}
if (!resolvedIds.has(submitted.id)) {
throw conflict("Interaction has already been resolved");
}
}
return { interaction, newlyResolvedItemIds: [] };
if (!current) throw notFound("Interaction not found");
if (current.companyId !== issue.companyId || current.issueId !== issue.id) {
throw notFound("Interaction not found");
}
if (current.kind !== "request_item_verdicts") {
throw unprocessable("Only request_item_verdicts interactions can receive item verdicts");
}
throw conflict("Interaction has already been resolved");
}
const expired = await expireStaleRequestConfirmationTarget(db, {
row: current,
actor,
const interaction = hydrateInteraction(current) as RequestItemVerdictsInteraction;
if (current.status !== "pending") {
if (current.status === "answered") {
const resolvedIds = new Set(interaction.result?.items.map((item) => item.id) ?? []);
const payloadIds = new Set(interaction.payload.items.map((item) => item.id));
for (const submitted of data.verdicts) {
if (!payloadIds.has(submitted.id)) {
throw unprocessable(`Unknown item verdict id: ${submitted.id}`);
}
if (!resolvedIds.has(submitted.id)) {
throw conflict("Interaction has already been resolved");
}
}
return { interaction, newlyResolvedItemIds: [], resolved: false };
}
throw conflict("Interaction has already been resolved");
}
const expired = await expireStaleRequestConfirmationTarget(tx, {
row: current,
actor,
});
if (expired) {
return { interaction: expired, newlyResolvedItemIds: [], resolved: false };
}
const now = new Date();
const { items, complete, newlyResolvedItemIds } = resolveRequestItemVerdictSubmissions({
interaction,
input: data,
actor,
now,
});
if (newlyResolvedItemIds.length === 0) {
return { interaction, newlyResolvedItemIds: [], resolved: false };
}
const result = {
version: 1,
outcome: "resolved",
complete,
items,
} satisfies RequestItemVerdictsResult;
const [updated] = await tx
.update(issueThreadInteractions)
.set({
status: complete ? "answered" : "pending",
result,
resolvedByAgentId: complete ? actor.agentId ?? null : null,
resolvedByUserId: complete ? actor.userId ?? null : null,
resolvedAt: complete ? now : null,
updatedAt: now,
})
.where(and(
eq(issueThreadInteractions.id, interactionId),
eq(issueThreadInteractions.status, "pending"),
))
.returning();
if (!updated) {
throw conflict("Interaction has already been resolved");
}
await touchIssue(tx, issue.id);
return {
interaction: hydrateInteraction(updated),
newlyResolvedItemIds,
resolved: complete,
};
});
if (expired) {
return { interaction: expired, newlyResolvedItemIds: [] };
}
const now = new Date();
const { items, complete, newlyResolvedItemIds } = resolveRequestItemVerdictSubmissions({
interaction,
input: data,
actor,
now,
});
if (newlyResolvedItemIds.length === 0) {
return { interaction, newlyResolvedItemIds: [] };
if (submission.resolved) {
await emitInteractionResolvedTelemetry(db, submission.interaction);
}
const result = {
version: 1,
outcome: "resolved",
complete,
items,
} satisfies RequestItemVerdictsResult;
const [updated] = await db
.update(issueThreadInteractions)
.set({
status: complete ? "answered" : "pending",
result,
resolvedByAgentId: complete ? actor.agentId ?? null : null,
resolvedByUserId: complete ? actor.userId ?? null : null,
resolvedAt: complete ? now : null,
updatedAt: now,
})
.where(and(
eq(issueThreadInteractions.id, interactionId),
eq(issueThreadInteractions.status, "pending"),
))
.returning();
if (!updated) {
throw conflict("Interaction has already been resolved");
}
await touchIssue(db, issue.id);
const submitted = hydrateInteraction(updated);
if (complete) {
await emitInteractionResolvedTelemetry(db, submitted);
}
return { interaction: submitted, newlyResolvedItemIds };
return submission;
},
rejectSuggestedTasks: async (