diff --git a/doc/DATABASE.md b/doc/DATABASE.md index 966595b7536..ed99988523b 100644 --- a/doc/DATABASE.md +++ b/doc/DATABASE.md @@ -49,6 +49,15 @@ and leaves all existing cases passive. Source identities are not grants. ### Task query performance +The attention feed checks for newer runs with company-scoped existence probes. +The probes match the failed run's agent and issue or task context. They preserve +the existing millisecond comparison and empty-context rules. Each batch holds +at most 1,000 failed runs. Retained run history stays in the database; this check +returns only the identifiers of failures that have a newer matching run. +For failures without a task key, it reads the latest keyless timestamp once per +agent. Unique agents are batched separately, so repeated failures do not repeat +the same history scan. Both paths retain the current millisecond comparison. + The issue-lock adapter's release read omits `result_json.stdout` and `result_json.stderr` before transferring a run to the controller. Native context also omits saved launch text, execution-continuation bodies, workspace and diff --git a/server/src/__tests__/attention-service.test.ts b/server/src/__tests__/attention-service.test.ts index 26934470fbd..bf4879957cd 100644 --- a/server/src/__tests__/attention-service.test.ts +++ b/server/src/__tests__/attention-service.test.ts @@ -2,7 +2,8 @@ import { createAgentLifecycle } from "../services/agent-lifecycle.js"; import { randomUUID } from "node:crypto"; import express from "express"; import request from "supertest"; -import { eq } from "drizzle-orm"; +import { eq, sql } from "drizzle-orm"; +import { createDatabaseResponseCapture, installDatabaseResponseCapture } from "@paperclipai/db/response-capture"; import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; import { activityLog, @@ -1034,6 +1035,177 @@ describeEmbeddedPostgres("attention service", () => { expect(feed.items.filter((item) => item.sourceKind === "failed_run")).toEqual([]); }); + it("hides a failed run only after a newer run of the same agent on the same issue or task", async () => { + const { companyId, workerId, reviewerId } = await seedCompany("ATK"); + const taskIssueId = await insertIssue({ companyId, identifier: "ATK-1", title: "Task-keyed", status: "in_progress" }); + const issueKeyedId = await insertIssue({ companyId, identifier: "ATK-2", title: "Issue-keyed", status: "in_progress" }); + const otherIssueId = await insertIssue({ companyId, identifier: "ATK-3", title: "Other work", status: "in_progress" }); + const failedAt = new Date("2026-07-09T12:00:00.000Z"); + const laterAt = new Date("2026-07-09T12:01:00.000Z"); + const taskKeyedFailureId = randomUUID(); + const issueKeyedFailureId = randomUUID(); + const unkeyedHiddenFailureId = randomUUID(); + const unkeyedKeptFailureId = randomUUID(); + const failureIds = [taskKeyedFailureId, issueKeyedFailureId, unkeyedHiddenFailureId, unkeyedKeptFailureId]; + const sameMillisecondRunId = randomUUID(); + const run = (id: string, agentId: string, status: string, contextSnapshot: Record, at: Date) => ({ + id, companyId, agentId, invocationSource: "automation", status, contextSnapshot, + createdAt: at, updatedAt: at, finishedAt: at, ...(status === "failed" ? { error: "adapter failed" } : {}), + }); + + await db.insert(heartbeatRuns).values([ + run(taskKeyedFailureId, workerId, "failed", { taskId: taskIssueId }, failedAt), + run(issueKeyedFailureId, workerId, "failed", { issueId: issueKeyedId }, failedAt), + run(unkeyedHiddenFailureId, workerId, "failed", {}, failedAt), + run(unkeyedKeptFailureId, reviewerId, "failed", {}, failedAt), + run(randomUUID(), workerId, "succeeded", { taskId: taskIssueId }, laterAt), + run(randomUUID(), workerId, "succeeded", { wakeReason: "timer" }, laterAt), + run(randomUUID(), reviewerId, "succeeded", { issueId: issueKeyedId }, laterAt), + run(randomUUID(), reviewerId, "succeeded", { issueId: otherIssueId }, laterAt), + run(randomUUID(), workerId, "succeeded", { issueId: otherIssueId }, laterAt), + run(randomUUID(), workerId, "succeeded", { issueId: otherIssueId, taskId: issueKeyedId }, laterAt), + run(randomUUID(), workerId, "succeeded", { issueId: "", taskId: issueKeyedId }, laterAt), + run(sameMillisecondRunId, workerId, "succeeded", { issueId: issueKeyedId }, failedAt), + run(randomUUID(), workerId, "succeeded", { issueId: issueKeyedId }, new Date("2026-07-09T11:59:00.000Z")), + ]); + await db.execute(sql`update heartbeat_runs set created_at = created_at + interval '400 microseconds' + where id in (${sql.join(failureIds.map((id) => sql`${id}::uuid`), sql`, `)})`); + await db.execute(sql`update heartbeat_runs set created_at = created_at + interval '600 microseconds' + where id = ${sameMillisecondRunId}::uuid`); + await db.insert(heartbeatRunEvents).values( + failureIds.map((runId) => ({ + companyId, + runId, + agentId: runId === unkeyedKeptFailureId ? reviewerId : 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").map((item) => item.subject.id).sort()) + .toEqual([issueKeyedFailureId, unkeyedKeptFailureId].sort()); + }); + + it("keeps newer-run response size bounded despite unrelated retained history", async () => { + const { companyId, workerId } = await seedCompany("ATW"); + const issueId = await insertIssue({ companyId, identifier: "ATW-1", title: "Exhausted task", status: "in_progress" }); + const otherIssueId = await insertIssue({ companyId, identifier: "ATW-2", title: "Other task", status: "done" }); + const failedId = randomUUID(), failedAt = new Date("2026-07-09T12:00:00.000Z"); + await db.insert(heartbeatRuns).values({ + id: failedId, companyId, agentId: workerId, status: "failed", contextSnapshot: { issueId }, + createdAt: failedAt, updatedAt: failedAt, finishedAt: failedAt, + }); + await db.insert(heartbeatRunEvents).values({ + companyId, agentId: workerId, runId: failedId, seq: 1, eventType: "lifecycle", + message: "Bounded retry exhausted after scheduled attempts", + }); + for (let batch = 0; batch < 4; batch += 1) { + await db.insert(heartbeatRuns).values(Array.from({ length: 500 }, (_, index) => ({ + id: randomUUID(), companyId, agentId: workerId, status: "succeeded", + contextSnapshot: { issueId: otherIssueId, prompt: "x".repeat(8_000) }, + createdAt: new Date(failedAt.getTime() + batch * 500 + index + 1), + updatedAt: failedAt, finishedAt: failedAt, + }))); + } + const events: Array> = [], now = Date.now(); + const capture = createDatabaseResponseCapture({ start: now - 100, end: now + 59_000 }, event => events.push(event)); + installDatabaseResponseCapture(db, { heartbeatRuns }, capture); + try { + const feed = await attentionService(db).list(companyId, { userId: "board-user" }); + expect(feed.items.filter(item => item.sourceKind === "failed_run").map(item => item.subject.id)).toEqual([failedId]); + } finally { capture.flush(); } + const reads = events.filter(event => event.event === "paperclip.db_response_capture.entry.v1" + && event.table === "heartbeat_runs" && (event.operation === "select" || event.operation === "execute")); + expect(reads.reduce((sum, event) => sum + Number(event.calls), 0)).toBeGreaterThan(0); + expect(reads.reduce((sum, event) => sum + Number(event.rows), 0)).toBeLessThan(64); + expect(reads.reduce((sum, event) => sum + Number(event.responseValueBytes), 0)).toBeLessThan(4096); + expect(reads.filter(event => event.operation === "execute") + .reduce((sum, event) => sum + Number(event.calls), 0)).toBeGreaterThan(0); + const [remaining] = await db.select({ count: sql`count(*)::int` }).from(heartbeatRuns) + .where(eq(heartbeatRuns.companyId, companyId)); + expect(remaining?.count).toBe(2001); + }); + + it("reads keyless history once per agent despite thousands of failures", async () => { + const { companyId, workerId, reviewerId } = await seedCompany("ATB"); + const failedAt = new Date("2026-07-09T12:00:00.000Z"); + const laterAt = new Date("2026-07-09T12:01:00.000Z"); + const hiddenIds = Array.from({ length: 16_390 }, () => randomUUID()); + const keptIds = Array.from({ length: 10 }, () => randomUUID()); + const failure = (id: string, agentId: string) => ({ + id, companyId, agentId, invocationSource: "automation", status: "failed", error: "adapter failed", + contextSnapshot: {}, createdAt: failedAt, updatedAt: failedAt, finishedAt: failedAt, + }); + const failures = hiddenIds.map((id) => failure(id, workerId)); + keptIds.forEach((id, index) => failures.splice(index * 1_700, 0, failure(id, reviewerId))); + for (let start = 0; start < failures.length; start += 1_000) { + const chunk = failures.slice(start, start + 1_000); + await db.insert(heartbeatRuns).values(chunk); + await db.insert(heartbeatRunEvents).values(chunk.map((run) => ({ + companyId, runId: run.id, agentId: run.agentId, 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"), + }))); + } + await db.insert(heartbeatRuns).values({ + id: randomUUID(), companyId, agentId: workerId, invocationSource: "automation", status: "succeeded", + contextSnapshot: { wakeReason: "timer" }, createdAt: laterAt, updatedAt: laterAt, finishedAt: laterAt, + }); + + const events: Array> = [], now = Date.now(); + const capture = createDatabaseResponseCapture({ start: now - 100, end: now + 59_000 }, event => events.push(event)); + installDatabaseResponseCapture(db, { heartbeatRuns }, capture); + try { + const feed = await attentionService(db).list(companyId, { + includeDismissed: true, all: true, allowUnscopedAll: true, + }); + expect(feed.items.filter((item) => item.sourceKind === "failed_run").map((item) => item.subject.id).sort()) + .toEqual([...keptIds].sort()); + } finally { capture.flush(); } + const reads = events.filter(event => event.event === "paperclip.db_response_capture.entry.v1" + && event.table === "heartbeat_runs" && (event.operation === "select" || event.operation === "execute")); + expect(reads.reduce((sum, event) => sum + Number(event.calls), 0)).toBeGreaterThan(0); + expect(reads.reduce((sum, event) => sum + Number(event.calls), 0)).toBeLessThan(4); + }, 180_000); + + it("checks more failed runs for a newer run than one statement can bind", async () => { + const { companyId, workerId, reviewerId } = await seedCompany("ATQ"); + const issueId = await insertIssue({ companyId, identifier: "ATQ-1", title: "Keyed retained failures", status: "in_progress" }); + const failedAt = new Date("2026-07-09T12:00:00.000Z"); + const laterAt = new Date("2026-07-09T12:01:00.000Z"); + const hiddenIds = Array.from({ length: 16_390 }, () => randomUUID()); + const keptIds = Array.from({ length: 10 }, () => randomUUID()); + const failure = (id: string, agentId: string) => ({ + id, companyId, agentId, invocationSource: "automation", status: "failed", error: "adapter failed", + contextSnapshot: { issueId }, createdAt: failedAt, updatedAt: failedAt, finishedAt: failedAt, + }); + const failures = hiddenIds.map((id) => failure(id, workerId)); + keptIds.forEach((id, index) => failures.splice(index * 1_700, 0, failure(id, reviewerId))); + for (let start = 0; start < failures.length; start += 1_000) { + const chunk = failures.slice(start, start + 1_000); + await db.insert(heartbeatRuns).values(chunk); + await db.insert(heartbeatRunEvents).values(chunk.map((run) => ({ + companyId, runId: run.id, agentId: run.agentId, 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"), + }))); + } + await db.insert(heartbeatRuns).values({ + id: randomUUID(), companyId, agentId: workerId, invocationSource: "automation", status: "succeeded", + contextSnapshot: { issueId }, createdAt: laterAt, updatedAt: laterAt, finishedAt: laterAt, + }); + + const feed = await attentionService(db).list(companyId, { + includeDismissed: true, all: true, allowUnscopedAll: true, + }); + expect(feed.items.filter(item => item.sourceKind === "failed_run").map(item => item.subject.id).sort()) + .toEqual([...keptIds].sort()); + }, 180_000); + 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 dd46de7d91e..90ed24f135e 100644 --- a/server/src/services/attention.ts +++ b/server/src/services/attention.ts @@ -1099,6 +1099,65 @@ function readRunIssueId(contextSnapshot: Record | null) { return typeof issueId === "string" && issueId.length > 0 ? issueId : null; } +type FailedRunProbe = { id: string; agentId: string; issueId: string | null; createdAt: Date }; + +const FAILED_RUN_PROBE_BATCH_SIZE = 1000; + +async function listFailedRunsWithNewerRun(db: Db, companyId: string, failedRuns: FailedRunProbe[]) { + const withNewerRun = new Set(); + const keyed = failedRuns.filter(run => run.issueId !== null); + for (let start = 0; start < keyed.length; start += FAILED_RUN_PROBE_BATCH_SIZE) { + const batch = keyed.slice(start, start + FAILED_RUN_PROBE_BATCH_SIZE); + for (const id of await listFailedRunBatchWithNewerRun(db, companyId, batch)) withNewerRun.add(id); + } + const unkeyed = failedRuns.filter(run => run.issueId === null); + if (unkeyed.length === 0) return withNewerRun; + const agentIds = [...new Set(unkeyed.map(run => run.agentId))]; + const oldestCreatedAt = unkeyed.reduce((oldest, run) => run.createdAt < oldest ? run.createdAt : oldest, + unkeyed[0]!.createdAt); + const latestByAgent = new Map(); + for (let start = 0; start < agentIds.length; start += FAILED_RUN_PROBE_BATCH_SIZE) { + const rows = await db.select({ + agentId: heartbeatRuns.agentId, + createdAt: sql`max(${heartbeatRuns.createdAt})`.mapWith(heartbeatRuns.createdAt), + }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, companyId), + inArray(heartbeatRuns.agentId, agentIds.slice(start, start + FAILED_RUN_PROBE_BATCH_SIZE)), + gt(heartbeatRuns.createdAt, oldestCreatedAt), + sql`coalesce(${heartbeatRuns.contextSnapshot} ->> 'issueId', ${heartbeatRuns.contextSnapshot} ->> 'taskId', '') = ''`, + )).groupBy(heartbeatRuns.agentId); + for (const row of rows) latestByAgent.set(row.agentId, row.createdAt); + } + for (const run of unkeyed) { + if ((latestByAgent.get(run.agentId)?.getTime() ?? 0) > run.createdAt.getTime()) withNewerRun.add(run.id); + } + return withNewerRun; +} + +async function listFailedRunBatchWithNewerRun(db: Db, companyId: string, failedRuns: FailedRunProbe[]) { + if (failedRuns.length === 0) return []; + const failed = sql.join( + failedRuns.map((run) => sql`(${run.id}::uuid, ${run.agentId}::uuid, ${run.issueId}::text, ${new Date(run.createdAt.getTime() + 1).toISOString()}::timestamptz)`), + sql`, `, + ); + const runIssueId = sql`(${heartbeatRuns.contextSnapshot} ->> 'issueId')`; + const runTaskId = sql`(${heartbeatRuns.contextSnapshot} ->> 'taskId')`; + const newerRunExists = (match: SQL) => sql`exists ( + select 1 from ${heartbeatRuns} + where ${heartbeatRuns.companyId} = ${companyId} + and ${heartbeatRuns.agentId} = failed.agent_id + and ${heartbeatRuns.createdAt} >= failed.newer_from + and ${match} + )`; + const rows = await db.execute(sql` + select failed.id + from (values ${failed}) as failed (id, agent_id, issue_id, newer_from) + where ${newerRunExists(sql`${runIssueId} = failed.issue_id`)} + or ${newerRunExists(sql`${runTaskId} = failed.issue_id and ${runIssueId} is null`)} + `) as unknown as Array<{ id: string }>; + return rows.map((row) => row.id); +} + export function attentionService(db: Db, serviceOptions: AttentionServiceOptions = {}) { const openDecisionLimit = Math.min( Math.max(Math.trunc(serviceOptions.openDecisionLimit ?? OPEN_DECISION_DEFAULT_LIMIT), 1), @@ -1717,12 +1776,7 @@ export function attentionService(db: Db, serviceOptions: AttentionServiceOptions const failedRows = await listAttentionExhaustedRuns(db, companyId); const failedIssueIds = failedRows.map((row) => readRunIssueId(row.contextSnapshot)); - 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 [failedIssueMap, failedImageMap, newerRuns] = await Promise.all([ + const [failedIssueMap, failedImageMap, failedRunsWithNewerRun] = await Promise.all([ issueSummaryMap( db, companyId, @@ -1730,38 +1784,16 @@ export function attentionService(db: Db, serviceOptions: AttentionServiceOptions readCondition, ), issueImageMap(db, companyId, failedIssueIds, readCondition), - oldestFailedRunCreatedAt && failedAgentIds.length > 0 - ? db - .select({ - agentId: heartbeatRuns.agentId, - createdAt: heartbeatRuns.createdAt, - // Project just the ids readRunIssueId needs; pulling the whole - // context_snapshot detoasts megabytes per feed build. - runIssueId: sql`${heartbeatRuns.contextSnapshot} ->> 'issueId'`, - runTaskId: sql`${heartbeatRuns.contextSnapshot} ->> 'taskId'`, - }) - .from(heartbeatRuns) - .where(and( - eq(heartbeatRuns.companyId, companyId), - inArray(heartbeatRuns.agentId, failedAgentIds), - gt(heartbeatRuns.createdAt, oldestFailedRunCreatedAt), - )) - : Promise.resolve([]), + listFailedRunsWithNewerRun(db, companyId, failedRows.map((row, index) => ({ + id: row.id, + agentId: row.agentId, + issueId: failedIssueIds[index], + createdAt: row.createdAt, + }))), ]); - const latestRunCreatedAtByKey = new Map(); - for (const newerRun of newerRuns) { - const newerRunIssueId = readRunIssueId({ issueId: newerRun.runIssueId, taskId: newerRun.runTaskId }); - const newerRunKey = `${newerRun.agentId}:${newerRunIssueId ?? ""}`; - 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; + if (failedRunsWithNewerRun.has(run.id)) continue; const issue = issueId ? failedIssueMap.get(issueId) ?? null : null; if (issueId && !issue) continue;