diff --git a/doc/observability.md b/doc/observability.md index 996d3a807f5..a96ae96058b 100644 --- a/doc/observability.md +++ b/doc/observability.md @@ -973,8 +973,8 @@ backend Sentry DSN and the audited SDK. Set `PAPERCLIP_SENTRY_RUN_FAILURES` to: | `all` | Every failed or timed-out run, including the known configuration, quota, and external connection failures. Unexpected started cancellations also report. | Missing, blank, or unrecognized values keep run reporting off. Expected operator -stops, shutdown interruptions, successful runs, and queued work do not report in -either enabled mode. A native run that is `failed` while it waits for a same-run +stops, shutdown interruptions, successful runs, and queued work do not report +terminal run failures in either enabled mode. A native run that is `failed` while it waits for a same-run retry (`nativePhase` `retryable_failure`) does not report either. It reports once if a later write ends the retry as a terminal failure. The mode does not change run status, retries, recovery, Telemetry, or OpenTelemetry export. It is process-wide, so an operator can enable @@ -996,6 +996,19 @@ They do not change the ambient Sentry scope, whose isolation is unavailable without an OpenTelemetry context manager. Later, unrelated exceptions must not inherit a previous run's identity or fingerprint. +When a task read produces the system-owned failed state displayed as +“Couldn’t finish this attempt. Your message is saved.”, either enabled mode also +reports a `task_attempt_failed` error with `event_kind=run_failure` and +`failure_source=task_interruption`. This catches saved-message failures without +a new terminal run transition. The event keeps the actual observed run status, +the adapter and the company, agent, task and run references. It uses fixed error +text and includes no message bodies, prompts or raw run diagnostics. Working, +scheduled-retry, resolved and user-action states do not produce this event. +Repeated reads of the same company/task/run/intent are suppressed by a bounded +process-local cache of the latest 1,000 reports. A server restart or cache eviction +can report an old failure again. This task-state event can accompany a separate +terminal run-failure report; the `failure_source` tag distinguishes them. + The versioned `paperclip` context supplies company and agent IDs and relative API paths for the saved run, events, log, and optional task. A valid HTTP(S) origin in `PAPERCLIP_PUBLIC_URL` also supplies absolute run, events, and log URLs. diff --git a/server/src/__tests__/run-failure-sentry.test.ts b/server/src/__tests__/run-failure-sentry.test.ts index 8d841c98af7..85de34731a1 100644 --- a/server/src/__tests__/run-failure-sentry.test.ts +++ b/server/src/__tests__/run-failure-sentry.test.ts @@ -91,6 +91,42 @@ afterEach(() => { }); describe("captureRunFailure", () => { + it("reports a saved-message failure with the run-failure tag and the actual run status", async () => { + const { sentryModule, captureException } = await importFreshSentryWithGateOpen(); + const event = { companyId: "company", agentId: "agent", taskId: "task", runId: "run", + agentAdapter: "process", runStatus: "succeeded" as const, intentId: "intent" }; + sentryModule.captureTaskInterruptionFailure(event); + sentryModule.captureTaskInterruptionFailure(event); + expect(captureException).toHaveBeenCalledTimes(1); + expect(captureException.mock.calls[0]![0]).toMatchObject({ + message: "Couldn’t finish this attempt. Your message is saved.", + }); + expect(captureException.mock.calls[0]![1]).toMatchObject({ + tags: { event_kind: "run_failure", failure_source: "task_interruption", run_id: "run", task_id: "task", + company_id: "company", agent_id: "agent", run_status: "succeeded", error_code: "task_attempt_failed" }, + contexts: { paperclip: { runPath: "/api/heartbeat-runs/run", taskPath: "/api/issues/task" } }, + fingerprint: ["task_attempt_failed", "process"], + }); + sentryModule.captureTaskInterruptionFailure({ ...event, intentId: "next-intent" }); + sentryModule.captureTaskInterruptionFailure({ ...event, runId: "next-run" }); + expect(captureException).toHaveBeenCalledTimes(3); + }); + + it("does not consume a saved-message report while reporting is off or capture fails", async () => { + const { sentryModule, captureException } = await importFreshSentryWithGateOpen(); + const event = { companyId: "company", agentId: "agent", taskId: "task", runId: "run", + agentAdapter: "process", runStatus: "failed" as const, intentId: null }; + vi.stubEnv("PAPERCLIP_SENTRY_RUN_FAILURES", "off"); + sentryModule.captureTaskInterruptionFailure(event); + expect(captureException).not.toHaveBeenCalled(); + vi.stubEnv("PAPERCLIP_SENTRY_RUN_FAILURES", "unexpected"); + captureException.mockImplementationOnce(() => { throw new Error("Sentry unavailable"); }); + expect(() => sentryModule.captureTaskInterruptionFailure(event)).not.toThrow(); + sentryModule.captureTaskInterruptionFailure(event); + sentryModule.captureTaskInterruptionFailure(event); + expect(captureException).toHaveBeenCalledTimes(2); + }); + it.each([undefined, "off", "true", "invalid"])("keeps ordinary server errors enabled while run reporting is disabled: %s", async mode => { const { sentryModule, captureException } = await importFreshSentryWithGateOpen(); vi.stubEnv("PAPERCLIP_SENTRY_RUN_FAILURES", mode); diff --git a/server/src/__tests__/task-interruption-query.test.ts b/server/src/__tests__/task-interruption-query.test.ts index 69e06c4dde0..dd357024bfd 100644 --- a/server/src/__tests__/task-interruption-query.test.ts +++ b/server/src/__tests__/task-interruption-query.test.ts @@ -1,10 +1,14 @@ import { randomUUID } from "node:crypto"; -import { afterAll, beforeAll, describe, expect, it } from "vitest"; -import { agents, companies, createDb, heartbeatRuns, issues, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS } from "@paperclipai/db"; +import { eq } from "drizzle-orm"; +import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; +import { agents, agentWakeupRequests, companies, createDb, heartbeatRuns, issues, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS } from "@paperclipai/db"; import { taskInterruptionPresentations } from "../services/task-interruption-presentation.js"; import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; import { captureInterruptionQuery, seedInterruptionFixture } from "./helpers/task-interruption-fixture.js"; +const captureTaskInterruptionFailure = vi.hoisted(() => vi.fn()); +vi.mock("../sentry.js", () => ({ captureTaskInterruptionFailure })); + describe("task interruption latest run query", () => { let database: Awaited>; let db: ReturnType; @@ -26,6 +30,10 @@ describe("task interruption latest run query", () => { expect(Math.max(...inspected)).toBeLessThan(30); 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({ + companyId: f.companyId, taskId: issue.id, runId: presentations.get(issue.id)!.observedRunId, + runStatus: "failed", agentAdapter: "process", + })); }); it("preserves native, first-class and text legacy precedence, ties, and tenant scope", async () => { @@ -63,4 +71,56 @@ describe("task interruption latest run query", () => { expect((await taskInterruptionPresentations(db, companyId, [upper])).get(upper.id)).toBeNull(); expect(await taskInterruptionPresentations(db, companyId, [])).toEqual(new Map()); }); + + it("reports the saved-message failure state without reporting work, retries or user actions", async () => { + const companyId = randomUUID(), agentId = randomUUID(), pausedAgentId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Failure states", issuePrefix: `F${companyId.slice(0, 7)}` }); + await db.insert(agents).values([ + { id: agentId, companyId, name: "Worker", adapterType: "process" }, + { id: pausedAgentId, companyId, name: "Paused worker", adapterType: "process", status: "paused" }, + ]); + const input = await db.insert(issues).values(Array.from({ length: 9 }, (_, n) => ({ + companyId, title: `State ${n}`, status: n === 5 ? "done" : "in_progress", + assigneeAgentId: n === 8 ? pausedAgentId : agentId, + }))).returning(); + const runs = await db.insert(heartbeatRuns).values(input.map((issue, n) => ({ + companyId, agentId: issue.assigneeAgentId!, issueId: issue.id, + status: n === 1 || n === 4 ? "succeeded" : n === 2 ? "running" : n === 7 ? "cancelled" : "failed", + errorCode: n === 6 ? "native_provider_approval_required" : null, + scheduledRetryAt: n === 3 ? new Date(Date.now() + 60_000) : null, + }))).returning(); + 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, { ...input[0]!, companyId: randomUUID() }]); + expect(captureTaskInterruptionFailure).toHaveBeenCalledTimes(3); + for (const n of [0, 1, 7]) { + 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, + }); + } + expect(result.get(input[2]!.id)?.state).toBe("working"); + expect(result.get(input[3]!.id)?.state).toBe("retry_scheduled"); + for (const n of [4, 5]) expect(result.get(input[n]!.id)?.state).toBe("resolved"); + for (const n of [6, 8]) expect(result.get(input[n]!.id)?.responsibility).toBe("user"); + }); + + 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({ + companyId: f.companyId, name: "Next worker", adapterType: "codex_local", + }).returning(); + const [reassigned] = await db.update(issues).set({ assigneeAgentId: nextAgent!.id }) + .where(eq(issues.id, f.input[0]!.id)).returning(); + captureTaskInterruptionFailure.mockClear(); + const result = await taskInterruptionPresentations(db, f.companyId, [reassigned!]); + expect(result.get(reassigned!.id)?.state).toBe("failed"); + expect(captureTaskInterruptionFailure).toHaveBeenCalledExactlyOnceWith(expect.objectContaining({ + taskId: reassigned!.id, agentAdapter: "process", + })); + }); }); diff --git a/server/src/sentry.ts b/server/src/sentry.ts index 1808cc9291c..991023cc81d 100644 --- a/server/src/sentry.ts +++ b/server/src/sentry.ts @@ -56,6 +56,7 @@ // `instrumentation.ts`. import os from "node:os"; +import type { HeartbeatRunStatus } from "@paperclipai/shared"; import { AdapterStopTimeoutError } from "./services/adapter-stop-timeout.js"; import { CloudPortfolioError } from "./services/cloud-portfolio-error.js"; import { storageFailureDiagnosticsFor } from "./storage/failure-diagnostics.js"; @@ -173,7 +174,8 @@ export interface RunFailureEvent { /** The agent's adapter type, or `"unknown"` when the agent row is absent. */ agentAdapter: string; /** The run status that triggered this report. */ - runStatus: RunFailureStatus; + runStatus: HeartbeatRunStatus; + failureSource?: "task_interruption"; /** Bounded, redacted diagnostics selected by the run failure reporter. */ diagnostics?: RunFailureDiagnostics; /** Recorded process exit evidence, when available. Validated before capture. */ @@ -206,8 +208,8 @@ function normalizeRunSignal(value: unknown): string | null { * change run control flow. * */ -export function captureRunFailure(event: RunFailureEvent): void { - if (!sentryHandle || readRunFailureReportingMode() === "off") return; +export function captureRunFailure(event: RunFailureEvent): boolean { + if (!sentryHandle || readRunFailureReportingMode() === "off") return false; const handle = sentryHandle; try { const errorCode = event.errorCode ?? "unknown"; @@ -270,6 +272,7 @@ export function captureRunFailure(event: RunFailureEvent): void { handle.captureException(exception, { tags: { event_kind: "run_failure", + ...(event.failureSource ? { failure_source: event.failureSource } : {}), ...(rootErrorCode ? { root_error_code: rootErrorCode } : {}), ...(failurePhase ? { failure_phase: failurePhase } : {}), run_id: event.runId, @@ -294,9 +297,35 @@ export function captureRunFailure(event: RunFailureEvent): void { }, fingerprint: [errorCode, event.agentAdapter, ...(failurePhase ? [failurePhase] : []), ...(rootErrorCode ? [rootErrorCode] : [])], }); + return true; } catch (err) { // eslint-disable-next-line no-console console.error("[paperclip] Sentry captureRunFailure failed", err); + return false; + } +} + +const reportedTaskInterruptions = new Set(); +const MAX_REPORTED_TASK_INTERRUPTIONS = 1_000; + +export function captureTaskInterruptionFailure(event: Pick & { intentId: string | null }): void { + const key = JSON.stringify([event.companyId, event.taskId, event.runId, event.intentId]); + if (reportedTaskInterruptions.has(key)) return; + if (!captureRunFailure({ + companyId: event.companyId, + agentId: event.agentId, + taskId: event.taskId, + runId: event.runId, + runStatus: event.runStatus, + agentAdapter: event.agentAdapter, + errorMessage: "Couldn’t finish this attempt. Your message is saved.", + errorCode: "task_attempt_failed", + failureSource: "task_interruption", + })) return; + reportedTaskInterruptions.add(key); + if (reportedTaskInterruptions.size > MAX_REPORTED_TASK_INTERRUPTIONS) { + reportedTaskInterruptions.delete(reportedTaskInterruptions.values().next().value!); } } diff --git a/server/src/services/task-interruption-presentation.ts b/server/src/services/task-interruption-presentation.ts index a614a066baf..8f2820fee54 100644 --- a/server/src/services/task-interruption-presentation.ts +++ b/server/src/services/task-interruption-presentation.ts @@ -2,7 +2,8 @@ import { and, desc, eq, inArray, isNull, sql } from "drizzle-orm"; import { agents, agentWakeupRequests, approvals, companies, heartbeatRuns, issueApprovals, issueRecoveryActions, issues, issueTreeHoldMembers, issueTreeHolds, projects, type Db } from "@paperclipai/db"; import { unionAll } from "drizzle-orm/pg-core"; -import type { TaskInterruptionPresentation } from "@paperclipai/shared"; +import { HEARTBEAT_RUN_STATUSES, type TaskInterruptionPresentation } from "@paperclipai/shared"; +import { captureTaskInterruptionFailure } from "../sentry.js"; import { isAiAuthenticationBlocked } from "./ai-auth-failure.js"; import { readSystemRecovery } from "./system-recovery-state.js"; import { PROJECT_ACCESS_DENIED } from "./project-access-denial.js"; @@ -104,13 +105,14 @@ export async function taskInterruptionPresentations(db: Db, companyId: string, i const projectIds = selected.flatMap(issue => issue.projectId ? [issue.projectId] : []); const [runs, waits, agentRows, projectRows, pauseRows, approvalRows, companyRows] = await Promise.all([ db.select({ issueId: requestedIssueId, id: heartbeatRuns.id, status: heartbeatRuns.status, - agentId: heartbeatRuns.agentId, + agentId: heartbeatRuns.agentId, agentAdapter: agents.adapterType, error: sql`case when ${heartbeatRuns.errorCode} = 'setup_failed' and ${heartbeatRuns.error} = 'Project is outside this low-trust boundary.' then ${heartbeatRuns.error} else null end`, errorCode: heartbeatRuns.errorCode, livenessState: heartbeatRuns.livenessState, scheduledRetryAt: heartbeatRuns.scheduledRetryAt, resultJson: sql>`jsonb_build_object('configurationIncomplete', ${heartbeatRuns.resultJson}->'configurationIncomplete', 'projectAccessDenied', ${heartbeatRuns.resultJson}->'projectAccessDenied')`, }).from(requested).innerJoinLateral(latestRuns, sql`true`).innerJoin(heartbeatRuns, and(eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.id, latestRuns.id))) + .leftJoin(agents, and(eq(agents.companyId, heartbeatRuns.companyId), eq(agents.id, heartbeatRuns.agentId))) .orderBy(requestedIssueId), db.selectDistinctOn([wakeIssue], { issueId: wakeIssue, id: agentWakeupRequests.id, requestedAt: agentWakeupRequests.requestedAt }) .from(agentWakeupRequests).where(and(eq(agentWakeupRequests.companyId, companyId), inArray(wakeIssue, ids), @@ -168,14 +170,22 @@ export async function taskInterruptionPresentations(db: Db, companyId: string, i for (const issue of selected) { const required = requirements.get(issue.id); const denied = required?.projectIds.filter(projectId => currentAccess.get(`${issue.assigneeAgentId}:${projectId}`) !== true); - result.set(issue.id, projectTaskInterruption(issue, { + const presentation = projectTaskInterruption(issue, { run: runMap.get(issue.id), action: actionMap.get(issue.id), wait: waitMap.get(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), - })); + }); + result.set(issue.id, presentation); + const run = runMap.get(issue.id); + const runStatus = HEARTBEAT_RUN_STATUSES.find(status => status === run?.status); + if (presentation?.responsibility === "system" && presentation.state === "failed" && run && runStatus) { + captureTaskInterruptionFailure({ companyId, taskId: issue.id, agentId: run.agentId, + runId: run.id, runStatus, intentId: presentation.intentId, + agentAdapter: run.agentAdapter ?? "unknown" }); + } } return result; }