From fbd45f3044a4d3d8efd04a3c8e0c374d10faab43 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Thu, 4 Jun 2026 00:40:25 +0200 Subject: [PATCH] feat(ai): durable provider-health ledger and credential-expired surfacing MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Promote the volatile per-worker last-working cache to a Postgres provider-health ledger so the fallback chain shares one health signal across every instance, modelled on the atomic-upsert rate limiter. The ledger holds a durable negative cache: an auth-class failure (401/403) benches that provider for a cooldown instead of re-burning the dead round-trip on every generation; a sustained hard failure backs off briefly. The local model is never benched — it stays the guaranteed floor of the chain. Every ledger interaction fails open, so a DB error falls back to today's chain walk exactly. When the user's primary provider fails auth-class, the Coach emits a distinct credential_expired frame that deep-links to reconnect rather than a generic "try again later". Adds the provider_health table (migration 0112) and the errorCredentialExpired copy across all locales. --- messages/de.json | 1 + messages/en.json | 1 + messages/es.json | 1 + messages/fr.json | 1 + messages/it.json | 1 + messages/pl.json | 1 + .../migration.sql | 43 +++ prisma/schema.prisma | 39 +++ src/app/api/insights/chat/route.ts | 12 + .../__tests__/message-thread.test.tsx | 8 + .../insights/coach-panel/message-thread.tsx | 5 + .../__tests__/provider-health-ledger.test.ts | 199 ++++++++++++ src/lib/ai/__tests__/provider-runner.test.ts | 228 +++++++++++++ src/lib/ai/provider-health-ledger.ts | 300 ++++++++++++++++++ src/lib/ai/provider-runner.ts | 112 ++++++- 15 files changed, 949 insertions(+), 3 deletions(-) create mode 100644 prisma/migrations/0112_v1110_provider_health_ledger/migration.sql create mode 100644 src/lib/ai/__tests__/provider-health-ledger.test.ts create mode 100644 src/lib/ai/provider-health-ledger.ts diff --git a/messages/de.json b/messages/de.json index 4c429f993..b80e23b18 100644 --- a/messages/de.json +++ b/messages/de.json @@ -1682,6 +1682,7 @@ "errorBudget": "Du hast das heutige Coach-Limit erreicht. Das Budget setzt sich um Mitternacht (UTC) zurück.", "errorProvider": "Der Coach konnte gerade keinen Auswertungs-Anbieter erreichen. Bitte versuche es in einem Moment erneut.", "errorNetwork": "Keine Internetverbindung — versuche es erneut, sobald du online bist.", + "errorCredentialExpired": "Deine Verbindung zum KI-Anbieter ist abgelaufen. Stelle sie in den Einstellungen wieder her, damit der Coach weiterläuft.", "tagline": "Persönlicher Gesundheitscoach, auf deine Daten gestützt.", "newChat": "Neuer Chat", "send": "Senden", diff --git a/messages/en.json b/messages/en.json index fb782b2f8..26a45e608 100644 --- a/messages/en.json +++ b/messages/en.json @@ -1682,6 +1682,7 @@ "errorBudget": "You've reached today's Coach usage limit. The budget refreshes at midnight UTC.", "errorProvider": "The Coach could not reach an Insights provider just now. Please try again in a moment.", "errorNetwork": "No internet connection — try again once you're back online.", + "errorCredentialExpired": "Your AI provider connection has expired. Reconnect it in Settings to keep the Coach running.", "tagline": "Personal health coach, grounded in your data.", "newChat": "New chat", "send": "Send", diff --git a/messages/es.json b/messages/es.json index 5941fbb4e..2eac6c022 100644 --- a/messages/es.json +++ b/messages/es.json @@ -1682,6 +1682,7 @@ "errorBudget": "Has alcanzado el límite diario del coach. El cupo se reinicia a medianoche (UTC).", "errorProvider": "El coach no ha podido contactar con un proveedor de análisis ahora mismo. Inténtalo en un momento.", "errorNetwork": "Sin conexión a Internet — inténtalo de nuevo cuando vuelvas a estar en línea.", + "errorCredentialExpired": "Tu conexión con el proveedor de IA ha caducado. Vuelve a conectarla en Ajustes para que el Coach siga funcionando.", "tagline": "Coach personal de salud, apoyado en tus datos.", "newChat": "Nueva conversación", "send": "Enviar", diff --git a/messages/fr.json b/messages/fr.json index 3d76f5dd0..e86f01bd4 100644 --- a/messages/fr.json +++ b/messages/fr.json @@ -1682,6 +1682,7 @@ "errorBudget": "Tu as atteint la limite quotidienne du coach. Le quota se réinitialise à minuit (UTC).", "errorProvider": "Le coach n’a pas pu joindre un fournisseur d’analyse maintenant. Réessaie dans un instant.", "errorNetwork": "Pas de connexion Internet — réessaie une fois reconnecté.", + "errorCredentialExpired": "Ta connexion au fournisseur d'IA a expiré. Reconnecte-la dans les Réglages pour que le Coach continue de fonctionner.", "tagline": "Coach personnel de santé, ancré sur tes données.", "newChat": "Nouvelle conversation", "send": "Envoyer", diff --git a/messages/it.json b/messages/it.json index b3b95e42a..e3b9134bf 100644 --- a/messages/it.json +++ b/messages/it.json @@ -1682,6 +1682,7 @@ "errorBudget": "Hai raggiunto il limite giornaliero del coach. Il budget si reimposta a mezzanotte (UTC).", "errorProvider": "Il coach non è riuscito a contattare un provider di analisi ora. Riprova fra poco.", "errorNetwork": "Nessuna connessione Internet — riprova quando sei di nuovo online.", + "errorCredentialExpired": "La connessione al provider di IA è scaduta. Riconnettila nelle Impostazioni per mantenere attivo il Coach.", "tagline": "Coach personale di salute, basato sui tuoi dati.", "newChat": "Nuova chat", "send": "Invia", diff --git a/messages/pl.json b/messages/pl.json index a1ae9b3bc..a8b99cbda 100644 --- a/messages/pl.json +++ b/messages/pl.json @@ -1682,6 +1682,7 @@ "errorBudget": "Osiągnięto dzienny limit coacha. Limit resetuje się o północy (UTC).", "errorProvider": "Coach nie mógł teraz skontaktować się z dostawcą analizy. Spróbuj za chwilę.", "errorNetwork": "Brak połączenia z Internetem — spróbuj ponownie, gdy wrócisz online.", + "errorCredentialExpired": "Połączenie z dostawcą AI wygasło. Połącz je ponownie w Ustawieniach, aby Coach działał dalej.", "tagline": "Osobisty coach zdrowia, oparty na twoich danych.", "newChat": "Nowa rozmowa", "send": "Wyślij", diff --git a/prisma/migrations/0112_v1110_provider_health_ledger/migration.sql b/prisma/migrations/0112_v1110_provider_health_ledger/migration.sql new file mode 100644 index 000000000..7b5a2509b --- /dev/null +++ b/prisma/migrations/0112_v1110_provider_health_ledger/migration.sql @@ -0,0 +1,43 @@ +-- v1.11.0 W1 — durable, multi-instance provider-health ledger. +-- +-- Promotes the volatile per-worker last-working cache in +-- `provider-runner.ts` to Postgres, modelled on the atomic-upsert +-- rate-limiter table so every worker shares one health signal. One row +-- per (user, provider chain entry): +-- - the runner records each generation outcome (ok / hard_failed / +-- auth_failed) and reads the row to deprioritise or skip a provider +-- inside its backoff window instead of re-burning a dead round-trip; +-- - an auth-class failure (401/403) sets `last_result = 'auth_failed'` +-- plus a `next_retry_at` cooldown — a durable negative cache that +-- also drives the proactive `credential_expired` surfacing. +-- +-- Purely additive: one new table, no enum changes, no backfill. Until a +-- provider fails the table is empty and behaviour matches today. +-- Reversibility: forward-only; dropping the table loses only the cache, +-- and the runner repopulates it on the next outcome. + +CREATE TABLE IF NOT EXISTS "provider_health" ( + "id" TEXT NOT NULL, + "user_id" TEXT NOT NULL, + "provider_type" TEXT NOT NULL, + "last_result" TEXT NOT NULL, + "last_status" INTEGER, + "consecutive_failures" INTEGER NOT NULL DEFAULT 0, + "last_ok_at" TIMESTAMP(3), + "last_failure_at" TIMESTAMP(3), + "next_retry_at" TIMESTAMP(3), + "updated_at" TIMESTAMP(3) NOT NULL, + + CONSTRAINT "provider_health_pkey" PRIMARY KEY ("id") +); + +CREATE UNIQUE INDEX IF NOT EXISTS "provider_health_user_id_provider_type_key" + ON "provider_health" ("user_id", "provider_type"); + +CREATE INDEX IF NOT EXISTS "provider_health_user_id_last_result_idx" + ON "provider_health" ("user_id", "last_result"); + +ALTER TABLE "provider_health" + ADD CONSTRAINT "provider_health_user_id_fkey" + FOREIGN KEY ("user_id") REFERENCES "users"("id") + ON DELETE CASCADE ON UPDATE CASCADE; diff --git a/prisma/schema.prisma b/prisma/schema.prisma index 213062dff..bc7918d87 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -424,6 +424,10 @@ model User { // scoped, time-limited, revocable read view of the health record to a // clinician holding the raw `hls_` token (stored only as an HMAC hash). clinicianShareLinks ClinicianShareLink[] + // v1.11.0 W1 — durable per-provider AI-health ledger (one row per + // provider chain entry) so a known-bad credential is negatively cached + // across workers instead of re-burning a 401 every generation. + providerHealth ProviderHealth[] // v1.7.0 profile — optional patient-identity fields surfaced on the // health-record export cover (PDF) and the FHIR `Patient` resource. @@ -2387,6 +2391,41 @@ model RateLimit { @@map("rate_limits") } +/// v1.11.0 W1 — durable, multi-instance provider-health ledger. +/// +/// Promotes the volatile per-worker `lastWorkingCache` +/// (`provider-runner.ts`) to Postgres, modelled on the atomic-upsert +/// rate-limiter pattern so every worker shares one signal. One row per +/// (user, providerType): the runner records each outcome and reads the +/// row to (a) deprioritise / skip a provider inside its backoff window +/// instead of re-burning a dead round-trip every call, and (b) hold a +/// durable negative-cache for an auth-class failure (401/403 = +/// credential dead) far longer than the old 1h in-memory TTL, surfacing +/// it rather than silently retrying. +/// +/// `nextRetryAt` is the backoff gate: while `lastResult = "auth_failed"` +/// and NOW() < `nextRetryAt`, the provider is skipped (negative cache). +/// On a success the row clears to `ok` with a null `nextRetryAt`. +/// Server-internal: never a Measurement, never ingested from a client. +model ProviderHealth { + id String @id @default(cuid()) + userId String @map("user_id") + providerType String @map("provider_type") // ProviderChainType tag + lastResult String @map("last_result") // "ok" | "hard_failed" | "auth_failed" + lastStatus Int? @map("last_status") // HTTP status of the last failure; null on ok/network + consecutiveFailures Int @default(0) @map("consecutive_failures") + lastOkAt DateTime? @map("last_ok_at") + lastFailureAt DateTime? @map("last_failure_at") + nextRetryAt DateTime? @map("next_retry_at") + updatedAt DateTime @updatedAt @map("updated_at") + + user User @relation(fields: [userId], references: [id], onDelete: Cascade) + + @@unique([userId, providerType]) + @@index([userId, lastResult]) + @@map("provider_health") +} + // ─── Idempotency Keys (mobile-client replay protection) ── /// Caches the response of a POST/PUT/PATCH/DELETE request keyed by diff --git a/src/app/api/insights/chat/route.ts b/src/app/api/insights/chat/route.ts index cbbd44bfb..e3e23e92a 100644 --- a/src/app/api/insights/chat/route.ts +++ b/src/app/api/insights/chat/route.ts @@ -336,8 +336,20 @@ Reply now as the assistant, in ${locale === "de" ? "German" : "English"}.`; meta: { attempts: err.attempts.length, firstStatus: err.attempts[0]?.httpStatus ?? null, + credentialExpired: err.primaryCredentialExpired, }, }); + // v1.11.0 W1 — when the user's PRIMARY provider failed with an + // auth-class status (401/403), the credential is dead, not the + // service. Surface a distinct `credential_expired` frame so the + // drawer can deep-link the user to reconnect rather than telling + // them to "try again later" — the gap that let an expired codex + // token silently kill all generation. + if (err.primaryCredentialExpired) { + return streamProviderError({ + code: "coach.provider.credential_expired", + }); + } // v1.4.25 W5 — distinguish provider rate-limit (every attempt // landed on 429) from generic unavailability. The drawer's // error-decoder surfaces the rate-limit copy with a warning diff --git a/src/components/insights/coach-panel/__tests__/message-thread.test.tsx b/src/components/insights/coach-panel/__tests__/message-thread.test.tsx index e837401bd..5f064674e 100644 --- a/src/components/insights/coach-panel/__tests__/message-thread.test.tsx +++ b/src/components/insights/coach-panel/__tests__/message-thread.test.tsx @@ -543,6 +543,14 @@ describe("errorCodeToI18nKey", () => { ); }); + // v1.11.0 W1 — a dead primary credential gets the reconnect copy, not + // the generic provider-unavailable copy. + it("maps coach.provider.credential_expired to the reconnect key", () => { + expect(errorCodeToI18nKey("coach.provider.credential_expired")).toBe( + "insights.coach.errorCredentialExpired", + ); + }); + it("forward-compats unknown codes with the namespaced key", () => { expect(errorCodeToI18nKey("errorProvider")).toBe( "insights.coach.errorProvider", diff --git a/src/components/insights/coach-panel/message-thread.tsx b/src/components/insights/coach-panel/message-thread.tsx index f57ae100c..684ee151f 100644 --- a/src/components/insights/coach-panel/message-thread.tsx +++ b/src/components/insights/coach-panel/message-thread.tsx @@ -88,6 +88,11 @@ export function errorCodeToI18nKey(code: string): string { // than a provider failure ("try again in a moment"). Split out // so the user sees the actionable copy in the offline branch. return "insights.coach.errorNetwork"; + case "coach.provider.credential_expired": + // v1.11.0 W1 — the user's primary AI provider credential is dead + // (auth-class failure). The next action is "reconnect", not "try + // again later", so it carries its own copy pointing at Settings. + return "insights.coach.errorCredentialExpired"; case "coach.provider.unavailable": case "coach.provider.empty": case "coach.provider.none": diff --git a/src/lib/ai/__tests__/provider-health-ledger.test.ts b/src/lib/ai/__tests__/provider-health-ledger.test.ts new file mode 100644 index 000000000..7f01c13f5 --- /dev/null +++ b/src/lib/ai/__tests__/provider-health-ledger.test.ts @@ -0,0 +1,199 @@ +/** + * v1.11.0 W1 — provider-health ledger. + * + * Two layers covered here: + * 1. The Postgres ledger's classification + atomic-upsert SQL shape + * (mocked `prisma`) — an auth failure benches a provider for the + * long cooldown, a hard failure for the short one, and a success + * clears the negative cache. Multi-instance correctness is + * structural (single atomic `INSERT … ON CONFLICT … DO UPDATE`), + * so we assert the upsert is emitted rather than re-proving the + * DB's atomicity. + * 2. The in-memory ledger's negative-cache semantics, which the + * runner tests reuse without standing up Postgres. + */ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +vi.mock("@/lib/db", () => ({ + prisma: { + providerHealth: { findMany: vi.fn() }, + $executeRaw: vi.fn(), + }, +})); + +import { prisma } from "@/lib/db"; +import { + AUTH_FAILURE_COOLDOWN_MS, + HARD_FAILURE_COOLDOWN_MS, + HARD_FAILURE_SKIP_THRESHOLD, + classifyFailure, + createInMemoryProviderHealthLedger, + findCredentialExpiredProviders, + postgresProviderHealthLedger, +} from "../provider-health-ledger"; + +beforeEach(() => { + vi.mocked(prisma.providerHealth.findMany).mockReset(); + vi.mocked(prisma.$executeRaw).mockReset(); + vi.mocked(prisma.$executeRaw).mockResolvedValue(1 as never); +}); + +afterEach(() => { + vi.useRealTimers(); +}); + +describe("classifyFailure", () => { + it("classifies 401 as auth_failed with the long cooldown", () => { + expect(classifyFailure(401)).toEqual({ + result: "auth_failed", + cooldownMs: AUTH_FAILURE_COOLDOWN_MS, + }); + }); + + it("classifies 403 as auth_failed", () => { + expect(classifyFailure(403).result).toBe("auth_failed"); + }); + + it("classifies 429 / 5xx / network as hard_failed with the short cooldown", () => { + expect(classifyFailure(429)).toEqual({ + result: "hard_failed", + cooldownMs: HARD_FAILURE_COOLDOWN_MS, + }); + expect(classifyFailure(503).result).toBe("hard_failed"); + expect(classifyFailure(null).result).toBe("hard_failed"); + }); +}); + +describe("postgresProviderHealthLedger", () => { + it("getSkipHints benches an auth-failed provider in cooldown as credential_expired", async () => { + const future = new Date(Date.now() + 60_000); + vi.mocked(prisma.providerHealth.findMany).mockResolvedValue([ + { + providerType: "codex", + lastResult: "auth_failed", + consecutiveFailures: 1, + nextRetryAt: future, + }, + ] as never); + + const hints = await postgresProviderHealthLedger.getSkipHints("u1"); + expect(hints.get("codex")?.reason).toBe("credential_expired"); + }); + + it("getSkipHints does not bench an expired-cooldown row", async () => { + const past = new Date(Date.now() - 60_000); + vi.mocked(prisma.providerHealth.findMany).mockResolvedValue([ + { + providerType: "codex", + lastResult: "auth_failed", + consecutiveFailures: 1, + nextRetryAt: past, + }, + ] as never); + + const hints = await postgresProviderHealthLedger.getSkipHints("u1"); + expect(hints.size).toBe(0); + }); + + it("getSkipHints benches a hard-failed provider only past the threshold", async () => { + const future = new Date(Date.now() + 60_000); + vi.mocked(prisma.providerHealth.findMany).mockResolvedValue([ + { + providerType: "openai", + lastResult: "hard_failed", + consecutiveFailures: HARD_FAILURE_SKIP_THRESHOLD - 1, + nextRetryAt: future, + }, + { + providerType: "anthropic", + lastResult: "hard_failed", + consecutiveFailures: HARD_FAILURE_SKIP_THRESHOLD, + nextRetryAt: future, + }, + ] as never); + + const hints = await postgresProviderHealthLedger.getSkipHints("u1"); + expect(hints.has("openai")).toBe(false); + expect(hints.get("anthropic")?.reason).toBe("backoff"); + }); + + it("getSkipHints fails open (empty) on a DB error", async () => { + vi.mocked(prisma.providerHealth.findMany).mockRejectedValue( + new Error("db down"), + ); + const hints = await postgresProviderHealthLedger.getSkipHints("u1"); + expect(hints.size).toBe(0); + }); + + it("recordSuccess emits an atomic upsert and never throws on DB error", async () => { + await postgresProviderHealthLedger.recordSuccess("u1", "codex"); + expect(prisma.$executeRaw).toHaveBeenCalledTimes(1); + + vi.mocked(prisma.$executeRaw).mockRejectedValueOnce(new Error("boom")); + await expect( + postgresProviderHealthLedger.recordSuccess("u1", "codex"), + ).resolves.toBeUndefined(); + }); + + it("recordFailure emits an atomic upsert and never throws on DB error", async () => { + await postgresProviderHealthLedger.recordFailure("u1", "codex", 401); + expect(prisma.$executeRaw).toHaveBeenCalledTimes(1); + + vi.mocked(prisma.$executeRaw).mockRejectedValueOnce(new Error("boom")); + await expect( + postgresProviderHealthLedger.recordFailure("u1", "codex", 500), + ).resolves.toBeUndefined(); + }); +}); + +describe("findCredentialExpiredProviders", () => { + it("returns the auth-failed providers still inside cooldown", async () => { + vi.mocked(prisma.providerHealth.findMany).mockResolvedValue([ + { providerType: "codex" }, + ] as never); + expect(await findCredentialExpiredProviders("u1")).toEqual(["codex"]); + }); + + it("fails open on a DB error", async () => { + vi.mocked(prisma.providerHealth.findMany).mockRejectedValue( + new Error("db down"), + ); + expect(await findCredentialExpiredProviders("u1")).toEqual([]); + }); +}); + +describe("createInMemoryProviderHealthLedger", () => { + it("benches an auth-failed provider immediately and clears on success", async () => { + const ledger = createInMemoryProviderHealthLedger(); + await ledger.recordFailure("u1", "codex", 401); + expect((await ledger.getSkipHints("u1")).get("codex")?.reason).toBe( + "credential_expired", + ); + + await ledger.recordSuccess("u1", "codex"); + expect((await ledger.getSkipHints("u1")).size).toBe(0); + }); + + it("benches a hard-failed provider only after the consecutive threshold", async () => { + const ledger = createInMemoryProviderHealthLedger(); + for (let i = 0; i < HARD_FAILURE_SKIP_THRESHOLD - 1; i += 1) { + await ledger.recordFailure("u1", "openai", 503); + } + expect((await ledger.getSkipHints("u1")).has("openai")).toBe(false); + + await ledger.recordFailure("u1", "openai", 503); + expect((await ledger.getSkipHints("u1")).get("openai")?.reason).toBe( + "backoff", + ); + }); + + it("lets the negative cache lapse after the cooldown", async () => { + vi.useFakeTimers(); + const ledger = createInMemoryProviderHealthLedger(); + await ledger.recordFailure("u1", "codex", 401); + expect((await ledger.getSkipHints("u1")).size).toBe(1); + + vi.advanceTimersByTime(AUTH_FAILURE_COOLDOWN_MS + 1); + expect((await ledger.getSkipHints("u1")).size).toBe(0); + }); +}); diff --git a/src/lib/ai/__tests__/provider-runner.test.ts b/src/lib/ai/__tests__/provider-runner.test.ts index 74e02efde..0b40adc17 100644 --- a/src/lib/ai/__tests__/provider-runner.test.ts +++ b/src/lib/ai/__tests__/provider-runner.test.ts @@ -1,5 +1,26 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import type { AIProvider, CompletionParams, CompletionResult } from "../types"; + +// v1.11.0 W1 — the runner now defaults to the Postgres-backed +// provider-health ledger. These pure chain tests do not stand up a DB, +// so swap the default for a no-op; the dedicated ledger-aware cases +// below inject an in-memory ledger explicitly. +vi.mock("../provider-health-ledger", async () => { + const actual = await vi.importActual< + typeof import("../provider-health-ledger") + >("../provider-health-ledger"); + return { + ...actual, + postgresProviderHealthLedger: { + async getSkipHints() { + return new Map(); + }, + async recordSuccess() {}, + async recordFailure() {}, + }, + }; +}); + import { AllProvidersFailedError, clearLastWorkingProviderCache, @@ -8,6 +29,10 @@ import { runRawCompletionWithFallback, runWithFallback, } from "../provider-runner"; +import { + AUTH_FAILURE_COOLDOWN_MS, + createInMemoryProviderHealthLedger, +} from "../provider-health-ledger"; const VALID_RESPONSE = JSON.stringify({ summary: "ok", @@ -433,3 +458,206 @@ describe("runWithFallback — empty input", () => { expect((caught as AllProvidersFailedError).httpStatus).toBe(422); }); }); + +// ── v1.11.0 W1 — durable provider-health ledger integration ────────── + +describe("provider-health ledger — auth-failure negative cache", () => { + it("records the auth failure and deprioritises the dead provider on the next call", async () => { + const ledger = createInMemoryProviderHealthLedger(); + const codex = new ScriptedProvider({ + type: "codex", + // First call 401, then (if ever re-tried) it would succeed — but + // the negative cache must keep it benched. + script: [{ ok: false, error: err(401, "OAuth expired") }, { ok: true }], + }); + const openai = new ScriptedProvider({ + type: "admin-key", + script: [{ ok: true, content: '{ "ok": true }' }, { ok: true }], + }); + const providers = [ + { providerType: "codex" as const, instance: codex }, + { providerType: "openai" as const, instance: openai }, + ]; + + // First call: codex 401 → openai succeeds. Ledger now benches codex. + await runRawCompletionWithFallback({ + userId: "u-neg", + providers, + params: { systemPrompt: "s", userPrompt: "u" }, + ledger, + }); + expect((await ledger.getSkipHints("u-neg")).get("codex")?.reason).toBe( + "credential_expired", + ); + + // Reset the per-worker last-working cache so the ONLY thing steering + // order on the second call is the durable ledger. + clearLastWorkingProviderCache(); + + // Second call: codex is benched to the tail by the negative cache, + // so openai is tried first and codex is never re-invoked. + const codexCallsBefore = codex.callCount; + await runRawCompletionWithFallback({ + userId: "u-neg", + providers, + params: { systemPrompt: "s", userPrompt: "u" }, + ledger, + }); + expect(codex.callCount).toBe(codexCallsBefore); // not re-burned + }); + + it("clears the negative cache after a provider succeeds again", async () => { + const ledger = createInMemoryProviderHealthLedger(); + await ledger.recordFailure("u-clear-neg", "codex", 401); + expect((await ledger.getSkipHints("u-clear-neg")).size).toBe(1); + + const codex = new ScriptedProvider({ + type: "codex", + script: [{ ok: true }], + }); + // A direct success on codex (e.g. user re-linked) clears the row. + await runRawCompletionWithFallback({ + userId: "u-clear-neg", + providers: [{ providerType: "codex", instance: codex }], + params: { systemPrompt: "s", userPrompt: "u" }, + ledger, + }); + expect((await ledger.getSkipHints("u-clear-neg")).size).toBe(0); + }); + + it("lets the benched provider back to the front once the cooldown lapses", async () => { + vi.useFakeTimers(); + const ledger = createInMemoryProviderHealthLedger(); + await ledger.recordFailure("u-lapse", "codex", 401); + expect((await ledger.getSkipHints("u-lapse")).has("codex")).toBe(true); + + vi.advanceTimersByTime(AUTH_FAILURE_COOLDOWN_MS + 1); + expect((await ledger.getSkipHints("u-lapse")).has("codex")).toBe(false); + }); +}); + +describe("provider-health ledger — local model as guaranteed floor", () => { + it("still reaches the local provider even when every keyed provider is benched", async () => { + const ledger = createInMemoryProviderHealthLedger(); + // Pre-bench every remote provider with an auth failure. + await ledger.recordFailure("u-floor", "codex", 401); + await ledger.recordFailure("u-floor", "openai", 401); + await ledger.recordFailure("u-floor", "anthropic", 401); + + const dead = (type: AIProvider["type"]) => + new ScriptedProvider({ type, script: [{ ok: false, error: err(401) }] }); + const local = new ScriptedProvider({ + type: "local", + script: [{ ok: true, content: '{ "floor": true }' }], + }); + + const result = await runRawCompletionWithFallback({ + userId: "u-floor", + providers: [ + { providerType: "codex", instance: dead("codex") }, + { providerType: "openai", instance: dead("admin-key") }, + { providerType: "anthropic", instance: dead("anthropic") }, + { providerType: "local", instance: local }, + ], + params: { systemPrompt: "s", userPrompt: "u" }, + ledger, + }); + + // The local floor is never benched by the negative cache and is + // reached even though every other provider is in an auth cooldown. + expect(result.result.content).toBe('{ "floor": true }'); + expect(result.workingProvider.providerType).toBe("local"); + expect(local.callCount).toBe(1); + }); + + it("never benches the local provider in the ledger even on a hard failure", async () => { + const ledger = createInMemoryProviderHealthLedger(); + const local = new ScriptedProvider({ + type: "local", + script: [{ ok: false, error: err(503) }, { ok: true }], + }); + // A 503 on local records a hard failure but the runner must keep it + // in the chain — there is no remote credential to "expire". + let caught: unknown; + try { + await runRawCompletionWithFallback({ + userId: "u-local-hard", + providers: [{ providerType: "local", instance: local }], + params: { systemPrompt: "s", userPrompt: "u" }, + ledger, + }); + } catch (e) { + caught = e; + } + expect(caught).toBeInstanceOf(AllProvidersFailedError); + + // Even after a recorded hard failure, the next call still tries + // local (NEVER_SKIP) and this time it succeeds. + clearLastWorkingProviderCache(); + const result = await runRawCompletionWithFallback({ + userId: "u-local-hard", + providers: [{ providerType: "local", instance: local }], + params: { systemPrompt: "s", userPrompt: "u" }, + ledger, + }); + expect(result.workingProvider.providerType).toBe("local"); + }); +}); + +describe("AllProvidersFailedError — primaryCredentialExpired", () => { + it("flags primaryCredentialExpired when the first hop is auth-class", async () => { + const ledger = createInMemoryProviderHealthLedger(); + const codex = new ScriptedProvider({ + type: "codex", + script: [{ ok: false, error: err(401) }], + }); + const openai = new ScriptedProvider({ + type: "admin-key", + script: [{ ok: false, error: err(503) }], + }); + let caught: unknown; + try { + await runRawCompletionWithFallback({ + userId: "u-primary-auth", + providers: [ + { providerType: "codex", instance: codex }, + { providerType: "admin-openai", instance: openai }, + ], + params: { systemPrompt: "s", userPrompt: "u" }, + ledger, + }); + } catch (e) { + caught = e; + } + const e = caught as AllProvidersFailedError; + expect(e.primaryCredentialExpired).toBe(true); + }); + + it("does NOT flag primaryCredentialExpired when the first hop is a 5xx", async () => { + const ledger = createInMemoryProviderHealthLedger(); + const codex = new ScriptedProvider({ + type: "codex", + script: [{ ok: false, error: err(503) }], + }); + const openai = new ScriptedProvider({ + type: "admin-key", + script: [{ ok: false, error: err(401) }], + }); + let caught: unknown; + try { + await runRawCompletionWithFallback({ + userId: "u-primary-5xx", + providers: [ + { providerType: "codex", instance: codex }, + { providerType: "admin-openai", instance: openai }, + ], + params: { systemPrompt: "s", userPrompt: "u" }, + ledger, + }); + } catch (e) { + caught = e; + } + const e = caught as AllProvidersFailedError; + expect(e.primaryCredentialExpired).toBe(false); + }); +}); diff --git a/src/lib/ai/provider-health-ledger.ts b/src/lib/ai/provider-health-ledger.ts new file mode 100644 index 000000000..f2bed04a8 --- /dev/null +++ b/src/lib/ai/provider-health-ledger.ts @@ -0,0 +1,300 @@ +/** + * v1.11.0 W1 — durable provider-health ledger. + * + * Epic B Pillar 4 (resilience floor). Promotes the volatile per-worker + * `lastWorkingCache` in `provider-runner.ts` to a Postgres table so the + * fallback chain shares one health signal across every worker, exactly + * like the rate limiter ("rate limits live in Postgres"). The ledger is + * a read-through reporter layered AROUND the existing + * `isHardProviderFailure` classifier — it never re-decides what a + * failure is; it only remembers the outcome. + * + * Two jobs: + * 1. **Negative cache** for an auth-class failure (401/403 = + * "credential dead, re-link required"). Instead of re-burning the + * dead round-trip on every generation, the runner skips a provider + * whose row reads `auth_failed` while `nextRetryAt` is in the + * future. The cooldown is far longer than the old 1h in-memory TTL + * so a single expired codex token does not silently kill every + * generation for the next hour-per-worker. + * 2. **Proactive surfacing.** `findCredentialExpiredProviders` lets + * the coach route / a health check report a known-bad credential + * ("ChatGPT connection expired — reconnect") rather than leaving + * the failure invisible until the next manual Coach turn. + * + * Resilience: every DB interaction is best-effort. The ledger is an + * optimisation, never a gate — a transient DB error must never take + * down generation, so reads fail open (empty result) and writes are + * fire-and-forget. This keeps the runner's existing behaviour the + * floor: worst case, the ledger contributes nothing and the chain walks + * exactly as it does today. + */ + +import { prisma } from "@/lib/db"; +import type { ProviderChainType } from "./provider-chain"; + +/** Auth-class negative-cache cooldown — a dead credential is skipped + * for this long before the chain re-probes it. Long enough that a + * lapsed codex/OpenAI key is not re-tried every generation, short + * enough that a re-linked credential recovers within a day without an + * explicit clear. A successful generation clears it immediately. */ +export const AUTH_FAILURE_COOLDOWN_MS = 6 * 60 * 60 * 1000; // 6h + +/** Backoff for non-auth hard failures (429 / 5xx / network). Short — + * these are usually transient brown-outs, so we only briefly + * deprioritise rather than skip. */ +export const HARD_FAILURE_COOLDOWN_MS = 5 * 60 * 1000; // 5min + +/** A provider stays in the skip set only while its consecutive-failure + * count is at or above this floor — a single transient hard failure + * does not bench a provider; a sustained one does. Auth failures skip + * immediately (count irrelevant) because retrying a dead credential is + * never productive. */ +export const HARD_FAILURE_SKIP_THRESHOLD = 3; + +export type ProviderHealthResult = "ok" | "hard_failed" | "auth_failed"; + +/** A provider the runner should skip (negative cache) on this call, + * with the reason so the caller can surface it. */ +export interface ProviderSkipHint { + providerType: ProviderChainType; + reason: "credential_expired" | "backoff"; + /** When the negative cache lifts (ledger `nextRetryAt`). */ + retryAt: Date | null; +} + +/** + * The ledger surface the runner depends on. Injectable so the pure + * fallback-chain unit tests can pass an in-memory / no-op implementation + * without standing up Postgres, while production wires the durable + * `postgresProviderHealthLedger`. + */ +export interface ProviderHealthLedger { + /** Providers to skip right now for this user, keyed by chain type. + * Fails open (empty) on any error. */ + getSkipHints(userId: string): Promise>; + /** Record a successful generation — clears any negative cache. */ + recordSuccess(userId: string, providerType: ProviderChainType): Promise; + /** Record a hard failure. `httpStatus` null = network/transport. */ + recordFailure( + userId: string, + providerType: ProviderChainType, + httpStatus: number | null, + ): Promise; +} + +/** Classify a failure status into the ledger result + cooldown. An + * auth-class status (401/403) is a dead credential; everything else + * that reaches the ledger is a transient hard failure. */ +export function classifyFailure(httpStatus: number | null): { + result: Exclude; + cooldownMs: number; +} { + if (httpStatus === 401 || httpStatus === 403) { + return { result: "auth_failed", cooldownMs: AUTH_FAILURE_COOLDOWN_MS }; + } + return { result: "hard_failed", cooldownMs: HARD_FAILURE_COOLDOWN_MS }; +} + +/** + * Durable, multi-instance-correct implementation backed by the + * `provider_health` table. Writes use an atomic SQL upsert (the + * rate-limiter pattern) so concurrent workers never clobber each other's + * counters. + */ +export const postgresProviderHealthLedger: ProviderHealthLedger = { + async getSkipHints(userId) { + const hints = new Map(); + try { + const rows = await prisma.providerHealth.findMany({ + where: { userId, lastResult: { in: ["auth_failed", "hard_failed"] } }, + select: { + providerType: true, + lastResult: true, + consecutiveFailures: true, + nextRetryAt: true, + }, + }); + const now = Date.now(); + for (const row of rows) { + const provider = row.providerType as ProviderChainType; + const inCooldown = + row.nextRetryAt !== null && row.nextRetryAt.getTime() > now; + if (!inCooldown) continue; + if (row.lastResult === "auth_failed") { + hints.set(provider, { + providerType: provider, + reason: "credential_expired", + retryAt: row.nextRetryAt, + }); + } else if (row.consecutiveFailures >= HARD_FAILURE_SKIP_THRESHOLD) { + hints.set(provider, { + providerType: provider, + reason: "backoff", + retryAt: row.nextRetryAt, + }); + } + } + } catch { + // Fail open — the ledger is an optimisation, not a gate. + return new Map(); + } + return hints; + }, + + async recordSuccess(userId, providerType) { + try { + // Atomic upsert; a success always clears the negative cache. + await prisma.$executeRaw` + INSERT INTO provider_health + (id, user_id, provider_type, last_result, last_status, + consecutive_failures, last_ok_at, last_failure_at, + next_retry_at, updated_at) + VALUES + (gen_random_uuid()::text, ${userId}, ${providerType}, 'ok', NULL, + 0, NOW(), NULL, NULL, NOW()) + ON CONFLICT (user_id, provider_type) DO UPDATE SET + last_result = 'ok', + last_status = NULL, + consecutive_failures = 0, + last_ok_at = NOW(), + next_retry_at = NULL, + updated_at = NOW() + `; + } catch { + // Fire-and-forget — a failed write never blocks the hot path. + } + }, + + async recordFailure(userId, providerType, httpStatus) { + const { result, cooldownMs } = classifyFailure(httpStatus); + const interval = `${cooldownMs} milliseconds`; + const status = httpStatus ?? null; + try { + // Atomic upsert. `consecutive_failures` accumulates across workers; + // an auth failure forces the cooldown regardless of count, a hard + // failure extends it. The fixed cooldown anchors on NOW() so a + // fresh failure always re-arms the skip window. + await prisma.$executeRaw` + INSERT INTO provider_health + (id, user_id, provider_type, last_result, last_status, + consecutive_failures, last_ok_at, last_failure_at, + next_retry_at, updated_at) + VALUES + (gen_random_uuid()::text, ${userId}, ${providerType}, ${result}, + ${status}, 1, NULL, NOW(), NOW() + ${interval}::interval, NOW()) + ON CONFLICT (user_id, provider_type) DO UPDATE SET + last_result = ${result}, + last_status = ${status}, + consecutive_failures = provider_health.consecutive_failures + 1, + last_failure_at = NOW(), + next_retry_at = NOW() + ${interval}::interval, + updated_at = NOW() + `; + } catch { + // Fire-and-forget. + } + }, +}; + +/** + * Providers whose credential is currently known-bad (auth-class failure + * inside the cooldown window) for a user. Drives the proactive + * `credential_expired` surfacing — the gap that let an expired codex + * token silently kill all generation. Fails open (empty). + */ +export async function findCredentialExpiredProviders( + userId: string, +): Promise { + try { + const rows = await prisma.providerHealth.findMany({ + where: { + userId, + lastResult: "auth_failed", + nextRetryAt: { gt: new Date() }, + }, + select: { providerType: true }, + }); + return rows.map((r) => r.providerType as ProviderChainType); + } catch { + return []; + } +} + +/** + * In-memory ledger for tests that want to assert the negative-cache + * behaviour (auth failure benches a provider, success clears it) + * without Postgres. Mirrors the Postgres semantics: auth failures skip + * immediately, hard failures skip after the consecutive threshold. + */ +export function createInMemoryProviderHealthLedger(): ProviderHealthLedger & { + /** Test inspection: current skip set for a user. */ + inspect(userId: string): Map; +} { + interface Row { + result: ProviderHealthResult; + status: number | null; + consecutiveFailures: number; + nextRetryAt: number | null; + } + const store = new Map>(); + const rowsFor = (userId: string) => { + let m = store.get(userId); + if (!m) { + m = new Map(); + store.set(userId, m); + } + return m; + }; + const hintsFor = (userId: string) => { + const out = new Map(); + const now = Date.now(); + for (const [provider, row] of rowsFor(userId)) { + if (row.nextRetryAt === null || row.nextRetryAt <= now) continue; + if (row.result === "auth_failed") { + out.set(provider, { + providerType: provider, + reason: "credential_expired", + retryAt: new Date(row.nextRetryAt), + }); + } else if ( + row.result === "hard_failed" && + row.consecutiveFailures >= HARD_FAILURE_SKIP_THRESHOLD + ) { + out.set(provider, { + providerType: provider, + reason: "backoff", + retryAt: new Date(row.nextRetryAt), + }); + } + } + return out; + }; + return { + async getSkipHints(userId) { + return hintsFor(userId); + }, + async recordSuccess(userId, providerType) { + rowsFor(userId).set(providerType, { + result: "ok", + status: null, + consecutiveFailures: 0, + nextRetryAt: null, + }); + }, + async recordFailure(userId, providerType, httpStatus) { + const { result, cooldownMs } = classifyFailure(httpStatus); + const prev = rowsFor(userId).get(providerType); + const priorCount = prev && prev.result !== "ok" ? prev.consecutiveFailures : 0; + rowsFor(userId).set(providerType, { + result, + status: httpStatus, + consecutiveFailures: priorCount + 1, + nextRetryAt: Date.now() + cooldownMs, + }); + }, + inspect(userId) { + return hintsFor(userId); + }, + }; +} diff --git a/src/lib/ai/provider-runner.ts b/src/lib/ai/provider-runner.ts index 31e4a6962..16a4c4bd8 100644 --- a/src/lib/ai/provider-runner.ts +++ b/src/lib/ai/provider-runner.ts @@ -5,8 +5,23 @@ import { } from "./generate-insight"; import { InsightSchemaError } from "./schema"; import type { ProviderChainType } from "./provider-chain"; +import { + postgresProviderHealthLedger, + type ProviderHealthLedger, + type ProviderSkipHint, +} from "./provider-health-ledger"; import { annotate } from "@/lib/logging/context"; +/** + * The local model is the guaranteed floor (Epic B Pillar 4): a + * self-hoster running Ollama / LM-Studio must always retain a last + * resort. These provider types have no remote credential that can + * expire, so the durable negative cache must never bench them — they + * stay in the chain even when every keyed provider is in a backoff / + * auth cooldown. + */ +const NEVER_SKIP: ReadonlySet = new Set(["local"]); + /** * v1.4.16 phase B5b — multi-provider redundancy runner. * @@ -133,6 +148,15 @@ export interface RunWithFallbackParams { userId: string; providers: ProviderChainResolved[]; params: CompletionParams; + /** + * v1.11.0 W1 — durable provider-health ledger. Defaults to the + * Postgres-backed implementation in production; unit tests of the pure + * chain pass an in-memory / no-op ledger to avoid standing up a DB. + * The ledger is read-through (skip-hint reorder) + write-through + * (record outcome); it never gates generation — a ledger error always + * fails open to today's behaviour. + */ + ledger?: ProviderHealthLedger; } export interface RunWithFallbackResult extends GenerateInsightOutcome { @@ -150,6 +174,16 @@ export interface RunWithFallbackResult extends GenerateInsightOutcome { export class AllProvidersFailedError extends Error { readonly httpStatus: number; readonly attempts: FallbackHop[]; + /** + * v1.11.0 W1 — true when the FIRST (highest-priority) chain entry + * failed with an auth-class status (401/403). That is the signal a + * user's primary credential is dead and should be surfaced as + * `credential_expired` ("reconnect ChatGPT") rather than a generic + * "try again later". Distinct from "every entry was auth-class" so we + * only deep-link to reconnect when the user's preferred provider is + * the thing that broke. + */ + readonly primaryCredentialExpired: boolean; constructor(attempts: FallbackHop[]) { super( @@ -159,6 +193,10 @@ export class AllProvidersFailedError extends Error { ); this.name = "AllProvidersFailedError"; this.attempts = attempts; + const first = attempts[0]; + this.primaryCredentialExpired = + first !== undefined && + (first.httpStatus === 401 || first.httpStatus === 403); if (attempts.length === 0) { this.httpStatus = 422; return; @@ -204,6 +242,64 @@ function applyLastWorkingCache( return reordered; } +/** + * Apply the durable negative cache. Providers the ledger reports as + * skippable (dead credential inside its cooldown, or a sustained hard + * failure in backoff) are moved to the BACK of the chain rather than + * dropped — so generation is never lost if every healthy provider also + * fails, but a known-bad credential no longer costs a round-trip on the + * hot path. The local floor (`NEVER_SKIP`) is always left in place. + * + * Order within the "deprioritised" tail is stable, preserving the + * original chain order so a re-linked credential resumes its priority + * the moment its cooldown lifts. + */ +function applyHealthLedgerSkips( + providers: ProviderChainResolved[], + skips: Map, +): ProviderChainResolved[] { + if (skips.size === 0) return providers; + const preferred: ProviderChainResolved[] = []; + const deprioritised: ProviderChainResolved[] = []; + for (const p of providers) { + const skip = skips.get(p.providerType); + if (skip && !NEVER_SKIP.has(p.providerType)) { + deprioritised.push(p); + } else { + preferred.push(p); + } + } + return [...preferred, ...deprioritised]; +} + +/** + * Compute the order the chain is walked: first the durable health-ledger + * skips (dead-credential / backoff providers pushed to the tail), then + * the volatile per-worker last-working reorder on top (a provider that + * succeeded most recently jumps to the front of whatever survives). The + * ledger read fails open — on any error the chain is the input order + * exactly as it walked before the ledger existed. + */ +async function resolveChainOrder( + userId: string, + providers: ProviderChainResolved[], + ledger: ProviderHealthLedger, +): Promise { + const skips = await ledger.getSkipHints(userId); + if (skips.size > 0) { + annotate({ + meta: { + ai_chain_skipped_count: skips.size, + ai_chain_credential_expired: Array.from(skips.values()).some( + (s) => s.reason === "credential_expired", + ), + }, + }); + } + const afterSkips = applyHealthLedgerSkips(providers, skips); + return applyLastWorkingCache(userId, afterSkips); +} + function summariseError(e: unknown): { reason: string; status: number | null } { const err = e as { message?: string; httpStatus?: number }; const status = typeof err.httpStatus === "number" ? err.httpStatus : null; @@ -224,12 +320,13 @@ export async function runWithFallback( args: RunWithFallbackParams, ): Promise { const { userId, providers, params } = args; + const ledger = args.ledger ?? postgresProviderHealthLedger; if (providers.length === 0) { throw new AllProvidersFailedError([]); } - const ordered = applyLastWorkingCache(userId, providers); + const ordered = await resolveChainOrder(userId, providers, ledger); const hops: FallbackHop[] = []; for (let i = 0; i < ordered.length; i += 1) { @@ -237,6 +334,7 @@ export async function runWithFallback( try { const outcome = await generateInsight(candidate.instance, params); rememberWorkingProvider(userId, candidate.providerType); + void ledger.recordSuccess(userId, candidate.providerType); annotate({ meta: { ai_chain_working_provider: candidate.providerType, @@ -250,10 +348,13 @@ export async function runWithFallback( }; } catch (error) { if (!isHardProviderFailure(error)) { - // Schema/validation error — bubble immediately. + // Schema/validation error — bubble immediately. NOT recorded in + // the health ledger: a malformed-JSON reply is a prompt-following + // issue, not a provider-availability failure. throw error; } const summary = summariseError(error); + void ledger.recordFailure(userId, candidate.providerType, summary.status); const hop: FallbackHop = { providerType: candidate.providerType, attempt: i + 1, @@ -307,14 +408,17 @@ export async function runRawCompletionWithFallback(args: { userId: string; providers: ProviderChainResolved[]; params: CompletionParams; + /** See `RunWithFallbackParams.ledger`. */ + ledger?: ProviderHealthLedger; }): Promise { const { userId, providers, params } = args; + const ledger = args.ledger ?? postgresProviderHealthLedger; if (providers.length === 0) { throw new AllProvidersFailedError([]); } - const ordered = applyLastWorkingCache(userId, providers); + const ordered = await resolveChainOrder(userId, providers, ledger); const hops: FallbackHop[] = []; for (let i = 0; i < ordered.length; i += 1) { @@ -322,6 +426,7 @@ export async function runRawCompletionWithFallback(args: { try { const result = await candidate.instance.generateCompletion(params); rememberWorkingProvider(userId, candidate.providerType); + void ledger.recordSuccess(userId, candidate.providerType); annotate({ meta: { ai_chain_working_provider: candidate.providerType, @@ -338,6 +443,7 @@ export async function runRawCompletionWithFallback(args: { throw error; } const summary = summariseError(error); + void ledger.recordFailure(userId, candidate.providerType, summary.status); const hop: FallbackHop = { providerType: candidate.providerType, attempt: i + 1,