From b39c54069e7f8601002bfc119df9393a364266f5 Mon Sep 17 00:00:00 2001 From: Michael Isaac <217397473+MPIsaac-Per@users.noreply.github.com> Date: Thu, 20 Aug 2026 16:43:20 -0500 Subject: [PATCH] Surface provider/error on thread.failed instead of a generic workflow failure (#1914) Provider failures persist as provider/error. thread.failed only read system/error, so workflows fell back to "Workflow worker failed" and hid 429s. Look up the newest of both types and format provider detail with category, HTTP status, and provider code. --- .../src/services/threads/thread-data.ts | 40 +++++- .../plugins/plugin-thread-events.test.ts | 121 ++++++++++++++++++ packages/db/src/data/events.ts | 29 +++++ packages/db/src/data/index.ts | 2 + packages/db/test/data/events.test.ts | 48 +++++++ 5 files changed, 236 insertions(+), 4 deletions(-) diff --git a/apps/server/src/services/threads/thread-data.ts b/apps/server/src/services/threads/thread-data.ts index 9cc53bc571..6e4eb23a11 100644 --- a/apps/server/src/services/threads/thread-data.ts +++ b/apps/server/src/services/threads/thread-data.ts @@ -1,13 +1,14 @@ import { findStoredEventRow as findStoredEventRowRecord, + getLatestThreadErrorEventRow, getLatestThreadOutputEventRow, - getLatestThreadSystemErrorEventRow, listStoredEventRows as listStoredEventRowRecords, } from "@bb/db"; import type { DbConnection, StoredEventRow } from "@bb/db"; import { buildThreadEventRow, parseStoredThreadEvent } from "@bb/domain"; import { threadScope, turnScope } from "@bb/domain"; import type { + StoredThreadEventDataForType, ThreadEvent, ThreadEventRow, ThreadEventScope, @@ -164,13 +165,44 @@ export function getLastThreadOutput( return null; } -/** Latest system/error message for a thread, or null when none exists. */ +/** + * Newest system/error or provider/error message for a thread, or null. + * Provider failures persist as provider/error, not system/error. + */ export function getLastThreadErrorMessage( db: DbConnection, threadId: string, ): string | null { - const row = getLatestThreadSystemErrorEventRow(db, { threadId }); + const row = getLatestThreadErrorEventRow(db, { threadId }); if (!row) return null; const eventRow = parseStoredEventRow(row); - return eventRow.type === "system/error" ? eventRow.data.message : null; + if (eventRow.type === "system/error") { + return eventRow.data.message; + } + if (eventRow.type === "provider/error") { + return formatProviderError(eventRow.data); + } + return null; +} + +/** Detail (else message), then category, HTTP status, and provider code. */ +function formatProviderError( + data: StoredThreadEventDataForType<"provider/error">, +): string { + const base = + data.detail !== undefined && data.detail.length > 0 + ? data.detail + : data.message; + const info = data.errorInfo; + if (info === undefined) { + return base; + } + const parts = [base, `category: ${info.category}`]; + if (info.httpStatusCode !== null) { + parts.push(`http: ${info.httpStatusCode}`); + } + if (info.providerCode !== null && info.providerCode.length > 0) { + parts.push(`providerCode: ${info.providerCode}`); + } + return parts.join(" | "); } diff --git a/apps/server/test/services/plugins/plugin-thread-events.test.ts b/apps/server/test/services/plugins/plugin-thread-events.test.ts index a0ea86e57a..701597afc7 100644 --- a/apps/server/test/services/plugins/plugin-thread-events.test.ts +++ b/apps/server/test/services/plugins/plugin-thread-events.test.ts @@ -197,6 +197,127 @@ describe("plugin thread lifecycle events", () => { } }); + it("delivers thread.failed with a provider/error detail and metadata", async () => { + const recorded: RecordedThreadPayload[] = []; + globals.__failedEvents = recorded; + const { harness, cleanup } = await setUpPluginHarness(` + export default function plugin(bb: any) { + bb.events.on("thread.failed", (payload: any) => { + (globalThis as any).__failedEvents.push(payload); + }); + } + `); + try { + const { environment, thread } = seedThreadFixture(harness, { + thread: { status: "active" }, + }); + seedEvent(harness.deps, { + threadId: thread.id, + environmentId: environment.id, + scope: threadScope(), + sequence: 1, + type: "provider/error", + providerThreadId: "prov-1", + data: { + threadId: thread.id, + providerThreadId: "prov-1", + message: "model errored", + detail: "429 rate limit: too many requests", + errorInfo: { + category: "rate-limit", + httpStatusCode: 429, + providerCode: "rate_limit_event", + }, + }, + }); + + const outcome = applyLoggedThreadLifecycleEvent(lifecycleDeps(harness), { + threadId: thread.id, + event: { type: "run.failed" }, + }); + expect(outcome.applied).toBe(true); + + await vi.waitFor(() => expect(recorded).toHaveLength(1)); + expect(recorded[0]?.thread.id).toBe(thread.id); + expect(recorded[0]?.thread.status).toBe("error"); + expect(recorded[0]?.error).toBe( + "429 rate limit: too many requests | category: rate-limit | http: 429 | providerCode: rate_limit_event", + ); + } finally { + delete globals.__failedEvents; + await cleanup(); + } + }); + + it("surfaces a terminal provider/error over an earlier willRetry provider/error", async () => { + const recorded: RecordedThreadPayload[] = []; + globals.__failedEvents = recorded; + const { harness, cleanup } = await setUpPluginHarness(` + export default function plugin(bb: any) { + bb.events.on("thread.failed", (payload: any) => { + (globalThis as any).__failedEvents.push(payload); + }); + } + `); + try { + const { environment, thread } = seedThreadFixture(harness, { + thread: { status: "active" }, + }); + seedEvent(harness.deps, { + threadId: thread.id, + environmentId: environment.id, + scope: threadScope(), + sequence: 1, + type: "provider/error", + providerThreadId: "prov-1", + data: { + threadId: thread.id, + providerThreadId: "prov-1", + message: "transient 429", + detail: "will retry", + willRetry: true, + errorInfo: { + category: "rate-limit", + httpStatusCode: 429, + providerCode: "rate_limit_event", + }, + }, + }); + seedEvent(harness.deps, { + threadId: thread.id, + environmentId: environment.id, + scope: threadScope(), + sequence: 2, + type: "provider/error", + providerThreadId: "prov-1", + data: { + threadId: thread.id, + providerThreadId: "prov-1", + message: "terminal failure after retries", + errorInfo: { + category: "internal", + httpStatusCode: 500, + providerCode: "internal_error", + }, + }, + }); + + const outcome = applyLoggedThreadLifecycleEvent(lifecycleDeps(harness), { + threadId: thread.id, + event: { type: "run.failed" }, + }); + expect(outcome.applied).toBe(true); + + await vi.waitFor(() => expect(recorded).toHaveLength(1)); + expect(recorded[0]?.error).toBe( + "terminal failure after retries | category: internal | http: 500 | providerCode: internal_error", + ); + } finally { + delete globals.__failedEvents; + await cleanup(); + } + }); + it("delivers thread.created from the thread creation seam", async () => { const recorded: RecordedThreadPayload[] = []; globals.__createdEvents = recorded; diff --git a/packages/db/src/data/events.ts b/packages/db/src/data/events.ts index fe9cab25db..e07c5aaf42 100644 --- a/packages/db/src/data/events.ts +++ b/packages/db/src/data/events.ts @@ -1183,6 +1183,10 @@ export interface GetLatestThreadSystemErrorEventRowArgs { threadId: string; } +export interface GetLatestThreadErrorEventRowArgs { + threadId: string; +} + export interface GetLatestThreadSequenceArgs { threadId: string; } @@ -3185,6 +3189,31 @@ export function getLatestThreadSystemErrorEventRow( ); } +/** + * Newest system/error or provider/error for one thread. + * Sequence descending: a later terminal failure outranks an earlier + * willRetry provider/error. + */ +export function getLatestThreadErrorEventRow( + db: DbConnection, + args: GetLatestThreadErrorEventRowArgs, +): StoredEventRow | null { + return ( + db + .select(storedEventRowFields) + .from(events) + .where( + and( + eq(events.threadId, args.threadId), + inArray(events.type, ["system/error", "provider/error"]), + ), + ) + .orderBy(desc(events.sequence)) + .limit(1) + .get() ?? null + ); +} + export function getLatestThreadSequence( db: DbConnection, args: GetLatestThreadSequenceArgs, diff --git a/packages/db/src/data/index.ts b/packages/db/src/data/index.ts index ded9bbfb5b..9f87801925 100644 --- a/packages/db/src/data/index.ts +++ b/packages/db/src/data/index.ts @@ -340,6 +340,7 @@ export { getLatestThreadOutputEventRow, getLatestStoredConversationOutlineSequence, getLatestThreadSystemErrorEventRow, + getLatestThreadErrorEventRow, getLatestThreadSequence, getLatestStoredEventRowByType, insertEvents, @@ -407,6 +408,7 @@ export type { GetStoredTurnRequestEventForTurnArgs, GetRootStoredTurnStartedSequenceArgs, GetLatestThreadInterruptedReasonArgs, + GetLatestThreadErrorEventRowArgs, GetLatestThreadSequenceArgs, HasStoredTurnStartedArgs, InsertEventInput, diff --git a/packages/db/test/data/events.test.ts b/packages/db/test/data/events.test.ts index 10a3e091ec..4f987c7780 100644 --- a/packages/db/test/data/events.test.ts +++ b/packages/db/test/data/events.test.ts @@ -25,6 +25,7 @@ import { getLastStoredProviderThreadId, getLastStoredTurnRequestEvent, getLatestThreadOutputEventRow, + getLatestThreadErrorEventRow, getLatestThreadSequence, getStoredTimelineWindowEventDataBytes, insertEvents, @@ -1111,6 +1112,53 @@ describe("events", () => { }); }); + it("returns the newest system/error or provider/error by sequence", () => { + const { db, thread } = setup(); + + insertEvents(db, noopNotifier, [ + { + threadId: thread.id, + sequence: 1, + type: "system/error", + ...threadEventFields, + data: JSON.stringify({ message: "older system" }), + }, + { + threadId: thread.id, + sequence: 2, + type: "provider/error", + ...threadEventFields, + providerThreadId: "prov-1", + data: JSON.stringify({ + message: "retryable", + willRetry: true, + }), + }, + { + threadId: thread.id, + sequence: 3, + type: "turn/completed", + ...createTurnEventFields({ turnId: "turn-1" }), + data: JSON.stringify({ status: "failed" }), + }, + { + threadId: thread.id, + sequence: 4, + type: "provider/error", + ...threadEventFields, + providerThreadId: "prov-1", + data: JSON.stringify({ message: "terminal" }), + }, + ]); + + expect( + getLatestThreadErrorEventRow(db, { threadId: thread.id }), + ).toMatchObject({ + sequence: 4, + type: "provider/error", + }); + }); + it("lists stored event rows by range and exclusion filters", () => { const { db, thread } = setup();