diff --git a/doc/run-log-events.md b/doc/run-log-events.md index b2b999a6f24..6c47dd19321 100644 --- a/doc/run-log-events.md +++ b/doc/run-log-events.md @@ -159,6 +159,14 @@ Provider identity diagnostics remain in the local run log. They record the notif Recovery lifecycle events retain the original structured failure code, retry attempt, next retry time, and predecessor/successor identifiers. Durable status delivery uses an idempotency marker; delivery grants no provider authority. Failed publication is retried without repeating provider work. These records are not first-party Telemetry. +Bounded retry exhaustion writes one lifecycle receipt per run, retry reason, +scheduled attempt, and retry limit. Repeated or concurrent recovery checks reuse +that receipt, including receipts from earlier builds, without advancing the event +sequence or publishing another live event. Attention reads select the latest +matching receipt in PostgreSQL and project only the run's issue/task identifiers +from its context, so historical duplicate receipts cannot multiply run contexts +in server memory. Existing duplicate events do not require deletion or migration. + ## Codex resume usage snapshot The native runner retains a bounded local `harness.diagnostic` event with code diff --git a/server/src/__tests__/attention-service.test.ts b/server/src/__tests__/attention-service.test.ts index 930a3dda2a4..889b9a9ace6 100644 --- a/server/src/__tests__/attention-service.test.ts +++ b/server/src/__tests__/attention-service.test.ts @@ -42,6 +42,7 @@ import { import { errorHandler } from "../middleware/index.js"; import { attentionRoutes } from "../routes/attention.js"; import { attentionService } from "../services/attention.js"; +import { listAttentionExhaustedRuns } from "../services/attention-exhausted-runs.js"; import { agentService } from "../services/agents.js"; import { ROUTABLE_BLOCKED_ROLLOUT_AT } from "../services/routable-blocked.js"; @@ -918,6 +919,63 @@ describeEmbeddedPostgres("attention service", () => { .toBeNull(); }); + it("reads one compact row per exhausted run despite thousands of historical receipts", async () => { + const { companyId, workerId, reviewerId } = await seedCompany("MEM"); + const other = await seedCompany("OTH"); + await db.update(agents).set({ status: "terminated" }).where(eq(agents.id, reviewerId)); + const issueId = await insertIssue({ + companyId, identifier: "MEM-1", title: "Failed task", status: "in_progress", + }); + const taskId = await insertIssue({ + companyId, identifier: "MEM-2", title: "Timed out task", status: "in_progress", + }); + const [failedId, timedOutId, succeededId, terminatedId, foreignId, noReceiptId] = + Array.from({ length: 6 }, () => randomUUID()); + const createdAt = new Date("2026-07-09T12:00:00.000Z"); + await db.insert(heartbeatRuns).values([ + { id: failedId, companyId, agentId: workerId, status: "failed", contextSnapshot: { issueId, prompt: "x".repeat(32_000) } }, + { id: timedOutId, companyId, agentId: workerId, status: "timed_out", contextSnapshot: { taskId, prompt: "y".repeat(32_000) } }, + { id: succeededId, companyId, agentId: workerId, status: "succeeded" }, + { id: terminatedId, companyId, agentId: reviewerId, status: "failed" }, + { id: foreignId, companyId: other.companyId, agentId: other.workerId, status: "failed" }, + { id: noReceiptId, companyId, agentId: workerId, status: "failed" }, + ].map((run) => ({ ...run, createdAt, updatedAt: createdAt, finishedAt: createdAt }))); + for (let batch = 0; batch < 5; batch += 1) { + await db.insert(heartbeatRunEvents).values(Array.from({ length: 500 }, (_, index) => ({ + companyId, agentId: workerId, runId: failedId, seq: batch * 500 + index + 1, + eventType: "lifecycle", message: `Bounded retry exhausted receipt ${batch * 500 + index + 1}`, + }))); + } + await db.insert(heartbeatRunEvents).values([ + { companyId, agentId: workerId, runId: timedOutId, seq: 1, eventType: "lifecycle", message: "Bounded retry exhausted timeout" }, + { companyId, agentId: workerId, runId: succeededId, seq: 1, eventType: "lifecycle", message: "Bounded retry exhausted success" }, + { companyId, agentId: reviewerId, runId: terminatedId, seq: 1, eventType: "lifecycle", message: "Bounded retry exhausted terminated" }, + { companyId: other.companyId, agentId: other.workerId, runId: foreignId, seq: 1, eventType: "lifecycle", message: "Bounded retry exhausted other company" }, + // A newer event must not replace the latest matching, company-scoped receipt. + { companyId: other.companyId, agentId: other.workerId, runId: failedId, seq: 2501, eventType: "lifecycle", message: "Bounded retry exhausted foreign receipt" }, + { companyId, agentId: workerId, runId: failedId, seq: 2502, eventType: "stdout", message: "Bounded retry exhausted quoted output" }, + { companyId, agentId: workerId, runId: failedId, seq: 2503, eventType: "lifecycle", message: "Unrelated lifecycle event" }, + ]); + + // Assert the database result itself: JavaScript feed deduplication used to + // hide the thousands of full run contexts already loaded into memory. + const rows = await listAttentionExhaustedRuns(db, companyId); + expect(rows).toHaveLength(2); + expect(rows.find((row) => row.id === failedId)).toMatchObject({ + exhaustionMessage: "Bounded retry exhausted receipt 2500", + contextSnapshot: { issueId, taskId: null }, + }); + expect(rows.find((row) => row.id === timedOutId)?.contextSnapshot).toEqual({ issueId: null, taskId }); + expect(Buffer.byteLength(JSON.stringify(rows))).toBeLessThan(4096); + + const feed = await attentionService(db).list(companyId, { + includeDismissed: true, all: true, allowUnscopedAll: true, + }); + const failures = feed.items.filter((item) => item.sourceKind === "failed_run"); + expect(failures.map((item) => item.subject.id).sort()).toEqual([failedId, timedOutId].sort()); + expect(failures.find((item) => item.subject.id === timedOutId)?.relatedIssue?.id).toBe(taskId); + }); + it("suppresses failed-run attention after a newer run for the same issue", async () => { const { companyId, workerId } = await seedCompany("ATN"); const issueId = await insertIssue({ diff --git a/server/src/__tests__/heartbeat-retry-scheduling.test.ts b/server/src/__tests__/heartbeat-retry-scheduling.test.ts index 38bbf2bb49b..3289bf0774e 100644 --- a/server/src/__tests__/heartbeat-retry-scheduling.test.ts +++ b/server/src/__tests__/heartbeat-retry-scheduling.test.ts @@ -55,6 +55,7 @@ vi.mock("../services/heartbeat-run-events.js", async (importOriginal) => { }); import { appendHeartbeatRunEvent } from "../services/heartbeat-run-events.js"; +import { subscribeCompanyLiveEvents } from "../services/live-events.js"; import { BOUNDED_TRANSIENT_HEARTBEAT_RETRY_DELAYS_MS, INTERACTION_CONTINUATION_INFRA_RETRY_REASON, @@ -1835,6 +1836,23 @@ describeEmbeddedPostgres("heartbeat bounded retry scheduling", () => { scheduledRetryAttempt: BOUNDED_TRANSIENT_HEARTBEAT_RETRY_DELAYS_MS.length, maxAttempts: BOUNDED_TRANSIENT_HEARTBEAT_RETRY_DELAYS_MS.length, }); + + const onLiveEvent = vi.fn(); + const unsubscribe = subscribeCompanyLiveEvents(companyId, onLiveEvent); + try { + const restartedHeartbeat = heartbeatService(createDb(tempDb!.connectionString)); + const repeated = await Promise.all(Array.from({ length: 8 }, (_, index) => + (index % 2 ? heartbeat : restartedHeartbeat).scheduleBoundedRetry(cappedRunId, { + now, random: () => 0.5, + }))); + expect(repeated).toEqual(Array.from({ length: 8 }, () => exhausted)); + expect(await db.select().from(heartbeatRunEvents) + .where(eq(heartbeatRunEvents.runId, cappedRunId))).toHaveLength(1); + expect((await heartbeat.getRun(cappedRunId))?.nextEventSeq).toBe(2); + expect(onLiveEvent).not.toHaveBeenCalled(); + } finally { + unsubscribe(); + } }); it("advances codex transient fallback stages across bounded retry attempts", async () => { diff --git a/server/src/__tests__/heartbeat-run-event-sequencing.test.ts b/server/src/__tests__/heartbeat-run-event-sequencing.test.ts index 8af2fe3251c..28c2dd00d24 100644 --- a/server/src/__tests__/heartbeat-run-event-sequencing.test.ts +++ b/server/src/__tests__/heartbeat-run-event-sequencing.test.ts @@ -15,6 +15,67 @@ import { import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; describe("P6-11..13 / P6-17 canonical event allocator", () => { + it.each([false, true])("deduplicates concurrent retry exhaustion (historical receipt: %s)", async (historical) => { + const temporary = await startEmbeddedPostgresTestDatabase("paperclip-exhaustion-events-"); + const db = createDb(temporary.connectionString); + const otherDb = createDb(temporary.connectionString); + const companyId = "20000000-0000-4000-8000-000000000001"; + const agentId = "20000000-0000-4000-8000-000000000002"; + const runId = "20000000-0000-4000-8000-000000000003"; + const otherRunId = "20000000-0000-4000-8000-000000000004"; + const exhaustion = { retryReason: "transient_failure", scheduledRetryAttempt: 2, maxAttempts: 2 }; + const event = { + companyId, agentId, runId, eventType: "lifecycle", + message: "Bounded retry exhausted after 2 scheduled attempts; no further automatic retry will be queued", + payload: exhaustion, + }; + try { + await db.insert(companies).values({ id: companyId, name: "Exhaustion fixture", issuePrefix: "EXH" }); + await db.insert(agents).values({ id: agentId, companyId, name: "Exhaustion agent" }); + await db.insert(heartbeatRuns).values([runId, otherRunId].map((id) => ({ + id, companyId, agentId, status: "failed", + }))); + if (historical) { + // Old builds wrote ordinary lifecycle rows with no new idempotency field. + await appendHeartbeatRunEvent(db, event); + } + const receipts = await Promise.all(Array.from({ length: 16 }, (_, index) => + appendHeartbeatRunEvent(index % 2 ? db : otherDb, { + ...event, retryExhaustion: exhaustion, + }))); + expect(receipts.filter((receipt) => receipt.disposition === "committed")) + .toHaveLength(historical ? 0 : 1); + expect(new Set(receipts.map((receipt) => receipt.row.id)).size).toBe(1); + expect(await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, runId))) + .toHaveLength(1); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)))[0]?.nextEventSeq) + .toBe(2); + + // The key is per run, reason, attempt, and budget. Distinct exhaustion + // decisions still get their own receipt and an ordinary sequence number. + for (const changed of [ + { ...exhaustion, retryReason: "bootstrap_failure" }, + { ...exhaustion, scheduledRetryAttempt: 3 }, + { ...exhaustion, maxAttempts: 3 }, + ]) { + expect((await appendHeartbeatRunEvent(db, { + ...event, payload: changed, retryExhaustion: changed, + })).disposition).toBe("committed"); + } + expect((await appendHeartbeatRunEvent(db, { + ...event, runId: otherRunId, retryExhaustion: exhaustion, + })).disposition).toBe("committed"); + const rows = await db.select().from(heartbeatRunEvents) + .where(eq(heartbeatRunEvents.runId, runId)).orderBy(heartbeatRunEvents.seq); + expect(rows.map((row) => row.seq)).toEqual([1, 2, 3, 4]); + await expect(appendHeartbeatRunEvent(db, { + ...event, companyId: "20000000-0000-4000-8000-000000000099", retryExhaustion: exhaustion, + })).rejects.toThrow("heartbeat_run_event_binding_mismatch"); + } finally { + await temporary.cleanup(); + } + }, 60_000); + it("serializes concurrent writers and rejects conflicting replay without cursor drift", async () => { const temporary = await startEmbeddedPostgresTestDatabase("paperclip-native-events-"); const db = createDb(temporary.connectionString); diff --git a/server/src/services/attention-exhausted-runs.ts b/server/src/services/attention-exhausted-runs.ts new file mode 100644 index 00000000000..29f3978dd22 --- /dev/null +++ b/server/src/services/attention-exhausted-runs.ts @@ -0,0 +1,52 @@ +import { and, asc, desc, eq, inArray, notInArray, sql } from "drizzle-orm"; +import { agents, heartbeatRunEvents, heartbeatRuns, type Db } from "@paperclipai/db"; + +export function listAttentionExhaustedRuns(db: Db, companyId: string) { + // Recovery can revisit an exhausted run. Deduplicate its historical events + // before joining run data so duplicate events never multiply the wire payload. + const latestExhaustion = db + .selectDistinctOn([heartbeatRunEvents.runId], { + runId: heartbeatRunEvents.runId, + eventId: heartbeatRunEvents.id, + message: heartbeatRunEvents.message, + }) + .from(heartbeatRunEvents) + .where(and( + eq(heartbeatRunEvents.companyId, companyId), + eq(heartbeatRunEvents.eventType, "lifecycle"), + sql`${heartbeatRunEvents.message} like 'Bounded retry exhausted%'`, + )) + .orderBy(asc(heartbeatRunEvents.runId), desc(heartbeatRunEvents.id)) + .as("latest_exhaustion"); + + return db + .select({ + id: heartbeatRuns.id, + companyId: heartbeatRuns.companyId, + agentId: heartbeatRuns.agentId, + agentName: agents.name, + status: heartbeatRuns.status, + error: heartbeatRuns.error, + errorCode: heartbeatRuns.errorCode, + // Preserve JSON types and issueId/taskId fallback semantics without + // transferring prompts, transcripts, or the rest of the run context. + contextSnapshot: sql>`jsonb_build_object( + 'issueId', ${heartbeatRuns.contextSnapshot} -> 'issueId', + 'taskId', ${heartbeatRuns.contextSnapshot} -> 'taskId' + )`, + createdAt: heartbeatRuns.createdAt, + updatedAt: heartbeatRuns.updatedAt, + finishedAt: heartbeatRuns.finishedAt, + exhaustionMessage: latestExhaustion.message, + }) + .from(heartbeatRuns) + .innerJoin(agents, eq(heartbeatRuns.agentId, agents.id)) + .innerJoin(latestExhaustion, eq(latestExhaustion.runId, heartbeatRuns.id)) + .where(and( + eq(heartbeatRuns.companyId, companyId), + eq(agents.companyId, companyId), + notInArray(agents.status, ["terminated"]), + inArray(heartbeatRuns.status, ["failed", "timed_out"]), + )) + .orderBy(desc(heartbeatRuns.createdAt), desc(latestExhaustion.eventId)); +} diff --git a/server/src/services/attention.ts b/server/src/services/attention.ts index 9f6a0f35806..e270f8906c2 100644 --- a/server/src/services/attention.ts +++ b/server/src/services/attention.ts @@ -11,7 +11,6 @@ import { decisionTrainingExamples, decisionTriage, decisions, - heartbeatRunEvents, heartbeatRuns, inboxDismissals, invites, @@ -49,6 +48,7 @@ import type { IssueReviewPolicy, } from "@paperclipai/shared"; import { badRequest } from "../errors.js"; +import { listAttentionExhaustedRuns } from "./attention-exhausted-runs.js"; import { budgetService } from "./budgets.js"; import { BLOCKER_ATTENTION_MAX_DEPTH, @@ -104,7 +104,6 @@ const SOURCE_RANK: Record = { const PENDING_INTERACTION_STATUSES = ["pending"] as const; const OPEN_RECOVERY_STATUSES = ["active", "escalated"] as const; const HUMAN_RECOVERY_OWNER_TYPES = ["user", "board"] as const; -const FAILED_RUN_STATUSES = ["failed", "timed_out"] as const; const DETAIL_EXCERPT_LENGTH = 160; const DETAIL_IMAGE_LIMIT = 3; const OPEN_DECISION_DEFAULT_LIMIT = 500; @@ -1662,40 +1661,7 @@ export function attentionService(db: Db, serviceOptions: AttentionServiceOptions })); } - const exhaustedRunRows = await db - .select({ - id: heartbeatRuns.id, - companyId: heartbeatRuns.companyId, - agentId: heartbeatRuns.agentId, - agentName: agents.name, - status: heartbeatRuns.status, - error: heartbeatRuns.error, - errorCode: heartbeatRuns.errorCode, - contextSnapshot: heartbeatRuns.contextSnapshot, - createdAt: heartbeatRuns.createdAt, - updatedAt: heartbeatRuns.updatedAt, - finishedAt: heartbeatRuns.finishedAt, - exhaustionMessage: heartbeatRunEvents.message, - }) - .from(heartbeatRuns) - .innerJoin(agents, eq(heartbeatRuns.agentId, agents.id)) - .innerJoin(heartbeatRunEvents, eq(heartbeatRunEvents.runId, heartbeatRuns.id)) - .where(and( - eq(heartbeatRuns.companyId, companyId), - eq(agents.companyId, companyId), - notInArray(agents.status, ["terminated"]), - inArray(heartbeatRuns.status, [...FAILED_RUN_STATUSES]), - eq(heartbeatRunEvents.companyId, companyId), - eq(heartbeatRunEvents.eventType, "lifecycle"), - sql`${heartbeatRunEvents.message} like 'Bounded retry exhausted%'`, - )) - .orderBy(desc(heartbeatRuns.createdAt), desc(heartbeatRunEvents.id)); - - const latestExhaustedByRunId = new Map(); - for (const row of exhaustedRunRows) { - if (!latestExhaustedByRunId.has(row.id)) latestExhaustedByRunId.set(row.id, row); - } - const failedRows = [...latestExhaustedByRunId.values()]; + 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) => { diff --git a/server/src/services/heartbeat-run-events.ts b/server/src/services/heartbeat-run-events.ts index 7759fde127b..e90f5c4e7d0 100644 --- a/server/src/services/heartbeat-run-events.ts +++ b/server/src/services/heartbeat-run-events.ts @@ -13,6 +13,12 @@ export interface AppendHeartbeatRunEventInput { color?: string | null; message?: string | null; payload?: Record | null; + /** Reuse an existing exhaustion receipt, including receipts from older builds. */ + retryExhaustion?: { + retryReason: string; + scheduledRetryAttempt: number; + maxAttempts: number; + }; nativeSource?: { sourceInstanceId: string; sourceEventId: string; @@ -74,6 +80,32 @@ export async function appendHeartbeatRunEvent( throw new Error("heartbeat_run_event_binding_mismatch"); } + if (input.retryExhaustion && !input.nativeSource) { + // The run lock also serializes concurrent recovery checks across server + // instances. Reusing the receipt must not allocate a sequence or publish + // another live event on each scheduler tick. + const existing = await tx + .select() + .from(heartbeatRunEvents) + .where(and( + eq(heartbeatRunEvents.companyId, input.companyId), + eq(heartbeatRunEvents.runId, input.runId), + eq(heartbeatRunEvents.agentId, input.agentId), + eq(heartbeatRunEvents.eventType, "lifecycle"), + sql`${heartbeatRunEvents.message} like 'Bounded retry exhausted%'`, + sql`${heartbeatRunEvents.payload} @> ${JSON.stringify(input.retryExhaustion)}::jsonb`, + )) + .limit(1) + .then((rows) => rows[0] ?? null); + if (existing) { + return { + row: existing, + disposition: "duplicate" as const, + highestContiguousSourceSeq: 0, + }; + } + } + const sourceHash = input.nativeSource ? nativeSha256(input.nativeSource.canonicalPayload) : null; diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index e982ba99059..9c0309254c1 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -176,6 +176,7 @@ import { publishLiveEvent } from "./live-events.js"; import { allocateHeartbeatRunEventSeq, appendHeartbeatRunEvent, + type AppendHeartbeatRunEventInput, } from "./heartbeat-run-events.js"; import { queuedCommentIdsFromWakePayload, @@ -13594,6 +13595,7 @@ export function heartbeatService( color?: string; message?: string; payload?: Record; + retryExhaustion?: AppendHeartbeatRunEventInput["retryExhaustion"]; }, ) { const eventAt = new Date(); @@ -13632,7 +13634,9 @@ export function heartbeatService( color: event.color, message: sanitizedMessage, payload: sanitizedPayload, + retryExhaustion: event.retryExhaustion, }); + if (persistedEvent.disposition === "duplicate") return; const seq = persistedEvent.row.seq; publishLiveEvent({ @@ -15150,16 +15154,18 @@ export function heartbeatService( const issueId = readNonEmptyString(contextSnapshot.issueId); if (!baseSchedule) { + const exhaustion = { + retryReason, + scheduledRetryAttempt: run.scheduledRetryAttempt ?? 0, + maxAttempts, + }; await appendRunEvent(run, { eventType: "lifecycle", stream: "system", level: "warn", message: `Bounded retry exhausted after ${run.scheduledRetryAttempt ?? 0} scheduled attempts; no further automatic retry will be queued`, - payload: { - retryReason, - scheduledRetryAttempt: run.scheduledRetryAttempt ?? 0, - maxAttempts, - }, + payload: exhaustion, + retryExhaustion: exhaustion, }); if (retryReason === INTERACTION_CONTINUATION_INFRA_RETRY_REASON) { await escalatePlanApprovalResumeFailureNeedsAttention({