From 12681d1611920f35ea4944ca31172f165059e002 Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Sat, 7 Nov 2026 11:31:46 -0800 Subject: [PATCH] fix(sentry): stop resending retained task interruption failures (#19041) Keep task-state Sentry reports within the recent saved-change window while retaining reports for newly failed recovery and delayed Retry preparation. Include the safe run error code without changing event grouping. Verified 402 relevant local tests, every server typecheck layer, all final-head CI checks, and Greptile 5/5 with no unresolved review threads. Co-Authored-By: Claude Opus 5.5 Co-Authored-By: Paperclip --- doc/observability.md | 11 +- .../src/__tests__/run-failure-sentry.test.ts | 13 ++ .../__tests__/task-interruption-query.test.ts | 134 +++++++++++++++++- server/src/sentry.ts | 8 +- .../task-interruption-presentation.ts | 55 +++++-- 5 files changed, 205 insertions(+), 16 deletions(-) diff --git a/doc/observability.md b/doc/observability.md index ff80fe858e7..46a39f5c749 100644 --- a/doc/observability.md +++ b/doc/observability.md @@ -1026,7 +1026,16 @@ receives a successful HTTP delivery response. The process retains the latest processing backpressure, network errors and rate limits can discard the event. Unconfirmed captures wait 60 seconds before a later task read can try again; this pending cache also holds at most 1,000 entries and starts no background -timer. A server restart or cache eviction can report an old failure again. +timer. Because these caches are process-local, a read reports only a state +change from the last hour: the run's finish time, a saved recovery transition, +a native resume's due or failure time, a scheduled retry time, a saved-message +or Retry request, or a Retry preparation's deadline or terminal update. Failed +or cancelled Retry preparations use their saved update time, so a new failure +is still reported after a long wait. Reading a task does not refresh these times. +A server restart, another replica or cache eviction therefore does not resend +every retained failure that a task list shows. A failed state that no task read observes within that hour is not +reported by this path. A valid run error code is added as the `run_error_code` +tag; it does not change the fingerprint. This task-state event can accompany a separate terminal run-failure report; the `failure_source` tag distinguishes them. diff --git a/server/src/__tests__/run-failure-sentry.test.ts b/server/src/__tests__/run-failure-sentry.test.ts index b8bcb5de4dc..6f42a6867d4 100644 --- a/server/src/__tests__/run-failure-sentry.test.ts +++ b/server/src/__tests__/run-failure-sentry.test.ts @@ -111,6 +111,19 @@ describe("captureRunFailure", () => { sentryModule.captureTaskInterruptionFailure({ ...event, intentId: "next-intent" }); sentryModule.captureTaskInterruptionFailure({ ...event, runId: "next-run" }); expect(captureException).toHaveBeenCalledTimes(3); + expect(captureException.mock.calls[2]![1]?.tags).not.toHaveProperty("run_error_code"); + }); + + it("tags a saved-message report with a valid run error code without splitting its group", async () => { + const { sentryModule, captureException } = await importFreshSentryWithGateOpen(); + const event = { companyId: "company", agentId: "agent", taskId: "task", agentAdapter: "process", + runStatus: "failed" as const, intentId: null }; + sentryModule.captureTaskInterruptionFailure({ ...event, runId: "run", runErrorCode: "native_workspace_sync_out_failed" }); + sentryModule.captureTaskInterruptionFailure({ ...event, runId: "other", runErrorCode: "token=abc secret" }); + expect(captureException.mock.calls[0]![1]).toMatchObject({ + tags: { run_error_code: "native_workspace_sync_out_failed" }, fingerprint: ["task_attempt_failed", "process"] }); + expect(captureException.mock.calls[1]![1]?.tags).not.toHaveProperty("run_error_code"); + expect(JSON.stringify(captureException.mock.calls[1])).not.toContain("secret"); }); it("does not consume a saved-message report while reporting is off or capture fails", async () => { diff --git a/server/src/__tests__/task-interruption-query.test.ts b/server/src/__tests__/task-interruption-query.test.ts index 02a1c12ba35..d41305b4a3c 100644 --- a/server/src/__tests__/task-interruption-query.test.ts +++ b/server/src/__tests__/task-interruption-query.test.ts @@ -1,8 +1,11 @@ import { randomUUID } from "node:crypto"; -import { eq } from "drizzle-orm"; +import { and, eq } from "drizzle-orm"; import { createRunSecretRedactionRegistry } from "../services/run-secret-redaction.js"; import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; -import { agents, agentWakeupRequests, companies, createDb, applyPendingMigrations, heartbeatRuns, issues, nativeRunFinalizations, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS } from "@paperclipai/db"; +import { agents, agentWakeupRequests, chatActions, chatEndpoints, companies, createDb, applyPendingMigrations, heartbeatRuns, issueRecoveryActions, issues, + toolApplications, toolConnections, + nativeRunFinalizations, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS } from "@paperclipai/db"; +import { createSystemRecovery } from "../services/system-recovery-state.js"; import { taskInterruptionPresentation, taskInterruptionPresentations } from "../services/task-interruption-presentation.js"; import { readNativeSessionResumptionAgent } from "../services/native-runtime/native-session-resumption-admission.js"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; @@ -24,6 +27,9 @@ describe.skipIf(!support.supported)("task interruption latest run query", () => db = createDb(database.connectionString); }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS); afterAll(async () => { await db?.$client.end(); await database?.cleanup(); }); + // The shared fixture's failures are historical; reporting tests need a current failure. + const markFailuresFresh = (companyId: string) => db.update(heartbeatRuns).set({ finishedAt: new Date() }) + .where(and(eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.status, "failed"))); it("shows a pending native same-run resume instead of a competing Retry", async () => { const companyId = randomUUID(), agentId = randomUUID(), runId = randomUUID(); @@ -290,6 +296,7 @@ describe.skipIf(!support.supported)("task interruption latest run query", () => visit(explain["QUERY PLAN"][0].Plan); expect(inspected.length).toBeGreaterThan(0); expect(Math.max(...inspected)).toBeLessThan(30); + await markFailuresFresh(f.companyId); const presentations = await taskInterruptionPresentations(db, f.companyId, f.input); for (const issue of f.input) expect(presentations.get(issue.id)).toMatchObject({ state: "failed", canRetry: true }); for (const issue of f.input) expect(captureTaskInterruptionFailure).toHaveBeenCalledWith(expect.objectContaining({ @@ -362,7 +369,7 @@ describe.skipIf(!support.supported)("task interruption latest run query", () => expect(result.get(input[n]!.id)).toMatchObject({ responsibility: "system", state: "failed" }); expect(captureTaskInterruptionFailure).toHaveBeenCalledWith({ companyId, agentId, taskId: input[n]!.id, runId: runs[n]!.id, - runStatus: runs[n]!.status, agentAdapter: "process", intentId: n === 1 ? wait!.id : null, + runStatus: runs[n]!.status, agentAdapter: "process", intentId: n === 1 ? wait!.id : null, runErrorCode: null, }); } expect(result.get(input[2]!.id)?.state).toBe("working"); @@ -371,6 +378,126 @@ describe.skipIf(!support.supported)("task interruption latest run query", () => for (const n of [6, 8]) expect(result.get(input[n]!.id)?.responsibility).toBe("user"); }); + it("reports a retained failure once at its state change, not on every later process's read", async () => { + const companyId = randomUUID(), agentId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Retained failures", issuePrefix: `R${companyId.slice(0, 7)}` }); + await db.insert(agents).values({ id: agentId, companyId, name: "Worker", adapterType: "process" }); + const input = await db.insert(issues).values(Array.from({ length: 3 }, (_, n) => ({ + companyId, title: `Retained ${n}`, status: "in_progress", assigneeAgentId: agentId, + }))).returning(); + const old = new Date(Date.now() - 22 * 60 * 60_000); + const runs = await db.insert(heartbeatRuns).values(input.map((issue, n) => ({ + companyId, agentId, issueId: issue.id, status: "failed", createdAt: old, + finishedAt: n === 2 ? new Date(Date.now() - 60_000) : old, + errorCode: n === 0 ? "native_workspace_sync_out_retry_exhausted" : null, + }))).returning(); + // A new saved message is a new state change for an old failed run. + const [wait] = await db.insert(agentWakeupRequests).values({ + companyId, agentId, source: "on_demand", status: "deferred_issue_execution", + requestedByActorType: "user", payload: { issueId: input[1]!.id }, requestedAt: new Date(Date.now() - 20_000), + }).returning(); + captureTaskInterruptionFailure.mockClear(); + const result = await taskInterruptionPresentations(db, companyId, input); + for (const issue of input) expect(result.get(issue.id)).toMatchObject({ responsibility: "system", state: "failed" }); + expect(captureTaskInterruptionFailure).toHaveBeenCalledTimes(2); + expect(captureTaskInterruptionFailure).toHaveBeenCalledWith(expect.objectContaining({ + taskId: input[1]!.id, runId: runs[1]!.id, intentId: wait!.id })); + expect(captureTaskInterruptionFailure).toHaveBeenCalledWith(expect.objectContaining({ + taskId: input[2]!.id, runId: runs[2]!.id, intentId: null })); + await db.update(heartbeatRuns).set({ finishedAt: new Date() }).where(eq(heartbeatRuns.id, runs[0]!.id)); + captureTaskInterruptionFailure.mockClear(); + await taskInterruptionPresentations(db, companyId, [input[0]!]); + expect(captureTaskInterruptionFailure).toHaveBeenCalledExactlyOnceWith(expect.objectContaining({ + taskId: input[0]!.id, runErrorCode: "native_workspace_sync_out_retry_exhausted" })); + }); + + it("reports recovery that newly fails on an old run once, then treats it as retained", async () => { + const companyId = randomUUID(), agentId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Old run recovery", issuePrefix: `O${companyId.slice(0, 7)}` }); + await db.insert(agents).values({ id: agentId, companyId, name: "Worker", adapterType: "process" }); + const [issue] = await db.insert(issues).values({ companyId, title: "Recovery", status: "in_progress", + assigneeAgentId: agentId }).returning(); + const old = new Date(Date.now() - 22 * 60 * 60_000); + const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId, issueId: issue!.id, status: "failed", + createdAt: old, finishedAt: old }).returning(); + // Exhausted recovery clears nextAttemptAt; only the saved transition time is recent. + const exhausted = { ...createSystemRecovery(run!.id), state: "exhausted" as const, attempts: 6, retriesUsed: 5, + firstFailureAt: old.getTime(), deadlineAt: old.getTime() + 30 * 60_000, nextAttemptAt: null }; + const [action] = await db.insert(issueRecoveryActions).values({ companyId, sourceIssueId: issue!.id, + kind: "workspace_recovery", status: "active", ownerType: "system", cause: "native_workspace_sync_out_failed", + fingerprint: `recovery:${run!.id}`, nextAction: "Recover saved files.", createdAt: old, updatedAt: new Date(), + evidence: { runId: run!.id, systemRecovery: exhausted } }).returning(); + captureTaskInterruptionFailure.mockClear(); + expect((await taskInterruptionPresentations(db, companyId, [issue!])).get(issue!.id)) + .toMatchObject({ responsibility: "system", state: "failed" }); + expect(captureTaskInterruptionFailure).toHaveBeenCalledExactlyOnceWith(expect.objectContaining({ + taskId: issue!.id, runId: run!.id })); + await db.update(issueRecoveryActions).set({ updatedAt: old }).where(eq(issueRecoveryActions.id, action!.id)); + captureTaskInterruptionFailure.mockClear(); + await taskInterruptionPresentations(db, companyId, [issue!]); + expect(captureTaskInterruptionFailure).not.toHaveBeenCalled(); + }); + + it("reports a native resume that is newly exhausted on an old run once, then treats it as retained", async () => { + const companyId = randomUUID(), agentId = randomUUID(), runId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Old native run", issuePrefix: `N${companyId.slice(0, 7)}` }); + await db.insert(agents).values({ id: agentId, companyId, name: "Native worker", adapterType: "paperclip_runner" }); + const [issue] = await db.insert(issues).values({ companyId, title: "Native resume", status: "in_progress", + assigneeAgentId: agentId }).returning(); + const old = new Date(Date.now() - 22 * 60 * 60_000); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, nativeIssueId: issue!.id, scopeKind: "issue", + status: "failed", runtimeMode: "native", errorCode: "native_session_interrupted", createdAt: old, finishedAt: old }); + // Only the coordinator's saved terminal transition is recent. + await db.insert(nativeRunFinalizations).values({ companyId, issueId: issue!.id, runId, phase: "terminal_failure", + attempt: 3, nextAttemptAt: null, createdAt: old, updatedAt: new Date() }); + captureTaskInterruptionFailure.mockClear(); + expect((await taskInterruptionPresentations(db, companyId, [issue!])).get(issue!.id)) + .toMatchObject({ responsibility: "system", state: "failed" }); + expect(captureTaskInterruptionFailure).toHaveBeenCalledExactlyOnceWith(expect.objectContaining({ + taskId: issue!.id, runId, runErrorCode: "native_session_interrupted" })); + await db.update(nativeRunFinalizations).set({ updatedAt: old }).where(eq(nativeRunFinalizations.runId, runId)); + captureTaskInterruptionFailure.mockClear(); + await taskInterruptionPresentations(db, companyId, [issue!]); + expect(captureTaskInterruptionFailure).not.toHaveBeenCalled(); + }); + + it.each(["failed", "cancelled"])("reports a delayed Retry that newly becomes %s, then ages out", async status => { + const companyId = randomUUID(), agentId = randomUUID(), runId = randomUUID(), intentId = randomUUID(), actionId = randomUUID(); + const old = new Date(Date.now() - 22 * 60 * 60_000), recent = new Date(Date.now() - 60_000); + await db.insert(companies).values({ id: companyId, name: "Delayed chat Retry", issuePrefix: "D" + companyId.slice(0, 7) }); + await db.insert(agents).values({ id: agentId, companyId, name: "Chat worker", adapterType: "claude_local" }); + const [issue] = await db.insert(issues).values({ companyId, title: "Deferred Retry", status: "in_progress", + assigneeAgentId: agentId }).returning(); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, issueId: issue!.id, + status: "failed", createdAt: old, finishedAt: old }); + const [application] = await db.insert(toolApplications).values({ companyId, name: "Chat", + applicationKey: randomUUID(), type: "chat" }).returning(); + const [connection] = await db.insert(toolConnections).values({ companyId, applicationId: application!.id, + name: "Chat", uid: randomUUID(), transport: "chat_sdk", connectionPurpose: "channel" }).returning(); + const [endpoint] = await db.insert(chatEndpoints).values({ companyId, connectionId: connection!.id, + assignedAgentId: agentId, provider: "slack", publicId: randomUUID(), status: "active" }).returning(); + const payload = { issueId: issue!.id, agentId, failedRunId: runId, taskRetryIntentId: intentId }; + await db.insert(chatActions).values({ id: actionId, companyId, endpointId: endpoint!.id, + kind: "failed_run_retry", providerActionId: randomUUID(), status, payload, createdAt: old, updatedAt: recent }); + await db.insert(agentWakeupRequests).values({ id: intentId, companyId, agentId, source: "on_demand", + status: "deferred_issue_execution", requestedByActorType: "user", requestedAt: old, + payload: { issueId: issue!.id, taskRetry: { sourceRunId: runId, chatActionId: actionId } } }); + captureTaskInterruptionFailure.mockClear(); + expect((await taskInterruptionPresentations(db, companyId, [issue!])).get(issue!.id)) + .toMatchObject({ responsibility: "system", state: "failed", nextRetryAt: null }); + expect(captureTaskInterruptionFailure).toHaveBeenCalledExactlyOnceWith(expect.objectContaining({ + taskId: issue!.id, runId, intentId })); + await db.update(chatActions).set({ updatedAt: old }).where(eq(chatActions.id, actionId)); + captureTaskInterruptionFailure.mockClear(); + await taskInterruptionPresentations(db, companyId, [issue!]); + expect(captureTaskInterruptionFailure).not.toHaveBeenCalled(); + await db.update(chatActions).set({ updatedAt: recent, payload: { ...payload, failedRunId: randomUUID() } }) + .where(eq(chatActions.id, actionId)); + captureTaskInterruptionFailure.mockClear(); + await taskInterruptionPresentations(db, companyId, [issue!]); + expect(captureTaskInterruptionFailure).not.toHaveBeenCalled(); + }); + it("keeps the failed run's adapter after the task is reassigned", async () => { const f = await seedInterruptionFixture(db, { selected: 1, history: 1, unrelated: 0, payloadBytes: 0 }); const [nextAgent] = await db.insert(agents).values({ @@ -378,6 +505,7 @@ describe.skipIf(!support.supported)("task interruption latest run query", () => }).returning(); const [reassigned] = await db.update(issues).set({ assigneeAgentId: nextAgent!.id }) .where(eq(issues.id, f.input[0]!.id)).returning(); + await markFailuresFresh(f.companyId); captureTaskInterruptionFailure.mockClear(); const result = await taskInterruptionPresentations(db, f.companyId, [reassigned!]); expect(result.get(reassigned!.id)?.state).toBe("failed"); diff --git a/server/src/sentry.ts b/server/src/sentry.ts index 6ab1ad469f0..e3d0787416d 100644 --- a/server/src/sentry.ts +++ b/server/src/sentry.ts @@ -220,6 +220,8 @@ export interface RunFailureEvent { /** The run status that triggered this report. */ runStatus: HeartbeatRunStatus; failureSource?: "task_interruption"; + /** A task-state event's observed run error code. It is a tag, not a fingerprint part. */ + runErrorCode?: string | null; /** Bounded, redacted diagnostics selected by the run failure reporter. */ diagnostics?: RunFailureDiagnostics; /** Recorded process exit evidence, when available. Validated before capture. */ @@ -269,6 +271,8 @@ function captureRunFailureEvent(event: RunFailureEvent): string | null { diagnostics?.exceptions.at(-1)?.code; const rootErrorCode = typeof rootCodeCandidate === "string" && /^[a-zA-Z][a-zA-Z0-9_]{1,99}$/.test(rootCodeCandidate) ? rootCodeCandidate : undefined; + const runErrorCode = typeof event.runErrorCode === "string" && /^[a-zA-Z][a-zA-Z0-9_]{1,99}$/.test(event.runErrorCode) + ? event.runErrorCode : undefined; // Without a usable root code, unrelated failures that share the error code, // adapter and phase would collapse into one group. Fall back to the root // exception's class name plus its HTTP status or numeric code. These are @@ -337,6 +341,7 @@ function captureRunFailureEvent(event: RunFailureEvent): string | null { event_kind: "run_failure", ...(event.failureSource ? { failure_source: event.failureSource } : {}), ...(rootErrorCode ? { root_error_code: rootErrorCode } : {}), + ...(runErrorCode ? { run_error_code: runErrorCode } : {}), ...(failurePhase ? { failure_phase: failurePhase } : {}), run_id: event.runId, ...(event.taskId ? { task_id: event.taskId } : {}), @@ -393,7 +398,7 @@ function confirmTaskInterruptionDelivery(eventId: string | undefined, statusCode } export function captureTaskInterruptionFailure(event: Pick & { intentId: string | null }): void { + "companyId" | "agentId" | "taskId" | "runId" | "runStatus" | "agentAdapter" | "runErrorCode"> & { intentId: string | null }): void { const key = JSON.stringify([event.companyId, event.taskId, event.runId, event.intentId]); if (reportedTaskInterruptions.has(key)) return; const now = Date.now(); @@ -411,6 +416,7 @@ export function captureTaskInterruptionFailure(event: Pick & Partial> & { + | "scheduledRetryReason" | "createdAt" | "finishedAt">> & { /** Due time of the native coordinator's own same-run resume, if one is pending. */ nativeResumeAt?: Date | null; + /** When the native coordinator saved this run's terminal failure, if it did. */ + nativeFailedAt?: Date | null; }; type ProjectionWait = { id: string; requestedAt: Date; sourceRunId?: string | null; executionWaitReason?: string | null; - preparationStatus?: string | null; preparationRetryAt?: number | string | null; + preparationStatus?: string | null; preparationRetryAt?: number | string | null; preparationUpdatedAt?: Date | null; /** When the user pressed Interrupt to deliver this queued message now. */ queuedInterruptRequestedAt?: string | null }; type ProjectionState = { @@ -65,6 +67,29 @@ type ProjectionState = { admissionWait?: { id: string; status: string; reason: string | null; agentId: string }; }; +// A task read reports the change that made a task fail, not every later read of +// retained history. The delivery cache is process-local, so a restart or another +// replica would otherwise resend each old failure whenever a list shows it. +const TASK_INTERRUPTION_REPORT_WINDOW_MS = 60 * 60_000; + +/** Latest saved fact that can move a task into the failed state, if known. The + * recovery row's update time covers recovery that newly fails on an old run: + * each saved recovery transition stamps it, while activity heartbeats do not. + * The native resume due time and terminal finalization cover a same-run resume + * that lapses or is exhausted on an old run. */ +export function taskInterruptionChangedAt(input: Pick): number | null { + const { run, wait, action } = input; + const retryWait = input.retryWait ?? wait; + const preparationFailedAt = retryWait?.preparationStatus === "failed" || retryWait?.preparationStatus === "cancelled" + ? retryWait.preparationUpdatedAt : null; + const times: unknown[] = [run?.finishedAt ?? run?.createdAt, run?.scheduledRetryAt, run?.nativeResumeAt, run?.nativeFailedAt, + wait?.requestedAt, retryWait?.requestedAt, action?.updatedAt, preparationFailedAt].map(value => value?.getTime()); + times.push(readSystemRecovery(action?.evidence)?.nextAttemptAt, + retryWait?.preparationRetryAt == null ? null : Number(retryWait.preparationRetryAt)); + const known = times.filter((time): time is number => typeof time === "number" && Number.isFinite(time)); + return known.length ? Math.max(...known) : null; +} + function boundedJsonString(value: SQL, limit: number) { return sql`case when jsonb_typeof(${value}) = 'string' then left(${value} #>> '{}', ${limit}) else null end`; } @@ -256,7 +281,7 @@ export async function taskInterruptionPresentations(db: Db, companyId: string, i const startingIssueIds = selected.filter(issue => issue.status === "todo" && issue.assigneeAgentId).map(issue => issue.id); const [runs, waits, agentRows, projectRows, pauseRows, approvalRows, companyRows, admissionWaits] = await Promise.all([ db.select({ issueId: requestedIssueId, id: heartbeatRuns.id, status: heartbeatRuns.status, - agentId: heartbeatRuns.agentId, companyId: heartbeatRuns.companyId, + createdAt: heartbeatRuns.createdAt, finishedAt: heartbeatRuns.finishedAt, agentId: heartbeatRuns.agentId, companyId: heartbeatRuns.companyId, nativeIssueId: heartbeatRuns.nativeIssueId, runtimeMode: heartbeatRuns.runtimeMode, startedAt: heartbeatRuns.startedAt, agentAdapter: agents.adapterType, error: sql`case when ${heartbeatRuns.errorCode} = 'setup_failed' @@ -273,6 +298,9 @@ export async function taskInterruptionPresentations(db: Db, companyId: string, i and ${nativeRunFinalizations.resultId} is null and ${nativeRunFinalizations.phase} = 'retryable_failure' and ${heartbeatRuns.contextSnapshot}->'nativeQuestionCancellation' is null and ${heartbeatRuns.runnerProfileJson}->'legacyWorkspaceRecoveryMigration' is null)`.mapWith(value => value == null ? null : new Date(value)), + nativeFailedAt: sql`(select ${nativeRunFinalizations.updatedAt} from ${nativeRunFinalizations} + where ${nativeRunFinalizations.companyId} = ${heartbeatRuns.companyId} and ${nativeRunFinalizations.runId} = ${heartbeatRuns.id} + and ${nativeRunFinalizations.phase} = 'terminal_failure')`.mapWith(value => value == null ? null : new Date(value)), // The run whose resultJson supplied providerFailure. A scheduled retry // shows its predecessor's reason, so redaction must use that run's registry. providerFailureRunId: sql`case when ${heartbeatRuns.resultJson}->'providerFailure' is null @@ -308,7 +336,8 @@ export async function taskInterruptionPresentations(db: Db, companyId: string, i db.selectDistinctOn([wakeIssue, isRetryWait], { issueId: wakeIssue, id: agentWakeupRequests.id, requestedAt: agentWakeupRequests.requestedAt, sourceRunId: boundedJsonString(sql`${agentWakeupRequests.payload}#>'{taskRetry,sourceRunId}'`, 160), executionWaitReason: boundedJsonString(sql`${agentWakeupRequests.payload}#>'{executionWait,reason}'`, 80), - preparationStatus: chatActions.status, preparationRetryAt: chatActionDeadline("failed_run_retry") }) + preparationStatus: chatActions.status, preparationRetryAt: chatActionDeadline("failed_run_retry"), + preparationUpdatedAt: chatActions.updatedAt }) .from(agentWakeupRequests).leftJoin(chatActions, and(eq(chatActions.companyId, agentWakeupRequests.companyId), eq(chatActions.id, canonicalExecutionUuid(sql`${agentWakeupRequests.payload}#>>'{taskRetry,chatActionId}'`)), eq(chatActions.kind, "failed_run_retry"), sql`${chatActions.payload}->>'taskRetryIntentId' = ${agentWakeupRequests.id}::text`, @@ -440,26 +469,30 @@ export async function taskInterruptionPresentations(db: Db, companyId: string, i const projectMap = new Map(projectRows.map(project => [project.id, project])); const pauses = new Set(pauseRows.map(row => row.issueId)), approvalsSet = new Set(approvalRows.map(row => row.issueId)); const admissionWaitMap = byIssue(admissionWaits); + const now = Date.now(); for (const issue of selected) { const required = requirements.get(issue.id); const denied = required?.projectIds.filter(projectId => currentAccess.get(`${issue.assigneeAgentId}:${projectId}`) !== true); - const presentation = projectTaskInterruption(issue, { - run: runMap.get(issue.id), action: actionMap.get(issue.id), wait: waitMap.get(issue.id), retryWait: retryWaitMap.get(issue.id), + const facts = { run: runMap.get(issue.id), action: actionMap.get(issue.id), wait: waitMap.get(issue.id), + retryWait: retryWaitMap.get(issue.id) }; + const presentation = projectTaskInterruption(issue, { ...facts, interruptWait: interruptWaitFor(issue.id), projectAccessAllowed: denied?.length === 0, projectAccessProjectId: denied?.includes(issue.projectId ?? "") ? issue.projectId : null, workspaceSelectionAvailable: required?.available ?? false, agent: agentMap.get(issue.assigneeAgentId ?? ""), project: projectMap.get(issue.projectId ?? ""), company: companyRows[0], paused: pauses.has(issue.id), approval: approvalsSet.has(issue.id), - ownerStopRequired: ownerStopRequired.has(`${issue.id}:${runMap.get(issue.id)?.id}`), + ownerStopRequired: ownerStopRequired.has(`${issue.id}:${facts.run?.id}`), admissionWait: admissionWaitMap.get(issue.id), - }); + }, now); result.set(issue.id, presentation); - const run = runMap.get(issue.id); + const run = facts.run; const runStatus = HEARTBEAT_RUN_STATUSES.find(status => status === run?.status); - if (presentation?.responsibility === "system" && presentation.state === "failed" && run && runStatus) { + const changedAt = taskInterruptionChangedAt(facts); + if (presentation?.responsibility === "system" && presentation.state === "failed" && run && runStatus + && (changedAt === null || changedAt > now - TASK_INTERRUPTION_REPORT_WINDOW_MS)) { captureTaskInterruptionFailure({ companyId, taskId: issue.id, agentId: run.agentId, - runId: run.id, runStatus, intentId: presentation.intentId, + runId: run.id, runStatus, intentId: presentation.intentId, runErrorCode: run.errorCode ?? null, agentAdapter: run.agentAdapter ?? "unknown" }); } }