Surface provider/error on thread.failed instead of a generic workflow failure

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

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;

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,

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,

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();