diff --git a/server/src/__tests__/attention-service.test.ts b/server/src/__tests__/attention-service.test.ts index 7fa7da5f73b..694929f7457 100644 --- a/server/src/__tests__/attention-service.test.ts +++ b/server/src/__tests__/attention-service.test.ts @@ -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(); diff --git a/server/src/services/attention.ts b/server/src/services/attention.ts index 84a8c2bebe7..203a263cf89 100644 --- a/server/src/services/attention.ts +++ b/server/src/services/attention.ts @@ -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((oldest, row) => { + if (!oldest || row.createdAt < oldest) return row.createdAt; + return oldest; + }, null); + const latestRunCreatedAtByKey = new Map(); + 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}`; diff --git a/server/src/services/issue-thread-interactions.ts b/server/src/services/issue-thread-interactions.ts index b9538de9a6e..5a2a3ae19da 100644 --- a/server/src/services/issue-thread-interactions.ts +++ b/server/src/services/issue-thread-interactions.ts @@ -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 (