fix(core): name each provider request failure

biome.json

@@ -46,7 +46,10 @@
 	},
 	"overrides": [
 		{
-			"includes": ["packages/server/src/http/errors.ts"],
+			"includes": [
+				"packages/core/src/providers/models/**/*.ts",
+				"packages/server/src/http/errors.ts"
+			],
 			"plugins": [
 				"./lint/errors-implement-domain-error.grit",
 				"./lint/no-effect-promise.grit",

packages/core/src/conversations/compaction/compaction.steps.ts

@@ -87,4 +87,4 @@ const generate = (
 /**
  * Why a compaction failed.
  */
-type CompactionFailure = Models.RequestFailed | Models.AnswerTimedOut | Models.UnusableAnswer;
+type CompactionFailure = Models.RequestFailure | Models.AnswerTimedOut | Models.UnusableAnswer;

packages/core/src/conversations/floor/facilitator.execution.test.ts

@@ -1,3 +1,4 @@
+import { DisplayName } from "@sugabots/errors";
 import { Effect, Layer } from "effect";
 import { beforeEach, describe, expect, it, vi } from "vitest";
 import { Database, transactional } from "../../database/database.ts";
@@ -47,7 +48,7 @@ const scope: FacilitatorScope = {
 };
 
 const unavailable = Models.fromStream(() =>
-	Effect.fail(new Models.RequestFailed({ message: "provider unavailable", reason: "unavailable" })),
+	Effect.fail(new Models.ModelNotOffered({ model: DisplayName.fromRecord("test-model") })),
 );
 
 describe("an attempt at facilitation", () => {

packages/core/src/conversations/floor/facilitator.ts

@@ -160,7 +160,7 @@ const decide = (
 	model: Models.Interface,
 ): Effect.Effect<
 	FacilitatorDecision,
-	Models.RequestFailed | Models.AnswerTimedOut | Models.UnusableAnswer,
+	Models.RequestFailure | Models.AnswerTimedOut | Models.UnusableAnswer,
 	Database
 > =>
 	Effect.gen(function* () {

packages/core/src/conversations/model-trials/trial.test.ts

@@ -1,3 +1,4 @@
+import { DisplayName } from "@sugabots/errors";
 import { Effect } from "effect";
 import { describe, expect, it, vi } from "vitest";
 import { noDatabase } from "../../database/testing.ts";
@@ -8,13 +9,11 @@ import { ACCURACY_NEEDED, BUDGET_MS, explain, rateAccuracy, rateSpeed, runTrial
 
 /** A model that answers whatever it is told to, so the grading is what is under test. */
 function scripted(
-	answer: (input: Models.StreamRequest) => string | Models.RequestFailed,
+	answer: (input: Models.StreamRequest) => string | Models.RequestFailure,
 ): Models.Interface {
 	return Models.fromStream((input) => {
 		const next = answer(input);
-		return next instanceof Models.RequestFailed
-			? Effect.fail(next)
-			: Effect.succeed(streamed(chunks(next)));
+		return typeof next !== "string" ? Effect.fail(next) : Effect.succeed(streamed(chunks(next)));
 	});
 }
 
@@ -73,9 +72,7 @@ describe("trying a model on the facilitator", () => {
 
 	it("counts a model that cannot answer at all as a failure, not an error", async () => {
 		const report = await trial(
-			scripted(
-				() => new Models.RequestFailed({ message: "provider unavailable", reason: "unavailable" }),
-			),
+			scripted(() => new Models.ModelNotOffered({ model: DisplayName.fromRecord("test-model") })),
 			"facilitate",
 		);
 

packages/core/src/conversations/summaries/summary.steps.test.ts

@@ -1,3 +1,4 @@
+import { DisplayName } from "@sugabots/errors";
 import { Effect, Layer, ManagedRuntime } from "effect";
 import { describe, expect, it, vi } from "vitest";
 import { effectRunner } from "../../database/database.ts";
@@ -62,7 +63,7 @@ describe("summarise", () => {
 			await execution;
 			expect(signal?.aborted).toBe(true);
 			expect(stream).toHaveBeenCalledTimes(1);
-			expect(ended).toEqual([{ failed: "The model did not answer in time." }]);
+			expect(ended).toEqual([{ failed: "The model didn't answer in time." }]);
 			expect(summaries.complete).not.toHaveBeenCalled();
 		} finally {
 			vi.useRealTimers();
@@ -112,7 +113,7 @@ describe("summarise", () => {
 		// whole summary being retried over one bad answer.
 		expect(stream).toHaveBeenCalledTimes(3);
 		expect(summaries.complete).not.toHaveBeenCalled();
-		expect(ended).toEqual([{ failed: "The model's answer could not be used." }]);
+		expect(ended).toEqual([{ failed: "The model answered in a form we couldn't use." }]);
 	});
 
 	it("takes the answer as soon as one of the asks comes back usable", async () => {
@@ -158,11 +159,12 @@ describe("summarise", () => {
 
 	it("does not re-ask a provider that is down, and records the failure", async () => {
 		const { summaries, turns, ended } = fakes();
-		const stream = vi.fn(() =>
-			Effect.fail(
-				new Models.RequestFailed({ message: "provider unavailable", reason: "unavailable" }),
-			),
-		);
+		const unreachable = new Models.ProviderUnreachable({
+			provider: DisplayName.fromRecord("Test provider"),
+			model: DisplayName.fromRecord("test-model"),
+			cause: new Error("ECONNREFUSED"),
+		});
+		const stream = vi.fn(() => Effect.fail(unreachable));
 
 		await runWithServices(
 			summarise(request, Models.fromStream(stream)).pipe(
@@ -173,7 +175,7 @@ describe("summarise", () => {
 		// Asking again would cost the same and fail the same way. The thread's
 		// next turn asks for a summary again.
 		expect(stream).toHaveBeenCalledTimes(1);
-		expect(ended).toEqual([{ failed: "The model provider could not answer." }]);
+		expect(ended).toEqual([{ failed: unreachable.userMessage }]);
 		expect(summaries.complete).not.toHaveBeenCalled();
 	});
 

packages/core/src/conversations/summaries/summary.steps.ts

@@ -142,4 +142,4 @@ function stripFence(text: string): string {
 }
 
 /** Why a summary failed. */
-type SummaryFailure = Models.RequestFailed | Models.AnswerTimedOut | Models.UnusableAnswer;
+type SummaryFailure = Models.RequestFailure | Models.AnswerTimedOut | Models.UnusableAnswer;

packages/core/src/conversations/threads/threads.test.ts

@@ -1,4 +1,5 @@
 import { handleFromName, threadChannel } from "@sugabots/contracts";
+import { DisplayName } from "@sugabots/errors";
 import { and, eq } from "drizzle-orm";
 import { Context, Effect } from "effect";
 import { afterAll, beforeEach, describe, expect, it } from "vitest";
@@ -861,9 +862,12 @@ describe.skipIf(!process.env.DATABASE_URL)("threads, against Postgres", async ()
 			{ content: "The work is complete.", collaborations: [], toolCalls: [] },
 			{ contextCapacity: 128_000, readKeptFrom: null, answeredCollaboration: false },
 		);
-		const providerDown = Models.fromStream(() =>
-			Effect.fail(new Models.RequestFailed({ message: "provider down", reason: "unavailable" })),
-		);
+		const unreachable = new Models.ProviderUnreachable({
+			provider: DisplayName.fromRecord("Test provider"),
+			model: DisplayName.fromRecord("test-model"),
+			cause: new Error("ECONNREFUSED"),
+		});
+		const providerDown = Models.fromStream(() => Effect.fail(unreachable));
 
 		await runOnPostgres(
 			summarise(
@@ -885,9 +889,7 @@ describe.skipIf(!process.env.DATABASE_URL)("threads, against Postgres", async ()
 					and(eq(thread.parentThreadId, details.thread.id), eq(thread.systemAgentKey, "summarise")),
 				),
 		);
-		expect(scribeTurns).toEqual([
-			{ status: "failed", error: "The model provider could not answer." },
-		]);
+		expect(scribeTurns).toEqual([{ status: "failed", error: unreachable.userMessage }]);
 	});
 
 	it("requests cancellation only once while an authorized turn remains active", async () => {

packages/core/src/conversations/turns/turn.segment.test.ts

@@ -1,7 +1,7 @@
-import { userText } from "@sugabots/errors";
+import { DisplayName, userText } from "@sugabots/errors";
 import { tool } from "ai";
 import { eq } from "drizzle-orm";
-import { Context, Effect, Layer, Schema } from "effect";
+import { Context, Duration, Effect, Layer, Schema } from "effect";
 import { afterAll, beforeEach, describe, expect, it, vi } from "vitest";
 import { EventBus } from "../../database/events/bus.ts";
 import type { CommittedEvent } from "../../database/events/outbox.ts";
@@ -42,7 +42,14 @@ describe.skipIf(!process.env.DATABASE_URL)("a turn's segment, against Postgres",
 	const execution = onPostgres({
 		prepare: Context.get(conversations, TurnExecution.Service).prepare,
 	});
-	const providerDown = userText`The model provider could not answer.`;
+	const testProvider = {
+		provider: DisplayName.fromRecord("Test provider"),
+		model: DisplayName.fromRecord("test-model"),
+	};
+	const providerDown = new Models.ProviderServerError({
+		...testProvider,
+		cause: new Error("provider down"),
+	});
 	let threadId: string;
 	let connectionId: string;
 	let hostId: string;
@@ -143,16 +150,16 @@ describe.skipIf(!process.env.DATABASE_URL)("a turn's segment, against Postgres",
 	});
 
 	it("records a failed run, which the workflow runs again", async () => {
-		const outcome = await segmentWith(
-			Models.fromStream(() =>
-				Effect.fail(new Models.RequestFailed({ message: "provider down", reason: "unavailable" })),
-			),
-		);
+		const outcome = await segmentWith(Models.fromStream(() => Effect.fail(providerDown)));
 
 		expect(outcome).toEqual({ _tag: "Retry" });
-		expect(await storedTurn()).toMatchObject({ status: "failed", error: providerDown });
+		expect(await storedTurn()).toMatchObject({ status: "failed", error: providerDown.userMessage });
 		expect(deliveredEvents()).toContainEqual(
-			expect.objectContaining({ type: "message.failed", willRetry: true, error: providerDown }),
+			expect.objectContaining({
+				type: "message.failed",
+				willRetry: true,
+				error: providerDown.userMessage,
+			}),
 		);
 	});
 
@@ -165,24 +172,39 @@ describe.skipIf(!process.env.DATABASE_URL)("a turn's segment, against Postgres",
 				.where(eq(turn.id, prepared.turnId)),
 		);
 
+		const outcome = await segmentWith(Models.fromStream(() => Effect.fail(providerDown)));
+
+		expect(outcome).toEqual({ _tag: "Finished" });
+		expect(await storedTurn()).toMatchObject({ status: "failed", runs: MAX_TURN_RUNS });
+		expect(deliveredEvents()).toContainEqual(
+			expect.objectContaining({ type: "message.failed", willRetry: false }),
+		);
+	});
+
+	it("fails the turn for good when the provider refused the request", async () => {
 		const outcome = await segmentWith(
 			Models.fromStream(() =>
-				Effect.fail(new Models.RequestFailed({ message: "provider down", reason: "unavailable" })),
+				Effect.fail(
+					new Models.ProviderQuotaExhausted({ ...testProvider, cause: new Error("402") }),
+				),
 			),
 		);
 
 		expect(outcome).toEqual({ _tag: "Finished" });
-		expect(await storedTurn()).toMatchObject({ status: "failed", runs: MAX_TURN_RUNS });
 		expect(deliveredEvents()).toContainEqual(
 			expect.objectContaining({ type: "message.failed", willRetry: false }),
 		);
 	});
 
-	it("fails the turn for good when the provider refused the request", async () => {
+	it("fails the turn for good when the provider asks to be left too long", async () => {
 		const outcome = await segmentWith(
 			Models.fromStream(() =>
 				Effect.fail(
-					new Models.RequestFailed({ message: "Provider returned 402", reason: "outOfCredit" }),
+					new Models.ProviderRateLimited({
+						...testProvider,
+						retryAfter: Duration.minutes(5),
+						cause: new Error("429"),
+					}),
 				),
 			),
 		);

packages/core/src/conversations/turns/turn.steps.ts

@@ -1,5 +1,5 @@
 import { streamEvent, threadChannel } from "@sugabots/contracts";
-import { userText } from "@sugabots/errors";
+import { type DomainError, userText } from "@sugabots/errors";
 import type { ToolApprovalConfiguration, ToolSet } from "ai";
 import {
 	Cause,
@@ -65,6 +65,11 @@ const MESSAGE_FLUSH_CHARACTERS = 500;
  */
 const CANCELLATION_CHECK_INTERVAL = Duration.seconds(15);
 const TURN_TIMEOUT = Duration.minutes(10);
+/**
+ * The longest a turn waits for a rate-limited provider before running again.
+ * Past it, the turn ends and its message tells people when to try.
+ */
+const MAX_RETRY_WAIT = Duration.minutes(1);
 /** How many model calls a turn may make, across its segments. */
 const TURN_MODEL_CALLS = 20;
 
@@ -144,7 +149,6 @@ export const runSegment = (
 	});
 
 const finished: SegmentOutcome = { _tag: "Finished" };
-const retry: SegmentOutcome = { _tag: "Retry" };
 
 const emptyReply: ReplyDraft = { content: "", collaborations: [], toolCalls: [] };
 
@@ -158,7 +162,7 @@ type StreamOutcome =
 
 /** Why a reply stopped streaming before the model finished, other than being cancelled. */
 type TurnFailure =
-	| Models.RequestFailed
+	| Models.RequestFailure
 	| ToolApprovalsIncomplete
 	| TurnTimedOut
 	| ApprovedToolChanged
@@ -260,20 +264,32 @@ const generateReply = (
 			 * Logs the failure and records what people are told of it; the turn
 			 * runs again only while that is safe and could help. A reply without
 			 * an answer does not: running it again would repeat every tool call it
-			 * made. Nor does a request the provider refused, which it would refuse again.
+			 * made. Nor does a model request that would fail the same way again,
+			 * or that asks to be left longer than the turn should wait.
 			 */
-			const failed = (failure: TurnFailure) =>
-				logTurnFailure(prepared, failure.message).pipe(
+			const failed = (failure: TurnFailure) => {
+				const request: DomainError | undefined = Models.isRequestFailure(failure)
+					? failure
+					: undefined;
+				const waitsTooLong =
+					request?.retryAfter !== undefined &&
+					Duration.isGreaterThan(request.retryAfter, MAX_RETRY_WAIT);
+				return logTurnFailure(prepared, failure.message).pipe(
 					Effect.andThen(
 						turns.fail(replyTurn, draft, {
 							userMessage: failure.userMessage,
 							mayRunAgain:
 								!(failure instanceof ReplyWithoutAnswer) &&
-								!(failure instanceof Models.RequestFailed && !failure.mayRetry),
+								(request?.isRetryable ?? true) &&
+								!waitsTooLong,
 						}),
 					),
-					Effect.map((willRetry) => (willRetry ? retry : finished)),
+					Effect.map(
+						(willRetry): SegmentOutcome =>
+							willRetry ? { _tag: "Retry", after: request?.retryAfter } : finished,
+					),
 				);
+			};
 
 			if (Exit.isSuccess(streamed)) {
 				if (streamed.value.kind === "suspended") {
@@ -355,7 +371,7 @@ const streamReply = (
 	reply: Ref.Ref<ReplyDraft>,
 ): Effect.Effect<
 	StreamOutcome,
-	| Models.RequestFailed
+	| Models.RequestFailure
 	| ToolApprovalsIncomplete
 	| TurnTimedOut
 	| ApprovedToolChanged

packages/core/src/conversations/turns/turn.workflow.test.ts

@@ -81,6 +81,23 @@ describe("the turn workflow", () => {
 		expect(abandon).not.toHaveBeenCalled();
 	});
 
+	it("waits as long as a rate-limited provider asked before running again", async () => {
+		segment.mockReturnValueOnce(Effect.succeed({ _tag: "Retry", after: Duration.seconds(30) }));
+		await runtime.runPromise(Turn.execute(request(), { discard: true }));
+		await vi.waitFor(() => expect(segment).toHaveBeenCalledTimes(1));
+
+		let waitedSeconds = 0;
+		await vi.waitFor(
+			async () => {
+				await runtime.runPromise(TestClock.adjust(Duration.seconds(1)));
+				waitedSeconds++;
+				expect(segment).toHaveBeenCalledTimes(2);
+			},
+			{ timeout: 5_000, interval: 1 },
+		);
+		expect(waitedSeconds).toBeGreaterThanOrEqual(30);
+	});
+
 	it("keeps the lane until every approval is decided, in whatever order", async () => {
 		segment.mockReturnValueOnce(
 			Effect.succeed({ _tag: "Suspended", approvals: ["first", "second"] }),

packages/core/src/conversations/turns/turn.workflow.ts

@@ -64,7 +64,10 @@ export const SegmentOutcome = Schema.Union([
 		/** The approvals the turn waits for, each decided through `approvalDecided`. */
 		approvals: Schema.Array(Schema.String),
 	}),
-	Schema.TaggedStruct("Retry", {}),
+	Schema.TaggedStruct("Retry", {
+		/** How long the provider asked to be left, when it said; the turn waits at least `RETRY_DELAY`. */
+		after: Schema.optional(Schema.DurationFromMillis),
+	}).pipe(Schema.encodeKeys({ after: "afterMillis" })),
 ]);
 export type SegmentOutcome = typeof SegmentOutcome.Type;
 
@@ -115,7 +118,10 @@ export const turnWorkflow = Lanes.workflow(Turn, {
 				const outcome = yield* turnActivities.activity("segment", request, run);
 				if (outcome._tag === "Finished") return;
 				if (outcome._tag === "Retry") {
-					yield* DurableClock.sleep({ name: `retry/${run}`, duration: RETRY_DELAY });
+					yield* DurableClock.sleep({
+						name: `retry/${run}`,
+						duration: Duration.max(RETRY_DELAY, outcome.after ?? Duration.zero),
+					});
 					continue;
 				}
 				const cancelled = yield* waitForApprovals(run, outcome.approvals);