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" }); } }