diff --git a/CHANGELOG.md b/CHANGELOG.md index 83428cc2..d5a3ef74 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,12 @@ # Changelog +## Unreleased + +- Preserve classified provider errors when streamed responses fail, including + aborts racing late message writes (#320). `DeltaStreamer.getOrCreateStreamId` + now accepts `{ ifAborted: "returnUndefined" }` for abort-aware writes; its + existing no-argument behavior is unchanged. + ## 0.7.1 - Persists oversized streamed files (#307) diff --git a/src/component/messages.test.ts b/src/component/messages.test.ts index 98037bb6..be00786f 100644 --- a/src/component/messages.test.ts +++ b/src/component/messages.test.ts @@ -263,9 +263,7 @@ describe("agent", () => { const { messages } = await t.mutation(api.messages.addMessages, { threadId: thread._id as Id<"threads">, order: "next", - messages: [ - { message: { role: "assistant", content: "separate reply" } }, - ], + messages: [{ message: { role: "assistant", content: "separate reply" } }], }); expect(messages[0]).toMatchObject({ order: 1, stepOrder: 0 }); @@ -1262,3 +1260,71 @@ describe("agent", () => { }); }); }); + +describe("late saves racing a failed pending message (issue #320)", () => { + const PROVIDER_ERROR = "invalid_prompt: Invalid prompt: flagged by policy."; + + test("keeps the first durable failure authoritative", async () => { + const t = initConvexTest(); + const thread = await t.mutation(api.threads.createThread, { + userId: "u1", + }); + const threadId = thread._id as Id<"threads">; + + const { messages: seeded } = await t.mutation(api.messages.addMessages, { + threadId, + messages: [ + { message: { role: "user", content: "hello" } }, + { message: { role: "assistant", content: [] }, status: "pending" }, + ], + }); + const pending = seeded.at(-1)!; + expect(pending.status).toBe("pending"); + + const streamId = await t.mutation(api.streams.create, { + threadId, + order: pending.order, + stepOrder: pending.stepOrder, + format: "UIMessageChunk", + }); + + await t.mutation(api.messages.finalizeMessage, { + messageId: pending._id as Id<"messages">, + result: { status: "failed", error: PROVIDER_ERROR }, + }); + await t.mutation(api.streams.abort, { streamId, reason: PROVIDER_ERROR }); + + const { messages: late } = await t.mutation(api.messages.addMessages, { + threadId, + pendingMessageId: pending._id as Id<"messages">, + finishStreamId: streamId, + failPendingSteps: false, + messages: [ + { message: { role: "assistant", content: "partial response" } }, + ], + }); + + const assistants = ( + await t.run(async (ctx) => + ctx.db + .query("messages") + .withIndex("threadId_status_tool_order_stepOrder", (q) => + q.eq("threadId", threadId), + ) + .collect(), + ) + ).filter((message) => message.message?.role === "assistant"); + + expect(late).toHaveLength(1); + expect(assistants).toHaveLength(1); + expect(assistants[0]!._id).toBe(pending._id); + expect(assistants[0]!.status).toBe("failed"); + expect(assistants[0]!.error).toBe(PROVIDER_ERROR); + expect(assistants[0]!.text).toBe("partial response"); + + const stream = await t.run((ctx) => + ctx.db.get("streamingMessages", streamId), + ); + expect(stream?.state.kind).toBe("aborted"); + }); +}); diff --git a/src/component/messages.ts b/src/component/messages.ts index 6b9aced3..e1e0daa0 100644 --- a/src/component/messages.ts +++ b/src/component/messages.ts @@ -327,8 +327,8 @@ async function addMessagesHandler( if (pendingMessage.status === "failed") { fail = true; error = - `Trying to update a message that failed: ${pendingMessageId}, ` + - `error: ${pendingMessage.error ?? error}`; + pendingMessage.error ?? + `Trying to update a message that failed: ${pendingMessageId}`; messageDoc.status = "failed"; messageDoc.error = error; } diff --git a/src/errors.test.ts b/src/errors.test.ts new file mode 100644 index 00000000..fbab8a8e --- /dev/null +++ b/src/errors.test.ts @@ -0,0 +1,76 @@ +import { APICallError } from "@ai-sdk/provider"; +import { describe, expect, test } from "vitest"; +import { errorToString } from "./errors.js"; + +describe("errorToString", () => { + test("preserves provider error classifications", () => { + const details = { + error: { + code: "invalid_prompt", + message: "Invalid prompt: flagged by policy", + }, + }; + const apiError = new APICallError({ + message: "Invalid prompt: flagged by policy", + url: "https://api.example.test", + requestBodyValues: {}, + statusCode: 400, + data: details, + }); + + expect(errorToString(details)).toBe( + "invalid_prompt: Invalid prompt: flagged by policy", + ); + expect(errorToString(apiError)).toBe( + "invalid_prompt: Invalid prompt: flagged by policy", + ); + expect(errorToString(new Error())).toBe("Error"); + expect(errorToString(new TypeError())).toBe("TypeError"); + const systemError = Object.assign(new Error("socket hang up"), { + code: "ECONNRESET", + }); + expect(errorToString(systemError)).toBe("socket hang up"); + const codeOnly = Object.assign(new Error("Request failed"), { + data: { code: "rate_limit" }, + }); + expect(errorToString(codeOnly)).toBe("rate_limit: Request failed"); + }); + + test("serializes objects without mistaking shared values for cycles", () => { + const shared = { detail: "provider disconnected" }; + const circular: Record = { shared }; + circular.self = circular; + + expect(errorToString({ x: shared, y: shared })).toBe( + '{"x":{"detail":"provider disconnected"},"y":{"detail":"provider disconnected"}}', + ); + expect(errorToString(circular)).toBe( + '{"shared":{"detail":"provider disconnected"},"self":"[Circular]"}', + ); + }); + + test("bounds stored error text without splitting surrogate pairs", () => { + const serialized = errorToString(`${"x".repeat(1022)}😀tail`); + + expect(serialized.length).toBeLessThanOrEqual(1024); + expect(serialized.endsWith("x…")).toBe(true); + }); + + test("does not throw when Error properties are hostile accessors", () => { + const error = new Error(); + Object.defineProperties(error, { + message: { + get() { + throw new Error("message getter failed"); + }, + }, + name: { + get() { + throw new Error("name getter failed"); + }, + }, + }); + + expect(errorToString(error)).toBe("Unknown error"); + }); +}); diff --git a/src/errors.ts b/src/errors.ts new file mode 100644 index 00000000..b94ca0dd --- /dev/null +++ b/src/errors.ts @@ -0,0 +1,113 @@ +const MAX_ERROR_LENGTH = 1024; + +export function errorToString(error: unknown): string { + return truncateError(describeError(error)); +} + +function describeError(error: unknown): string { + if (typeof error === "string") return error; + if (error instanceof Error) { + const message = property(error, "message"); + if (typeof message !== "string" || message.length === 0) { + const name = property(error, "name"); + return typeof name === "string" && name.length > 0 + ? name + : safeString(error); + } + const nested = errorDetails( + property(error, "error") ?? property(error, "data"), + ); + return ( + formatDetails({ + message: nested.message ?? message, + code: nested.code, + }) ?? message + ); + } + + const details = formatDetails(errorDetails(error)); + if (details) return details; + + if (error && typeof error === "object") { + try { + const ancestors: object[] = []; + const serialized = JSON.stringify(error, function (_key, value: unknown) { + if (typeof value === "bigint") return value.toString(); + if (!value || typeof value !== "object") return value; + while (ancestors.length > 0 && ancestors.at(-1) !== this) { + ancestors.pop(); + } + if (ancestors.includes(value)) return "[Circular]"; + ancestors.push(value); + return value; + }); + if (serialized) return serialized; + } catch { + return safeString(error); + } + } + + return safeString(error); +} + +function safeString(error: unknown): string { + try { + return String(error); + } catch { + return "Unknown error"; + } +} + +function errorDetails(error: unknown): { message?: string; code?: string } { + let current = error; + let message: string | undefined; + let code: string | undefined; + for (let depth = 0; depth < 3; depth++) { + if (typeof current === "string") { + message ??= current; + break; + } + if (!current || typeof current !== "object") break; + + const currentMessage = property(current, "message"); + if (typeof currentMessage === "string" && currentMessage.length > 0) { + message ??= currentMessage; + } + const currentCode = property(current, "code"); + if (typeof currentCode === "string" || typeof currentCode === "number") { + code ??= String(currentCode); + } + if (message && code) break; + current = property(current, "error") ?? property(current, "data"); + } + return { message, code }; +} + +function property(value: object, key: string): unknown { + try { + return (value as Record)[key]; + } catch { + return undefined; + } +} + +function formatDetails({ + message, + code, +}: { + message?: string; + code?: string; +}): string | undefined { + if (message && code) { + return message.startsWith(`${code}:`) ? message : `${code}: ${message}`; + } + return message ?? code; +} + +function truncateError(error: string): string { + if (error.length <= MAX_ERROR_LENGTH) return error; + let truncated = error.slice(0, MAX_ERROR_LENGTH - 1); + const last = truncated.charCodeAt(truncated.length - 1); + if (last >= 0xd800 && last <= 0xdbff) truncated = truncated.slice(0, -1); + return `${truncated}…`; +} diff --git a/src/vercel/client/streamText.test.ts b/src/vercel/client/streamText.test.ts index 4328011b..4a564218 100644 --- a/src/vercel/client/streamText.test.ts +++ b/src/vercel/client/streamText.test.ts @@ -11,13 +11,27 @@ import { import { v } from "convex/values"; import { components, initConvexTest } from "./setup.test.js"; import { mockModel } from "./mockModel.js"; -import { runAbortCleanup } from "./streamText.js"; +import { runStreamCleanup } from "./streamText.js"; +import { errorToString } from "./utils.js"; const schema = defineSchema({}); type DataModel = DataModelFromSchemaDefinition; const action = actionGeneric as ActionBuilder; const FINAL_TEXT = "Hello from the model"; +const PROVIDER_FAILURE_TEXT = "Mock provider failure"; +const CLEANUP_FAILURE_TEXT = "finalizeMessage rejected"; + +function hasKeys( + value: unknown, + keys: string[], +): value is Record { + return ( + value !== null && + typeof value === "object" && + keys.every((key) => key in value) + ); +} const agent = new Agent(components.agent, { name: "stream-test", @@ -34,6 +48,14 @@ const emptyAgent = new Agent(components.agent, { }), }); +const failingAgent = new Agent(components.agent, { + name: "failing-stream-test", + languageModel: mockModel({ + content: [{ type: "text", text: "partial response" }], + fail: { error: PROVIDER_FAILURE_TEXT }, + }), +}); + // Action that exercises streamText with saveStreamDeltas.returnImmediately=true. // It consumes the stream after streamText returns, simulating the HTTP response // path described in issue #265. @@ -91,11 +113,49 @@ export const streamTextEmptyReturnImmediately = action({ }, }); +export const streamTextCleanupFailure = action({ + args: { threadId: v.string() }, + handler: async (ctx, { threadId }) => { + const providerErrors: string[] = []; + let aborts = 0; + const failingCtx = { + ...ctx, + runMutation: (async (reference, args) => { + if (hasKeys(args, ["messageId", "result"])) { + throw new Error(CLEANUP_FAILURE_TEXT); + } + return ctx.runMutation(reference, args); + }) as typeof ctx.runMutation, + }; + let caught: string | undefined; + try { + await failingAgent.streamText( + failingCtx, + { threadId }, + { + prompt: "Test", + onError: ({ error }) => { + providerErrors.push(errorToString(error)); + }, + onAbort: () => { + aborts += 1; + }, + }, + { saveStreamDeltas: { chunking: "word", throttleMs: 0 } }, + ); + } catch (error) { + caught = errorToString(error); + } + return { providerErrors, aborts, caught }; + }, +}); + const testApi: ApiFromModules<{ fns: { streamTextReturnImmediately: typeof streamTextReturnImmediately; streamTextEmptyAwaited: typeof streamTextEmptyAwaited; streamTextEmptyReturnImmediately: typeof streamTextEmptyReturnImmediately; + streamTextCleanupFailure: typeof streamTextCleanupFailure; }; }>["fns"] = anyApi["streamText.test"] as any; @@ -147,28 +207,64 @@ describe("streamText with saveStreamDeltas.returnImmediately (issue #265)", () = }); describe("streamText abort cleanup", () => { - test("attempts every cleanup and rethrows the first internal failure", async () => { + test("finishes durable cleanup before invoking onAbort", async () => { const calls: string[] = []; - const firstFailure = new Error("failed pending message cleanup"); + let resolveStreamer!: () => void; + const syncFailure = new Error("synchronous pending message cleanup"); - await expect( - runAbortCleanup({ - failCall: async () => { - calls.push("call.fail"); - throw firstFailure; - }, - failStreamer: async () => { + const cleanup = runStreamCleanup({ + failCall: () => { + calls.push("call.fail"); + throw syncFailure; + }, + failStreamer: () => + new Promise((resolve) => { calls.push("streamer.fail"); - throw new Error("failed stream cleanup"); - }, - onAbort: () => { - calls.push("user.onAbort"); - }, - }), - ).rejects.toBe(firstFailure); + resolveStreamer = resolve; + }), + onAbort: () => { + calls.push("user.onAbort"); + }, + }); + + await Promise.resolve(); + expect(calls).toEqual(["call.fail", "streamer.fail"]); + resolveStreamer(); + await expect(cleanup).rejects.toBe(syncFailure); expect(calls).toEqual(["call.fail", "streamer.fail", "user.onAbort"]); }); + + test("surfaces a cleanup failure without hiding the provider error", async () => { + const t = initConvexTest(schema); + const threadId = await t.run(async (ctx) => + createThread(ctx, components.agent, { userId: "u1" }), + ); + + const { providerErrors, aborts, caught } = await t.action( + testApi.streamTextCleanupFailure, + { threadId }, + ); + + expect(providerErrors).toEqual([PROVIDER_FAILURE_TEXT]); + expect(aborts).toBe(0); + expect(caught).toBe(CLEANUP_FAILURE_TEXT); + + const streaming = await t.run(async (ctx) => + ctx.runQuery(components.agent.streams.list, { + threadId, + statuses: ["streaming"], + }), + ); + const aborted = await t.run(async (ctx) => + ctx.runQuery(components.agent.streams.list, { + threadId, + statuses: ["aborted"], + }), + ); + expect(streaming).toHaveLength(0); + expect(aborted).toHaveLength(1); + }); }); describe("streamText with an empty final step (issue #274)", () => { diff --git a/src/vercel/client/streamText.ts b/src/vercel/client/streamText.ts index af2bc15c..f21a13ee 100644 --- a/src/vercel/client/streamText.ts +++ b/src/vercel/client/streamText.ts @@ -27,19 +27,22 @@ import { getModelName, getProviderName } from "../../shared.js"; import { errorToString, willContinue } from "./utils.js"; import { materializeUIMessageChunkFiles } from "../fileMaterialization.js"; -/** Finish every abort cleanup path before surfacing an internal failure. */ -export async function runAbortCleanup(cleanup: { +export async function runStreamCleanup(cleanup: { failCall: () => Promise; failStreamer: () => Promise; onAbort?: () => PromiseLike | void; }): Promise { const results = await Promise.allSettled([ - cleanup.failCall(), - cleanup.failStreamer(), + Promise.resolve().then(() => cleanup.failCall()), + Promise.resolve().then(() => cleanup.failStreamer()), ]); - await cleanup.onAbort?.(); - const failure = results.find((result) => result.status === "rejected"); - if (failure) throw failure.reason; + const [abortResult] = await Promise.allSettled([ + Promise.resolve().then(() => cleanup.onAbort?.()), + ]); + const failure = [...results, abortResult].find( + (result) => result.status === "rejected", + ); + if (failure?.status === "rejected") throw failure.reason; } /** @@ -103,15 +106,11 @@ export async function streamText< Tools, object, RUNTIME_CONTEXT - >( - ctx, - component, - streamTextArgs, - options, - "streamText", - ); + >(ctx, component, streamTextArgs, options, "streamText"); const steps: StepResult[] = []; + let firstStreamError: string | undefined; + let streamCleanupFailure: { error: unknown } | undefined; let initialResponseMessages: ModelMessage[] = []; let initialResponseMessagesSaved = false; const responseMessagesForStep = ( @@ -180,15 +179,26 @@ export async function streamText< ), onError: async (error) => { console.error("onError", error); - await call.fail(errorToString(error.error)); - await streamer?.fail(errorToString(error.error)); + const reason = (firstStreamError ??= errorToString(error.error)); + try { + await runStreamCleanup({ + failCall: () => call.fail(reason), + failStreamer: async () => streamer?.fail(reason), + }); + } catch (cleanupError) { + streamCleanupFailure ??= { error: cleanupError }; + console.error("Failed to clean up errored stream:", cleanupError); + } return streamTextArgs.onError?.(error); }, onAbort: async (event) => { + const providerTriggeredAbort = + firstStreamError !== undefined && !args.abortSignal?.aborted; + if (providerTriggeredAbort) return; const reason = args.abortSignal?.reason ? errorToString(args.abortSignal.reason) : "streamText aborted"; - await runAbortCleanup({ + await runStreamCleanup({ failCall: () => call.fail(reason), failStreamer: async () => streamer?.fail(reason), onAbort: () => streamTextArgs.onAbort?.(event), @@ -227,13 +237,17 @@ export async function streamText< // returnImmediately path: streamText is about to return without // awaiting consumption, so the deferred-save block below won't // see this step. Save inline now (issue #265). - const finishStreamId = await streamer.getOrCreateStreamId(); - await call.save( - { step, responseMessages: responseMessagesForStep(step) }, - false, - finishStreamId, - ); - initialResponseMessagesSaved = true; + const finishStreamId = await streamer.getOrCreateStreamId({ + ifAborted: "returnUndefined", + }); + if (finishStreamId) { + await call.save( + { step, responseMessages: responseMessagesForStep(step) }, + false, + finishStreamId, + ); + initialResponseMessagesSaved = true; + } } } else { await call.save( @@ -258,8 +272,11 @@ export async function streamText< // If the stream errored (e.g. onStepFinish threw), the DeltaStreamer's // finish() was never called, leaving the streaming message stuck in // "streaming" state. Clean it up by marking it as aborted. - await streamer?.fail(e instanceof Error ? e.message : String(e)); - // Save the deferred final step if it was already generated but not yet persisted + try { + await streamer?.fail(errorToString(e)); + } catch (cleanupError) { + streamCleanupFailure ??= { error: cleanupError }; + } if (pendingFinalStep) { try { await call.save(pendingFinalStep, false); @@ -272,10 +289,17 @@ export async function streamText< } } + if (streamCleanupFailure) throw streamCleanupFailure.error; + // If we deferred the final step save, do it now with atomic stream finish. if (pendingFinalStep && streamer) { - const finishStreamId = await streamer.getOrCreateStreamId(); - await call.save(pendingFinalStep, false, finishStreamId); + const finishStreamId = await streamer.getOrCreateStreamId({ + ifAborted: "returnUndefined", + }); + if (finishStreamId) { + await call.save(pendingFinalStep, false, finishStreamId); + } + pendingFinalStep = undefined; } const metadata: GenerationOutputMetadata = { promptMessageId, diff --git a/src/vercel/client/streaming.test.ts b/src/vercel/client/streaming.test.ts index b10baf70..6a60b91f 100644 --- a/src/vercel/client/streaming.test.ts +++ b/src/vercel/client/streaming.test.ts @@ -5,10 +5,7 @@ import type { GenericSchema, SchemaDefinition } from "convex/server"; import { streamText } from "ai"; import { components, initConvexTest } from "./setup.test.js"; import { mockModel } from "./mockModel.js"; -import { - compressUIMessageChunks, - DeltaStreamer, -} from "./streaming.js"; +import { compressUIMessageChunks, DeltaStreamer } from "./streaming.js"; import { getParts } from "../deltas.js"; import type { TestConvex } from "convex-test"; @@ -200,12 +197,73 @@ describe("DeltaStreamer", () => { expect(streamer.abortController.signal.aborted).toBe(true); await streamer.addParts(["ignored"]); expect(streamer.streamId).toBeUndefined(); + await expect( + streamer.getOrCreateStreamId({ ifAborted: "returnUndefined" }), + ).resolves.toBeUndefined(); await expect(streamer.getOrCreateStreamId()).rejects.toThrow( "Cannot create a stream after it has been aborted", ); + await expect(streamer.getStreamId()).rejects.toThrow( + "Cannot create a stream after it has been aborted", + ); + }); + }); + + test("preserves the public throwing behavior after an existing stream aborts", async () => { + await t.run(async (ctx) => { + const streamer = new DeltaStreamer( + components.agent, + ctx, + { ...defaultTestOptions }, + { ...testMetadata, threadId }, + ); + const streamId = await streamer.getStreamId(); + + await streamer.fail("provider error"); + + expect(streamer.streamId).toBe(streamId); + await expect(streamer.getStreamId()).rejects.toThrow( + "Cannot create a stream after it has been aborted", + ); + await expect(streamer.getOrCreateStreamId()).rejects.toThrow( + "Cannot create a stream after it has been aborted", + ); + await expect( + streamer.getOrCreateStreamId({ ifAborted: "returnUndefined" }), + ).resolves.toBeUndefined(); }); }); + test("does not return an ID when abort wins during stream creation", async () => { + let resolveCreate!: (streamId: string) => void; + const creatingStream = new Promise((resolve) => { + resolveCreate = resolve; + }); + const runMutation = vi + .fn() + .mockImplementationOnce(() => creatingStream) + .mockResolvedValueOnce(true); + const streamer = new DeltaStreamer( + components.agent, + { runMutation } as unknown as MutationCtx, + { ...defaultTestOptions }, + { ...testMetadata, threadId }, + ); + + const getting = streamer.getOrCreateStreamId({ + ifAborted: "returnUndefined", + }); + const failing = streamer.fail("provider error"); + resolveCreate("stream-1"); + + await failing; + await expect(getting).resolves.toBeUndefined(); + expect(streamer.streamId).toBe("stream-1"); + await expect(streamer.getStreamId()).rejects.toThrow( + "Cannot create a stream after it has been aborted", + ); + }); + test("shares signal and fail cleanup while stream creation is in flight", async () => { let resolveCreate!: (streamId: string) => void; const creatingStream = new Promise((resolve) => { @@ -293,10 +351,13 @@ describe("DeltaStreamer", () => { }); test("aborts the component stream when a delta write fails", async () => { + const deltaFailure = { + error: { code: "provider_disconnected", message: "Provider dropped" }, + }; const runMutation = vi .fn() .mockResolvedValueOnce("stream-1") - .mockRejectedValueOnce(new Error("delta failed")) + .mockRejectedValueOnce(deltaFailure) .mockResolvedValueOnce(undefined); let abortReason: string | undefined; const streamer = new DeltaStreamer( @@ -314,11 +375,14 @@ describe("DeltaStreamer", () => { await streamer.addParts(["A"]); await streamer.finish(); - expect(abortReason).toBe("delta failed"); + expect(abortReason).toBe("provider_disconnected: Provider dropped"); expect(runMutation).toHaveBeenNthCalledWith( 3, components.agent.streams.abort, - { streamId: "stream-1", reason: "delta failed" }, + { + streamId: "stream-1", + reason: "provider_disconnected: Provider dropped", + }, ); }); diff --git a/src/vercel/client/streaming.ts b/src/vercel/client/streaming.ts index 5f9961c3..4f42f2c7 100644 --- a/src/vercel/client/streaming.ts +++ b/src/vercel/client/streaming.ts @@ -8,6 +8,7 @@ import { type UIMessageChunk, } from "ai"; import { v } from "convex/values"; +import { errorToString } from "../../errors.js"; import { vMessageDoc, vPaginationResult, @@ -300,7 +301,10 @@ export class DeltaStreamer { if (this.#finishedExternally) { return; } - await this.getStreamId(); + const streamId = await this.getOrCreateStreamId({ + ifAborted: "returnUndefined", + }); + if (!streamId) return; this.#nextParts.push(...parts); if ( !this.#ongoingWrite && @@ -319,9 +323,7 @@ export class DeltaStreamer { // A provider can throw while responding to an abort. Join the durable // abort transition here, outside the active delta writer, before // preserving the provider error for the caller. - await this.#abort( - error instanceof Error ? error.message : "stream consumption failed", - ).catch(() => {}); + await this.#abort(errorToString(error)).catch(() => {}); throw error; } // Skip finish if it will be handled externally (atomically with message save) @@ -348,8 +350,26 @@ export class DeltaStreamer { * Get the stream ID, waiting for it to be created if necessary. * Useful for passing to addMessages for atomic finish. */ - public async getOrCreateStreamId(): Promise { - return this.getStreamId(); + public async getOrCreateStreamId(): Promise; + public async getOrCreateStreamId(options: { + ifAborted: "returnUndefined"; + }): Promise; + public async getOrCreateStreamId(options?: { + ifAborted?: "returnUndefined"; + }): Promise { + if (options?.ifAborted !== "returnUndefined") { + return this.getStreamId(); + } + if (this.abortController.signal.aborted) { + await this.#abortPromise; + return undefined; + } + const streamId = await this.getStreamId(); + if (this.abortController.signal.aborted) { + await this.#abortPromise; + return undefined; + } + return streamId; } async #sendDelta() { @@ -368,7 +388,7 @@ export class DeltaStreamer { delta, ); } catch (e) { - await this.#abortDelta(e instanceof Error ? e.message : "unknown error"); + await this.#abortDelta(errorToString(e)); return; } if (!success) { diff --git a/src/vercel/client/utils.ts b/src/vercel/client/utils.ts index 558bac27..751684f3 100644 --- a/src/vercel/client/utils.ts +++ b/src/vercel/client/utils.ts @@ -1,6 +1,8 @@ import type { Context } from "@ai-sdk/provider-utils"; import type { StepResult, StopCondition, ToolSet } from "ai"; +export { errorToString } from "../../errors.js"; + /** * A stop condition that only matches tool calls which completed * successfully (i.e. produced a `tool-result`, not a `tool-error`). @@ -45,10 +47,3 @@ export async function willContinue< } return !!stopWhen && !(await stopWhen({ steps })); } - -export function errorToString(error: unknown): string { - if (error instanceof Error) { - return error.message; - } - return String(error); -}