diff --git a/docs/api/openapi.yaml b/docs/api/openapi.yaml index c4a5f4dba..2e6074491 100644 --- a/docs/api/openapi.yaml +++ b/docs/api/openapi.yaml @@ -14374,6 +14374,43 @@ components: - type: string - type: "null" description: "`...` + last 4 chars of the stored OpenAI key, or null." + responseTimeoutSeconds: + anyOf: + - type: integer + minimum: -9007199254740991 + maximum: 9007199254740991 + - type: "null" + description: v1.22 (#89) — per-user response timeout for AI generation, in seconds (10–600). Null = the generous + built-in default (~180 s). Mainly for slow local/self-hosted backends. + ocrEnabled: + type: boolean + description: v1.22 (#90) — whether a dedicated document-scan (Lab-OCR) provider is active. When false the OCR path uses + the main provider chain. + ocrProvider: + anyOf: + - type: string + enum: + - OPENAI + - ANTHROPIC + - LOCAL + - type: "null" + description: The dedicated document-scan provider, or null when unset. + ocrModel: + anyOf: + - type: string + - type: "null" + ocrBaseUrl: + anyOf: + - type: string + - type: "null" + description: Custom base URL (LOCAL OCR provider only); null otherwise. + hasOcrKey: + type: boolean + ocrKeyPreview: + anyOf: + - type: string + - type: "null" + description: "`...` + last 4 chars of the stored document-scan provider key, or null." required: - provider - model @@ -14385,6 +14422,13 @@ components: - hasLocalKey - hasOpenaiKey - openaiKeyPreview + - responseTimeoutSeconds + - ocrEnabled + - ocrProvider + - ocrModel + - ocrBaseUrl + - hasOcrKey + - ocrKeyPreview additionalProperties: false description: The calling user's AI-provider configuration plus the effective availability (`aiAvailable` + `managedBy`). Reports key presence + a masked preview only — never a plaintext key. diff --git a/messages/de.json b/messages/de.json index c690c5de5..2e4707abb 100644 --- a/messages/de.json +++ b/messages/de.json @@ -4494,7 +4494,24 @@ "completeness": "{answered} von {total} beantwortet", "questionsTitle": "Der Coach hat Rückfragen", "questionsOpenCoach": "Im Coach beantworten" - } + }, + "responseTimeoutHeading": "Antwort-Timeout", + "responseTimeoutBody": "Wie lange auf die Antwort der KI gewartet wird, bevor abgebrochen wird. Erhöhe den Wert für ein lokales oder selbst gehostetes Modell, das beim ersten Aufruf Zeit zum Laden braucht.", + "responseTimeoutLabel": "Antwort-Timeout (Sekunden)", + "responseTimeoutPlaceholder": "Standard (~180)", + "responseTimeoutInvalid": "Gib eine Zahl zwischen 10 und 600 Sekunden ein.", + "ocrHeading": "Dokumentenscan (Laborberichte)", + "ocrBody": "Verwende einen eigenen Anbieter zum Auslesen hochgeladener Laborberichte. Ist die Option aus, nutzt der Scan deinen Hauptanbieter.", + "ocrEnableLabel": "Eigenen Anbieter für den Dokumentenscan aktivieren", + "ocrProviderLabel": "Scan-Anbieter", + "ocrProviderOption": { + "OPENAI": "OpenAI (dein API-Key)", + "ANTHROPIC": "Anthropic (Claude)", + "LOCAL": "Lokales Modell (OpenAI-kompatibel)" + }, + "ocrModelPlaceholder": "z. B. gpt-4o", + "ocrVisionHint": "Dieses Modell kann möglicherweise keine Bilder lesen. Wähle für den Dokumentenscan ein Modell mit Bildverständnis.", + "ocrKeyLabel": "API-Key" }, "dashboard": { "units": { diff --git a/messages/en.json b/messages/en.json index 9d9382cc6..03e44452f 100644 --- a/messages/en.json +++ b/messages/en.json @@ -4494,7 +4494,24 @@ "completeness": "{answered} of {total} answered", "questionsTitle": "The Coach has follow-up questions", "questionsOpenCoach": "Answer them in the Coach" - } + }, + "responseTimeoutHeading": "Response timeout", + "responseTimeoutBody": "How long to wait for the AI to respond before giving up. Raise it for a local or self-hosted model that needs time to load on the first request.", + "responseTimeoutLabel": "Response timeout (seconds)", + "responseTimeoutPlaceholder": "Default (~180)", + "responseTimeoutInvalid": "Enter a number between 10 and 600 seconds.", + "ocrHeading": "Document scanning (lab reports)", + "ocrBody": "Use a dedicated provider for reading uploaded lab reports. When off, scanning uses your main provider.", + "ocrEnableLabel": "Enable a dedicated document-scanning provider", + "ocrProviderLabel": "Scanning provider", + "ocrProviderOption": { + "OPENAI": "OpenAI (your API key)", + "ANTHROPIC": "Anthropic (Claude)", + "LOCAL": "Local model (OpenAI-compatible)" + }, + "ocrModelPlaceholder": "e.g. gpt-4o", + "ocrVisionHint": "This model may not be able to read images. Choose a vision-capable model for document scanning.", + "ocrKeyLabel": "API key" }, "dashboard": { "units": { diff --git a/messages/es.json b/messages/es.json index 3a14f746a..cbe9974df 100644 --- a/messages/es.json +++ b/messages/es.json @@ -4436,7 +4436,24 @@ "completeness": "{answered} de {total} respondidas", "questionsTitle": "El Coach tiene preguntas", "questionsOpenCoach": "Responderlas en el Coach" - } + }, + "responseTimeoutHeading": "Tiempo de espera de la respuesta", + "responseTimeoutBody": "Cuánto esperar a que responda la IA antes de cancelar. Aumenta el valor para un modelo local o autoalojado que necesita tiempo para cargarse en la primera solicitud.", + "responseTimeoutLabel": "Tiempo de espera de la respuesta (segundos)", + "responseTimeoutPlaceholder": "Predeterminado (~180)", + "responseTimeoutInvalid": "Introduce un número entre 10 y 600 segundos.", + "ocrHeading": "Escaneo de documentos (informes de laboratorio)", + "ocrBody": "Usa un proveedor dedicado para leer los informes de laboratorio subidos. Si está desactivado, el escaneo usa tu proveedor principal.", + "ocrEnableLabel": "Activar un proveedor dedicado para el escaneo de documentos", + "ocrProviderLabel": "Proveedor de escaneo", + "ocrProviderOption": { + "OPENAI": "OpenAI (tu clave de API)", + "ANTHROPIC": "Anthropic (Claude)", + "LOCAL": "Modelo local (compatible con OpenAI)" + }, + "ocrModelPlaceholder": "p. ej. gpt-4o", + "ocrVisionHint": "Puede que este modelo no pueda leer imágenes. Elige un modelo con capacidad de visión para el escaneo de documentos.", + "ocrKeyLabel": "Clave de API" }, "dashboard": { "units": { diff --git a/messages/fr.json b/messages/fr.json index fd8340c16..59ce7d7f8 100644 --- a/messages/fr.json +++ b/messages/fr.json @@ -4436,7 +4436,24 @@ "completeness": "{answered} sur {total} renseignés", "questionsTitle": "Le Coach a des questions", "questionsOpenCoach": "Y répondre dans le Coach" - } + }, + "responseTimeoutHeading": "Délai de réponse", + "responseTimeoutBody": "Durée d'attente de la réponse de l'IA avant d'abandonner. Augmentez-la pour un modèle local ou auto-hébergé qui a besoin de temps pour se charger à la première requête.", + "responseTimeoutLabel": "Délai de réponse (secondes)", + "responseTimeoutPlaceholder": "Par défaut (~180)", + "responseTimeoutInvalid": "Saisissez un nombre entre 10 et 600 secondes.", + "ocrHeading": "Numérisation de documents (comptes rendus de laboratoire)", + "ocrBody": "Utilisez un fournisseur dédié pour lire les comptes rendus de laboratoire importés. Désactivé, la numérisation utilise votre fournisseur principal.", + "ocrEnableLabel": "Activer un fournisseur dédié à la numérisation de documents", + "ocrProviderLabel": "Fournisseur de numérisation", + "ocrProviderOption": { + "OPENAI": "OpenAI (votre clé API)", + "ANTHROPIC": "Anthropic (Claude)", + "LOCAL": "Modèle local (compatible OpenAI)" + }, + "ocrModelPlaceholder": "p. ex. gpt-4o", + "ocrVisionHint": "Ce modèle ne peut peut-être pas lire les images. Choisissez un modèle capable de vision pour la numérisation de documents.", + "ocrKeyLabel": "Clé API" }, "dashboard": { "units": { diff --git a/messages/it.json b/messages/it.json index 8a81972dd..ac0c8924c 100644 --- a/messages/it.json +++ b/messages/it.json @@ -4436,7 +4436,24 @@ "completeness": "{answered} su {total} compilati", "questionsTitle": "Il Coach ha alcune domande", "questionsOpenCoach": "Rispondi nel Coach" - } + }, + "responseTimeoutHeading": "Timeout della risposta", + "responseTimeoutBody": "Quanto attendere la risposta dell'IA prima di annullare. Aumentalo per un modello locale o self-hosted che ha bisogno di tempo per caricarsi alla prima richiesta.", + "responseTimeoutLabel": "Timeout della risposta (secondi)", + "responseTimeoutPlaceholder": "Predefinito (~180)", + "responseTimeoutInvalid": "Inserisci un numero tra 10 e 600 secondi.", + "ocrHeading": "Scansione documenti (referti di laboratorio)", + "ocrBody": "Usa un provider dedicato per leggere i referti di laboratorio caricati. Se disattivato, la scansione usa il tuo provider principale.", + "ocrEnableLabel": "Attiva un provider dedicato per la scansione dei documenti", + "ocrProviderLabel": "Provider di scansione", + "ocrProviderOption": { + "OPENAI": "OpenAI (la tua chiave API)", + "ANTHROPIC": "Anthropic (Claude)", + "LOCAL": "Modello locale (compatibile con OpenAI)" + }, + "ocrModelPlaceholder": "es. gpt-4o", + "ocrVisionHint": "Questo modello potrebbe non essere in grado di leggere le immagini. Scegli un modello con capacità di visione per la scansione dei documenti.", + "ocrKeyLabel": "Chiave API" }, "dashboard": { "units": { diff --git a/messages/pl.json b/messages/pl.json index 8535f958a..a3c258548 100644 --- a/messages/pl.json +++ b/messages/pl.json @@ -4436,7 +4436,24 @@ "completeness": "Wypełnione: {answered} z {total}", "questionsTitle": "Coach ma pytania", "questionsOpenCoach": "Odpowiedz w Coachu" - } + }, + "responseTimeoutHeading": "Limit czasu odpowiedzi", + "responseTimeoutBody": "Jak długo czekać na odpowiedź AI, zanim nastąpi przerwanie. Zwiększ wartość dla lokalnego lub samodzielnie hostowanego modelu, który potrzebuje czasu na załadowanie przy pierwszym żądaniu.", + "responseTimeoutLabel": "Limit czasu odpowiedzi (sekundy)", + "responseTimeoutPlaceholder": "Domyślnie (~180)", + "responseTimeoutInvalid": "Podaj liczbę od 10 do 600 sekund.", + "ocrHeading": "Skanowanie dokumentów (wyniki badań)", + "ocrBody": "Użyj dedykowanego dostawcy do odczytu przesłanych wyników badań. Gdy wyłączone, skanowanie używa Twojego głównego dostawcy.", + "ocrEnableLabel": "Włącz dedykowanego dostawcę do skanowania dokumentów", + "ocrProviderLabel": "Dostawca skanowania", + "ocrProviderOption": { + "OPENAI": "OpenAI (Twój klucz API)", + "ANTHROPIC": "Anthropic (Claude)", + "LOCAL": "Model lokalny (zgodny z OpenAI)" + }, + "ocrModelPlaceholder": "np. gpt-4o", + "ocrVisionHint": "Ten model może nie odczytywać obrazów. Wybierz model obsługujący obraz do skanowania dokumentów.", + "ocrKeyLabel": "Klucz API" }, "dashboard": { "units": { diff --git a/prisma/migrations/0198_v122_ai_settings/migration.sql b/prisma/migrations/0198_v122_ai_settings/migration.sql new file mode 100644 index 000000000..d74386379 --- /dev/null +++ b/prisma/migrations/0198_v122_ai_settings/migration.sql @@ -0,0 +1,26 @@ +-- v1.22 (#89 + #90) — AI-settings wave. +-- +-- Two additive, UI/DB-backed features on the per-user AI config (no new env +-- vars — every value is configured in the app): +-- +-- #89 ai_response_timeout_seconds — per-user response timeout for AI +-- generation, in SECONDS. Threaded onto CompletionParams.timeoutMs +-- (x1000). NULL = the generous built-in default (~180 s). Surfaced in +-- Settings -> AI mainly for local/self-hosted servers where the first +-- request can take >60 s while the model loads. +-- +-- #90 dedicated document-scan (Lab-OCR) provider. OFF by default +-- (ai_ocr_enabled = false) -> the OCR ingestion uses the main provider +-- chain unchanged. When enabled, the OCR path resolves this dedicated +-- provider/model/key instead. ai_ocr_key_encrypted is AES-256-GCM +-- ciphertext like every other *_encrypted column (registered in the +-- encrypted-columns registry + the rotation script). +-- +-- All columns are nullable / defaulted, so existing rows need no backfill. +ALTER TABLE "User" + ADD COLUMN "ai_response_timeout_seconds" INTEGER, + ADD COLUMN "ai_ocr_enabled" BOOLEAN NOT NULL DEFAULT false, + ADD COLUMN "ai_ocr_provider" TEXT, + ADD COLUMN "ai_ocr_model" TEXT, + ADD COLUMN "ai_ocr_base_url" TEXT, + ADD COLUMN "ai_ocr_key_encrypted" TEXT; diff --git a/prisma/schema.prisma b/prisma/schema.prisma index 60f2af0a6..dfcb5020a 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -254,6 +254,27 @@ model User { // behaviour (single-provider resolution via `aiProvider`). aiProviderChain Json? @map("ai_provider_chain") + // v1.22 (#89) — per-user response timeout for AI generation, in SECONDS. + // Threaded onto `CompletionParams.timeoutMs` (×1000). Null = the generous + // built-in default (~180 s). Surfaced in Settings → AI mainly for + // local/self-hosted servers (an MLX/exo backend can take >60 s on the + // first request while it loads the model), where the legacy 60 s client + // default timed the Coach out before the first token landed. + aiResponseTimeoutSeconds Int? @map("ai_response_timeout_seconds") + + // v1.22 (#90) — dedicated document-scan (Lab-OCR) provider. OFF by default: + // the lab-report OCR path uses the main provider chain unchanged. When + // `aiOcrEnabled` is true the OCR ingestion resolves THIS provider/model/key + // instead, so a user can keep a cheap text Coach on one provider and route + // expensive vision OCR to a vision-capable one. The key is AES-256-GCM at + // rest like every other `*Encrypted` column (registered in + // `src/lib/crypto/encrypted-columns.ts` + the rotation script). + aiOcrEnabled Boolean @default(false) @map("ai_ocr_enabled") + aiOcrProvider String? @map("ai_ocr_provider") // "OPENAI" | "ANTHROPIC" | "LOCAL" + aiOcrModel String? @map("ai_ocr_model") + aiOcrBaseUrl String? @map("ai_ocr_base_url") // LOCAL OCR provider only + aiOcrKeyEncrypted String? @map("ai_ocr_key_encrypted") + // Onboarding onboardingCompletedAt DateTime? @map("onboarding_completed_at") // v1.4.25 W14b — last persisted onboarding wizard step. diff --git a/scripts/rotate-encryption-key.ts b/scripts/rotate-encryption-key.ts index 54e725986..41e8ee4af 100644 --- a/scripts/rotate-encryption-key.ts +++ b/scripts/rotate-encryption-key.ts @@ -191,6 +191,7 @@ async function main() { "aiAnthropicKeyEncrypted", "aiLocalKeyEncrypted", "aiOpenaiKeyEncrypted", + "aiOcrKeyEncrypted", "insuranceNumberEncrypted", ]; for (const field of userFields) { diff --git a/src/app/api/insights/chat/__tests__/route-snapshot-once.test.ts b/src/app/api/insights/chat/__tests__/route-snapshot-once.test.ts index eeaea14f3..39423d43b 100644 --- a/src/app/api/insights/chat/__tests__/route-snapshot-once.test.ts +++ b/src/app/api/insights/chat/__tests__/route-snapshot-once.test.ts @@ -50,12 +50,12 @@ vi.mock("@/lib/i18n/server-locale", () => ({ resolveServerLocale: vi.fn(async () => "en"), })); -const { runRawCompletionWithFallback } = vi.hoisted(() => ({ - runRawCompletionWithFallback: vi.fn(), +const { runStreamingRawCompletionWithFallback } = vi.hoisted(() => ({ + runStreamingRawCompletionWithFallback: vi.fn(), })); vi.mock("@/lib/ai/provider-runner", () => ({ AllProvidersFailedError: class extends Error {}, - runRawCompletionWithFallback, + runStreamingRawCompletionWithFallback, })); vi.mock("@/lib/ai/provider", () => ({ // v1.20.0 (F1) — pin a no-tools provider so this suite exercises the legacy @@ -138,17 +138,18 @@ vi.mock("@/lib/validations/coach-prefs", () => ({ parseCoachPrefs: vi.fn(() => ({ defaultWindow: undefined })), DEFAULT_REMINDER_SUGGESTION_PREFS: {}, })); +// v1.22 (#89) — the provider call now runs INSIDE the stream producer. Capture +// the producer promise so the test can await it before asserting on call args. +const sse = vi.hoisted(() => ({ done: Promise.resolve() as Promise })); vi.mock("@/lib/sse/create-stream", () => ({ - // Drain the producer synchronously so the handler completes; we only care - // about the provider call args, not the wire frames. createSseStream: ( producer: (c: { signal: { aborted: boolean }; enqueue: () => void; }) => void | Promise, ) => { - void Promise.resolve( - producer({ signal: { aborted: true }, enqueue: () => {} }), + sse.done = Promise.resolve( + producer({ signal: { aborted: false }, enqueue: () => {} }), ); return new ReadableStream(); }, @@ -158,6 +159,12 @@ import { POST } from "../route"; const post = POST as unknown as (req: Request) => Promise; +async function postAndDrain(body: Record): Promise { + const res = await post(chatReq(body)); + await sse.done; + return res; +} + function chatReq(body: Record): Request { return new Request("http://localhost/api/insights/chat", { method: "POST", @@ -167,7 +174,8 @@ function chatReq(body: Record): Request { } function lastUserPrompt(): string { - const calls = runRawCompletionWithFallback.mock.calls as unknown as Array< + const calls = runStreamingRawCompletionWithFallback.mock + .calls as unknown as Array< [{ params: { messages: Array<{ role: string; content: string }> } }] >; const last = calls[calls.length - 1]; @@ -178,14 +186,14 @@ function lastUserPrompt(): string { describe("coach chat — snapshot sent once per conversation (C4)", () => { beforeEach(() => { vi.clearAllMocks(); - runRawCompletionWithFallback.mockResolvedValue({ + runStreamingRawCompletionWithFallback.mockResolvedValue({ result: { content: "Your BP looks stable.", tokensUsed: 42, model: "m" }, workingProvider: { providerType: "admin-openai" }, }); }); it("ships the full SNAPSHOT + grounding on the first turn", async () => { - await post(chatReq({ message: "How is my BP?" })); + await postAndDrain({ message: "How is my BP?" }); const prompt = lastUserPrompt(); expect(prompt).toContain(SNAPSHOT_JSON); expect(prompt).toContain(GROUNDING); @@ -201,7 +209,7 @@ describe("coach chat — snapshot sent once per conversation (C4)", () => { ], }); - await post(chatReq({ conversationId: "c1", message: "High?" })); + await postAndDrain({ conversationId: "c1", message: "High?" }); const prompt = lastUserPrompt(); // The expensive figures + grounding must be gone … expect(prompt).not.toContain(SNAPSHOT_JSON); diff --git a/src/app/api/insights/chat/__tests__/route-tool-mode.test.ts b/src/app/api/insights/chat/__tests__/route-tool-mode.test.ts index aa53e254b..38f51811c 100644 --- a/src/app/api/insights/chat/__tests__/route-tool-mode.test.ts +++ b/src/app/api/insights/chat/__tests__/route-tool-mode.test.ts @@ -52,12 +52,12 @@ vi.mock("@/lib/i18n/server-locale", () => ({ resolveServerLocale: vi.fn(async () => "en"), })); -const { runRawCompletionWithFallback } = vi.hoisted(() => ({ - runRawCompletionWithFallback: vi.fn(), +const { runStreamingRawCompletionWithFallback } = vi.hoisted(() => ({ + runStreamingRawCompletionWithFallback: vi.fn(), })); vi.mock("@/lib/ai/provider-runner", () => ({ AllProvidersFailedError: class extends Error {}, - runRawCompletionWithFallback, + runStreamingRawCompletionWithFallback, })); const { resolveProviderChain } = vi.hoisted(() => ({ @@ -194,6 +194,10 @@ const { parseKeyValuesSentinel } = await import("@/lib/ai/coach/keyvalues"); const { parseSuggestReminder } = await import("@/lib/ai/coach/suggest-reminder"); +// v1.22 (#89) — the provider call + guards + persistence now run INSIDE the +// stream producer (heartbeat-fronted). The mock captures the producer's promise +// so tests can await it after `post()` before asserting on persisted effects. +const sse = vi.hoisted(() => ({ done: Promise.resolve() as Promise })); vi.mock("@/lib/sse/create-stream", () => ({ createSseStream: ( producer: (c: { @@ -201,8 +205,9 @@ vi.mock("@/lib/sse/create-stream", () => ({ enqueue: () => void; }) => void | Promise, ) => { - void Promise.resolve( - producer({ signal: { aborted: true }, enqueue: () => {} }), + // aborted:false so the producer streams to completion (persist + frames). + sse.done = Promise.resolve( + producer({ signal: { aborted: false }, enqueue: () => {} }), ); return new ReadableStream(); }, @@ -212,6 +217,13 @@ import { POST } from "../route"; const post = POST as unknown as (req: Request) => Promise; +/** Run a POST and wait for the in-stream producer to finish. */ +async function postAndDrain(body: Record): Promise { + const res = await post(chatReq(body)); + await sse.done; + return res; +} + function chatReq(body: Record): Request { return new Request("http://localhost/api/insights/chat", { method: "POST", @@ -233,10 +245,10 @@ describe("coach chat — tool-mode routing (F1)", () => { resolveProviderChain.mockResolvedValue([ { providerType: "anthropic", instance: {} }, // no supportsTools=false ⇒ tools on ]); - await post(chatReq({ message: "How is my BP?" })); + await postAndDrain({ message: "How is my BP?" }); expect(runCoachToolLoop).toHaveBeenCalledTimes(1); - expect(runRawCompletionWithFallback).not.toHaveBeenCalled(); + expect(runStreamingRawCompletionWithFallback).not.toHaveBeenCalled(); // The base context carries the inventory, NOT the snapshot figures. const loopArgs = (runCoachToolLoop.mock.calls[0] as unknown[])[0] as { messages: Array<{ content: string }>; @@ -277,6 +289,7 @@ describe("coach chat — tool-mode routing (F1)", () => { ); const res = await post(chatReq({ message: "How is my BP?" })); + await sse.done; // No throw → no 500. The provider-error frame streams over a 200. expect(res.status).toBe(200); @@ -293,7 +306,7 @@ describe("coach chat — tool-mode routing (F1)", () => { resolveProviderChain.mockResolvedValue([ { providerType: "anthropic", instance: {} }, ]); - await post(chatReq({ message: "How is my BP?" })); + await postAndDrain({ message: "How is my BP?" }); const calls = (appendMessage as ReturnType).mock.calls; const assistantCall = calls.find( (c) => (c[0] as { role: string }).role === "assistant", @@ -338,7 +351,7 @@ describe("coach chat — tool-mode routing (F1)", () => { ], }), ); - await post(chatReq({ message: "How is my BP?" })); + await postAndDrain({ message: "How is my BP?" }); const calls = (appendMessage as ReturnType).mock.calls; const assistantCall = calls.find( (c) => (c[0] as { role: string }).role === "assistant", @@ -353,14 +366,14 @@ describe("coach chat — tool-mode routing (F1)", () => { { providerType: "anthropic", instance: {} }, { providerType: "local", instance: { supportsTools: false } }, ]); - runRawCompletionWithFallback.mockResolvedValue({ + runStreamingRawCompletionWithFallback.mockResolvedValue({ result: { content: "Your BP is steady.", tokensUsed: 42, model: "m" }, workingProvider: { providerType: "anthropic" }, }); - await post(chatReq({ message: "How is my BP?" })); + await postAndDrain({ message: "How is my BP?" }); expect(runCoachToolLoop).not.toHaveBeenCalled(); - expect(runRawCompletionWithFallback).toHaveBeenCalledTimes(1); + expect(runStreamingRawCompletionWithFallback).toHaveBeenCalledTimes(1); // Single-round budget reservation in the fallback path; the 4th arg is the // provider-aware daily cap (F1). expect(reserveBudget).toHaveBeenCalledWith( @@ -371,7 +384,7 @@ describe("coach chat — tool-mode routing (F1)", () => { ); // The legacy path ships the snapshot figures in the user turn. const params = ( - (runRawCompletionWithFallback.mock.calls[0] as unknown[])[0] as { + (runStreamingRawCompletionWithFallback.mock.calls[0] as unknown[])[0] as { params: { messages: Array<{ content: string }> }; } ).params; diff --git a/src/app/api/insights/chat/route.ts b/src/app/api/insights/chat/route.ts index f43d84ec0..d6a9b79b8 100644 --- a/src/app/api/insights/chat/route.ts +++ b/src/app/api/insights/chat/route.ts @@ -40,7 +40,7 @@ import { requireModuleEnabled } from "@/lib/modules/gate"; import { resolveServerLocale } from "@/lib/i18n/server-locale"; import { AllProvidersFailedError, - runRawCompletionWithFallback, + runStreamingRawCompletionWithFallback, } from "@/lib/ai/provider-runner"; import { resolveProviderChain, resolveProvider } from "@/lib/ai/provider"; import { assertConsentForChain } from "@/lib/ai/consent-guard"; @@ -134,6 +134,17 @@ function encodeFrame(event: CoachStreamEvent): Uint8Array { return new TextEncoder().encode(`data: ${JSON.stringify(event)}\n\n`); } +/** + * v1.22 (#89) — keepalive heartbeat. An SSE COMMENT frame (a line starting + * with `:`) carries no `data:` payload, so the Coach client's frame parser + * (`parseSseChunk` finds the `data:` line; a comment frame has none) drops it + * silently — exactly what a keepalive should be. Flushed on `HEARTBEAT_MS` + * while the provider is still loading the model / generating, so a reverse + * proxy never idle-drops the long-lived SSE connection. + */ +const HEARTBEAT_MS = 12_000; +const HEARTBEAT_FRAME = new TextEncoder().encode(": ka\n\n"); + /** * Split a full assistant reply into ~roughly-word-sized chunks so the * UI gets a "streaming" feel even when the underlying provider client @@ -371,9 +382,24 @@ async function handleChatRequest(request: NextRequest): Promise { // we don't accidentally widen narrow per-call scopes. const prefsRow = await prisma.user.findUnique({ where: { id: userId }, - select: { coachPrefsJson: true, displayName: true }, + select: { + coachPrefsJson: true, + displayName: true, + // v1.22 (#89) — per-user response timeout (seconds), mainly for slow + // local/self-hosted backends. Threaded onto the provider call below. + aiResponseTimeoutSeconds: true, + }, }); const coachPrefs = parseCoachPrefs(prefsRow?.coachPrefsJson); + // v1.22 (#89) — the upstream timeout for THIS turn's provider call. On the + // streaming (local) path it is the per-idle-gap ceiling; on the buffered path + // it is the whole-call ceiling. A generous ~180 s default replaces the legacy + // 60 s client default that timed the Coach out on an MLX/exo backend whose + // first request loads the model. Clamped to sane bounds at write-time. + const aiResponseTimeoutMs = + prefsRow?.aiResponseTimeoutSeconds != null + ? prefsRow.aiResponseTimeoutSeconds * 1000 + : 180_000; const effectiveScope = scope?.window === undefined && coachPrefs.defaultWindow ? { ...(scope ?? {}), window: coachPrefs.defaultWindow } @@ -561,564 +587,652 @@ Reply now as the assistant, in ${locale === "de" ? "German" : "English"}.`; return streamProviderError({ code: "coach.budget.exceeded" }); } - let result: CompletionResult; - let workingProviderType: string; - let toolTrace: CoachToolTrace[] = []; - // v1.21.0 (P6) — the present tool-result payloads this turn, for the post-hoc - // prose number-verifier. Empty on the no-tools path. - let toolResultPayloads: unknown[] = []; - // v1.21.2 (A8) — no-tools/local-provider parity for the prose number-verifier. - // The tool path grades prose against the figures the tools returned; the - // no-tools path has no tools, so the authoritative set is the SNAPSHOT the - // model was actually shown this turn — `snapshot.sections`, the structured - // record `snapshotJson` is serialised from, which already carries the - // correlations-snapshot block. Populated only when the full figures were - // delivered this turn (`includeFullSnapshot`); on a cheap follow-up the block - // was not re-sent, so there is no fresh authoritative set to grade against. - let noToolsSnapshotPayloads: unknown[] = []; - let totalTokensSpent: number; - // v1.21.0 (F3) — cached-input tokens to subtract at reconcile (prompt-cached - // input the user did not re-pay for must not be billed to the daily meter). - let cachedTokensSpent = 0; - try { - if (toolMode) { - // v1.20.0 (F1) — base context: the full system prompt + a tool-mode - // grounding addendum, with the tiny DATA INVENTORY manifest and the - // transcript on the user turn. The figures are NOT in the prompt — the - // model pulls only what it needs via the retrieval tools. The inventory - // build reuses the snapshot we already computed (60s LRU), so the tools - // that fire this turn share its reads. - const inventory = await buildCoachDataInventory(userId, effectiveScope); - const toolSystem = `${systemPrompt}\n\n${buildToolModeAddendum(locale)}`; - // v1.21.0 (D1) — when the Coach was opened from a metric page/card, thread - // the launch sources into a one-line FOCUS hint so tool mode honours the - // metric the user is looking at (the no-tools path already narrows the - // snapshot; the inventory probes the full set, so this is the tool-mode - // equivalent of that narrowing). Empty string on a generic open. - const focusHint = renderFocusHint(effectiveScope?.sources); - const focusBlock = focusHint ? `${focusHint}\n\n` : ""; - const messages: AiMessage[] = [ - { - role: "user", - content: `${focusBlock}${renderDataInventory(inventory)}${guidedBlock} + // v1.22 (#89) — the provider call + every safety guard + persistence run + // INSIDE the SSE stream now (see `produceReply` + the heartbeat-fronted + // emit below). This is the real fix for a slow local backend: the HTTP + // response headers flush immediately, a keepalive comment frame goes out + // every few seconds while the model is still loading / generating so the + // reverse proxy never idle-drops the connection, and the no-tools (local) + // path streams real provider tokens with a per-idle-gap timeout instead of + // one buffered fetch under a total-timeout. Client-visible token frames + // still carry the FULLY-GUARDED text — every guard runs on the complete + // reply before the first token frame leaves, exactly as before. + type ReplyOutcome = + | { + ok: true; + replyText: string; + provenance: typeof snapshot.provenance; + suggestion: CoachSuggestion | null; + action: CoachSuggestedAction | null; + messageId: string; + totalTokens: number; + model: string | null; + } + | { ok: false; code: string }; + + async function produceReply(): Promise { + let result: CompletionResult; + let workingProviderType: string; + let toolTrace: CoachToolTrace[] = []; + // v1.21.0 (P6) — the present tool-result payloads this turn, for the post-hoc + // prose number-verifier. Empty on the no-tools path. + let toolResultPayloads: unknown[] = []; + // v1.21.2 (A8) — no-tools/local-provider parity for the prose number-verifier. + // The tool path grades prose against the figures the tools returned; the + // no-tools path has no tools, so the authoritative set is the SNAPSHOT the + // model was actually shown this turn — `snapshot.sections`, the structured + // record `snapshotJson` is serialised from, which already carries the + // correlations-snapshot block. Populated only when the full figures were + // delivered this turn (`includeFullSnapshot`); on a cheap follow-up the block + // was not re-sent, so there is no fresh authoritative set to grade against. + let noToolsSnapshotPayloads: unknown[] = []; + let totalTokensSpent: number; + // v1.21.0 (F3) — cached-input tokens to subtract at reconcile (prompt-cached + // input the user did not re-pay for must not be billed to the daily meter). + let cachedTokensSpent = 0; + try { + if (toolMode) { + // v1.20.0 (F1) — base context: the full system prompt + a tool-mode + // grounding addendum, with the tiny DATA INVENTORY manifest and the + // transcript on the user turn. The figures are NOT in the prompt — the + // model pulls only what it needs via the retrieval tools. The inventory + // build reuses the snapshot we already computed (60s LRU), so the tools + // that fire this turn share its reads. + const inventory = await buildCoachDataInventory(userId, effectiveScope); + const toolSystem = `${systemPrompt}\n\n${buildToolModeAddendum(locale)}`; + // v1.21.0 (D1) — when the Coach was opened from a metric page/card, thread + // the launch sources into a one-line FOCUS hint so tool mode honours the + // metric the user is looking at (the no-tools path already narrows the + // snapshot; the inventory probes the full set, so this is the tool-mode + // equivalent of that narrowing). Empty string on a generic open. + const focusHint = renderFocusHint(effectiveScope?.sources); + const focusBlock = focusHint ? `${focusHint}\n\n` : ""; + const messages: AiMessage[] = [ + { + role: "user", + content: `${focusBlock}${renderDataInventory(inventory)}${guidedBlock} CONVERSATION ${transcript} Reply now as the assistant, in ${locale === "de" ? "German" : "English"}. Fetch any figures you cite with the tools first.`, - }, - ]; - const loop = await runCoachToolLoop({ - userId, - providers: chain, - system: toolSystem, - messages, - tools: COACH_TOOL_DEFS, - temperature: AI_BUDGETS.coach.temperature, - maxTokens: AI_BUDGETS.coach.maxTokens, - fallbackWindow: effectiveScope?.window, - // v1.21.0 (D5-1) — share the inventory's full-source snapshot across - // every tool so the turn builds ONE snapshot, not one per tool. The - // probe scope is the exact scope the inventory was built against, so the - // per-tool reads land its 60s LRU entry. - sharedScope: inventory.probeScope, - // v1.20.1 — thread the abort signal so a mid-generation disconnect tears - // down the per-round provider calls instead of paying the full cost. - signal: request.signal, - }); - result = loop.result; - workingProviderType = loop.workingProviderType; - toolTrace = loop.toolTrace; - toolResultPayloads = (loop.toolResults ?? []).map((r) => r.data); - totalTokensSpent = loop.totalTokens; - cachedTokensSpent = loop.cachedTokens; - } else { - const fallback = await runRawCompletionWithFallback({ - userId, - providers: chain, - // v1.20.0 — the no-tools path still builds one assembled user turn (the - // transcript-flattening preserves the once-per-conversation snapshot - // trick + grounding exactly), so it ships as a single user message. The - // stable persona/grounding rides `system` and is now cache-eligible. - params: singleUserTurn({ - system: systemPrompt, - user: userPrompt, + }, + ]; + const loop = await runCoachToolLoop({ + userId, + providers: chain, + system: toolSystem, + messages, + tools: COACH_TOOL_DEFS, temperature: AI_BUDGETS.coach.temperature, maxTokens: AI_BUDGETS.coach.maxTokens, - // v1.20.1 — thread the request's abort signal so a mid-generation - // client disconnect tears the upstream provider call down instead of - // paying the full token cost into a closed connection. + fallbackWindow: effectiveScope?.window, + // v1.21.0 (D5-1) — share the inventory's full-source snapshot across + // every tool so the turn builds ONE snapshot, not one per tool. The + // probe scope is the exact scope the inventory was built against, so the + // per-tool reads land its 60s LRU entry. + sharedScope: inventory.probeScope, + // v1.20.1 — thread the abort signal so a mid-generation disconnect tears + // down the per-round provider calls instead of paying the full cost. signal: request.signal, - }), - }); - result = fallback.result; - workingProviderType = fallback.workingProvider.providerType; - totalTokensSpent = result.tokensUsed ?? 0; - cachedTokensSpent = result.cachedInputTokens ?? 0; - // v1.21.2 (A8) — the no-tools path showed the model the full SNAPSHOT only - // when `includeFullSnapshot` was set; otherwise it shipped the - // grounded-elsewhere pointer with no fresh figures, so there is nothing to - // grade. When figures WERE delivered, the authoritative set is the - // structured snapshot record (incl. the correlations block). - if (includeFullSnapshot) { - noToolsSnapshotPayloads = [snapshot.sections]; - } - } - } catch (err) { - // The provider chain failed outright — no tokens were billed, so refund - // the full reservation before surfacing the error frame. - await reconcileSpend(userId, reservation.reserved, 0, reqDateKey).catch( - () => {}, - ); - if (err instanceof AllProvidersFailedError) { - annotate({ - action: { name: "insights.coach.providerFailed" }, - 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.22 (#89) — per-user response timeout for each tool-round call. + timeoutMs: aiResponseTimeoutMs, }); - } - // 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 - // toast instead of the generic provider-down message, so the - // user understands the limit is transient. - const allRateLimited = - err.attempts.length > 0 && - err.attempts.every((a) => a.httpStatus === 429); - return streamProviderError({ - code: allRateLimited - ? "coach.provider.rate_limited" - : "coach.provider.unavailable", - }); - } - // v1.21.3 — defence in depth. The chain runner wraps every hard provider - // failure in `AllProvidersFailedError`, but a provider client can still - // throw a tagged wire error that reaches here un-wrapped (e.g. a Codex 400 - // raised mid tool-loop on a path the chain runner did not catch). Such an - // error is a PROVIDER failure, not a server bug — surface the same graceful - // `coach.provider.*` frame the chain path uses rather than rethrowing into - // an HTTP 500 (the bug that took the live Coach down for codex users). Only - // a genuinely unexpected error (no upstream tag, no httpStatus) keeps the - // 500 + GlitchTip path so real defects stay visible. - const providerError = classifyBubblingProviderError(err); - if (providerError) { - annotate({ - action: { name: "insights.coach.providerFailed" }, - meta: { - attempts: 1, - firstStatus: providerError.httpStatus, - credentialExpired: providerError.code === "credential_expired", - unwrapped: true, - }, - }); - return streamProviderError({ - code: `coach.provider.${providerError.code}`, - }); - } - throw err; - } - - // v1.18.7 (SENIOR-DEV MEDIUM) — the provider call returned, so its tokens - // were billed regardless of reply quality. Reconcile the reservation - // against the actual count NOW, before any empty/sentinel short-circuit, so - // an empty or sentinel-only reply still records its burned cost (the old - // post-hoc `recordSpend` ran only on the happy path, undercounting these). - // v1.20.0 (F1) — reconcile against the SUMMED tokens across every tool round - // (the loop accumulates them); the no-tools path sums to the single call. - await reconcileSpend( - userId, - reservation.reserved, - totalTokensSpent, - reqDateKey, - cachedTokensSpent, - ).catch(() => { - // Ledger reconcile is best-effort; a failure leaves the conservative - // reservation in place (never an undercount) and never breaks the turn. - }); - - const rawReply = (result.content ?? "").trim(); - if (!rawReply) { - return streamProviderError({ code: "coach.provider.empty" }); - } - - // v1.4.22 — strip the optional `---KEYVALUES---` … `---END---` - // sentinel out of the prose. The stripped prose is what we stream - // to the client and persist; the parsed entries enrich the - // provenance envelope so the UI can render the collapsible - // "Worauf bezieht sich das?" disclosure. - const sentinel = parseKeyValuesSentinel(rawReply); - const proseAfterStrip = sentinel.prose.trim(); - // v1.4.22 W5 reconcile (Code-H1) — when the model emits a - // sentinel-only / malformed reply, `sentinel.prose` is empty after - // stripping. The previous fallback `sentinel.prose.trim() || rawReply` - // surfaced raw `---KEYVALUES---` markers to the user. The empty-prose - // condition signals an unusable provider response: short-circuit to - // the structured `coach.provider.empty` error frame instead of - // streaming the raw sentinel body. - if (!proseAfterStrip) { - annotate({ - action: { name: "coach.keyvalues.parse_failed" }, - meta: { - kept: sentinel.keyValues.length, - reason: "empty_prose_after_strip", - promptVersion: PROMPT_VERSION, - }, - }); - return streamProviderError({ code: "coach.provider.empty" }); - } - // v1.18.1 (Workstream C) — strip the optional `---SUGGEST-REMINDER---` - // block out of the prose-after-keyvalues. The model proposes a cadence; - // the gate decides whether it actually surfaces (module-toggle + opt-out - // + dismissal memory + cooldown + dedup against a live COACH reminder). - // A suppressed proposal leaves the prose unchanged and emits no card. - const suggestParse = parseSuggestReminder(proseAfterStrip); - let replyText = suggestParse.prose.trim() || proseAfterStrip; - - // v1.22 (B2) — strip the optional `---REMEMBER---` block and capture the - // reminder INLINE on this turn (not in the >20-turn memory worker), so a - // casual "remind me about X" in a SHORT chat is no longer lost. The note is - // the user's OWN explicit request → captured silently as `active` (no confirm - // gate); the model acknowledges it in prose. A missing note / invalid `when` - // drops the block (the user never sees the raw marker). Fire-and-forget: the - // capture write must never break the chat turn. - const rememberParse = parseRememberSentinel(replyText, new Date()); - replyText = rememberParse.prose.trim() || replyText; - if (rememberParse.reminder) { - const capture = rememberParse.reminder; - void captureReminderFromSentinel({ - userId, - conversationId: workingConversationId, - parsed: capture, - }).catch(() => { - // Reminder capture is best-effort; never sink the turn. - }); - } else if (rememberParse.malformed) { - annotate({ - action: { name: "coach.reminder.capture_malformed" }, - meta: { conversationId: workingConversationId }, - }); - } - - // v1.22 (F6) — strip the optional `---SUGGEST-ACTION---` block (the - // generalised confirm→apply moat). The model names ONE action from the closed - // allowlist (`checkup.create` / `reminder.note`); the card surfaces additively - // and NOTHING is created until the user taps confirm (the entity is built - // server-side, field-by-field, by `POST /api/coach/suggested-actions`). - const actionParse = parseSuggestAction(replyText); - replyText = actionParse.prose.trim() || replyText; - - // v1.18.10 (HIGH-2) — OUTBOUND safety screen on the assembled assistant - // reply, before persistence and streaming. The inbound `detectRefusal` - // guards the user's message; this guards the model's reply for a - // dose-prescription or a fabricated clinical risk score that slipped past - // the system-prompt GLP-1/grounding contracts. On a trip the turn is - // replaced with a calm, grounded fallback and any reminder suggestion / - // key-value provenance is dropped — the user never sees the unsafe text. - const outbound = screenCoachReply(replyText); - if (outbound.block && outbound.reason) { - replyText = coachOutboundFallback(outbound.reason, locale); - annotate({ - action: { name: "insights.coach.outbound_blocked" }, - meta: { reason: outbound.reason, promptVersion: PROMPT_VERSION }, - }); - await auditLog("insights.coach.outbound_blocked", { - userId, - details: { - conversationId: workingConversationId, - reason: outbound.reason, - }, - }); - } - - // v1.21.0 (P6 / C2-5) — post-hoc numeric verifier on the Coach prose. Cross- - // check every number the model cited against this turn's authoritative figure - // set; an unmatched number (transcription / paraphrase drift) is soft-stripped - // to "[unverified]" and annotated. Cheap, non-blocking, and a no-op when there - // is no authoritative set — the prompt-level grounding rule remains the - // backstop, exactly like the briefing's "no signals → skip". A blocked turn - // already carries canned fallback prose, so skip it. - // - // v1.21.2 (A8) — the authoritative set is the figures the tools returned on the - // tool path, and the SNAPSHOT the model was shown on the no-tools/local path - // (`snapshot.sections`, which already carries the correlations-snapshot block). - // Exactly one is populated per turn; the grading, tolerance, and exemptions are - // identical, so a number the model invents is flagged the same way on both. - const authoritativePayloads = - toolResultPayloads.length > 0 - ? toolResultPayloads - : noToolsSnapshotPayloads; - if (!outbound.block && authoritativePayloads.length > 0) { - const unverified = findUnverifiedCoachNumbers( - replyText, - authoritativePayloads, - ); - if (unverified.length > 0) { - const { prose: corrected, stripped } = stripUnverifiedNumbers( - replyText, - unverified, - ); - replyText = corrected; - annotate({ - action: { name: "coach.prose.number_unverified" }, - meta: { - flagged: unverified.length, - stripped, - // No raw values — just the count + truncated tokens for ops triage. - tokens: unverified.slice(0, 6).map((u) => u.source), - promptVersion: PROMPT_VERSION, - }, - }); - } - } - - // v1.21.0 (NEW-C C-3) — Learn-link post-filter. The prompt instructs the - // model to only link a published `/learn/`, but that is guidance, not - // enforcement: a fabricated `/learn/` would otherwise ship as - // a dead link. Scrub any reference whose slug is not in the catalog (a real - // one is kept verbatim). A blocked turn carries canned fallback prose with no - // links, so skip it. - if (!outbound.block && replyText.includes("/learn/")) { - const scrubbed = scrubUnknownLearnLinks(replyText); - if (scrubbed.dropped.length > 0) { - replyText = scrubbed.text; - annotate({ - action: { name: "coach.learn.link_dropped" }, - meta: { - dropped: scrubbed.dropped.length, - // Truncated slug tokens for ops triage — no user content. - slugs: scrubbed.dropped.slice(0, 6), - promptVersion: PROMPT_VERSION, - }, - }); - } - } - - let surfacedSuggestion: CoachSuggestion | null = null; - if (!outbound.block && suggestParse.cadence) { - const decision = await gateSuggestion({ - prisma, - userId, - cadence: suggestParse.cadence, - prefs: - coachPrefs.reminderSuggestions ?? DEFAULT_REMINDER_SUGGESTION_PREFS, - }); - if (decision.surface) { - const cadence = suggestParse.cadence; - surfacedSuggestion = { - cadenceId: cadence.id, - measurementType: cadence.measurementType, - label: cadence.labelKey, - }; - // Stamp the cooldown anchor (frequency cap) onto the prefs blob. - const nextSuggestionPrefs = { - ...(coachPrefs.reminderSuggestions ?? - DEFAULT_REMINDER_SUGGESTION_PREFS), - lastSuggestedAt: new Date().toISOString(), - }; - void prisma.user - .update({ - where: { id: userId }, - data: { - coachPrefsJson: { - ...coachPrefs, - reminderSuggestions: nextSuggestionPrefs, - }, + result = loop.result; + workingProviderType = loop.workingProviderType; + toolTrace = loop.toolTrace; + toolResultPayloads = (loop.toolResults ?? []).map((r) => r.data); + totalTokensSpent = loop.totalTokens; + cachedTokensSpent = loop.cachedTokens; + } else { + // v1.22 (#89) — the no-tools path (local / Ollama / exo, and any chain + // that includes a non-tool provider) runs through the STREAMING runner so + // the local client emits real tokens as they arrive and the per-idle-gap + // timeout governs. `onDelta` counts streamed chunks for observability; the + // heartbeat keeps the proxy connection warm and the assembled reply is + // returned in full so every guard below still runs on the complete text. + let streamedDeltas = 0; + const fallback = await runStreamingRawCompletionWithFallback({ + userId, + providers: chain, + onDelta: () => { + streamedDeltas += 1; }, - }) - .catch(() => { - // Cooldown stamp is best-effort: a write failure at worst lets a - // second suggestion through sooner, never breaks the chat turn. + // v1.20.0 — the no-tools path still builds one assembled user turn (the + // transcript-flattening preserves the once-per-conversation snapshot + // trick + grounding exactly), so it ships as a single user message. The + // stable persona/grounding rides `system` and is now cache-eligible. + params: singleUserTurn({ + system: systemPrompt, + user: userPrompt, + temperature: AI_BUDGETS.coach.temperature, + maxTokens: AI_BUDGETS.coach.maxTokens, + // v1.20.1 — thread the request's abort signal so a mid-generation + // client disconnect tears the upstream provider call down instead of + // paying the full token cost into a closed connection. + signal: request.signal, + // v1.22 (#89) — per-idle-gap timeout for the streaming local call / + // whole-call timeout for the buffered cloud fallback. + timeoutMs: aiResponseTimeoutMs, + }), }); + annotate({ + action: { name: "coach.stream.deltas" }, + meta: { deltas: streamedDeltas }, + }); + result = fallback.result; + workingProviderType = fallback.workingProvider.providerType; + totalTokensSpent = result.tokensUsed ?? 0; + cachedTokensSpent = result.cachedInputTokens ?? 0; + // v1.21.2 (A8) — the no-tools path showed the model the full SNAPSHOT only + // when `includeFullSnapshot` was set; otherwise it shipped the + // grounded-elsewhere pointer with no fresh figures, so there is nothing to + // grade. When figures WERE delivered, the authoritative set is the + // structured snapshot record (incl. the correlations block). + if (includeFullSnapshot) { + noToolsSnapshotPayloads = [snapshot.sections]; + } + } + } catch (err) { + // The provider chain failed outright — no tokens were billed, so refund + // the full reservation before surfacing the error frame. + await reconcileSpend(userId, reservation.reserved, 0, reqDateKey).catch( + () => {}, + ); + if (err instanceof AllProvidersFailedError) { + annotate({ + action: { name: "insights.coach.providerFailed" }, + 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 { ok: false, 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 + // toast instead of the generic provider-down message, so the + // user understands the limit is transient. + const allRateLimited = + err.attempts.length > 0 && + err.attempts.every((a) => a.httpStatus === 429); + return { + ok: false, + code: allRateLimited + ? "coach.provider.rate_limited" + : "coach.provider.unavailable", + }; + } + // v1.21.3 — defence in depth. The chain runner wraps every hard provider + // failure in `AllProvidersFailedError`, but a provider client can still + // throw a tagged wire error that reaches here un-wrapped (e.g. a Codex 400 + // raised mid tool-loop on a path the chain runner did not catch). Such an + // error is a PROVIDER failure, not a server bug — surface the same graceful + // `coach.provider.*` frame the chain path uses rather than rethrowing into + // an HTTP 500 (the bug that took the live Coach down for codex users). Only + // a genuinely unexpected error (no upstream tag, no httpStatus) keeps the + // 500 + GlitchTip path so real defects stay visible. + const providerError = classifyBubblingProviderError(err); + if (providerError) { + annotate({ + action: { name: "insights.coach.providerFailed" }, + meta: { + attempts: 1, + firstStatus: providerError.httpStatus, + credentialExpired: providerError.code === "credential_expired", + unwrapped: true, + }, + }); + return { ok: false, code: `coach.provider.${providerError.code}` }; + } + throw err; + } + + // v1.18.7 (SENIOR-DEV MEDIUM) — the provider call returned, so its tokens + // were billed regardless of reply quality. Reconcile the reservation + // against the actual count NOW, before any empty/sentinel short-circuit, so + // an empty or sentinel-only reply still records its burned cost (the old + // post-hoc `recordSpend` ran only on the happy path, undercounting these). + // v1.20.0 (F1) — reconcile against the SUMMED tokens across every tool round + // (the loop accumulates them); the no-tools path sums to the single call. + await reconcileSpend( + userId, + reservation.reserved, + totalTokensSpent, + reqDateKey, + cachedTokensSpent, + ).catch(() => { + // Ledger reconcile is best-effort; a failure leaves the conservative + // reservation in place (never an undercount) and never breaks the turn. + }); + + const rawReply = (result.content ?? "").trim(); + if (!rawReply) { + return { ok: false, code: "coach.provider.empty" }; + } + + // v1.4.22 — strip the optional `---KEYVALUES---` … `---END---` + // sentinel out of the prose. The stripped prose is what we stream + // to the client and persist; the parsed entries enrich the + // provenance envelope so the UI can render the collapsible + // "Worauf bezieht sich das?" disclosure. + const sentinel = parseKeyValuesSentinel(rawReply); + const proseAfterStrip = sentinel.prose.trim(); + // v1.4.22 W5 reconcile (Code-H1) — when the model emits a + // sentinel-only / malformed reply, `sentinel.prose` is empty after + // stripping. The previous fallback `sentinel.prose.trim() || rawReply` + // surfaced raw `---KEYVALUES---` markers to the user. The empty-prose + // condition signals an unusable provider response: short-circuit to + // the structured `coach.provider.empty` error frame instead of + // streaming the raw sentinel body. + if (!proseAfterStrip) { annotate({ - action: { name: "coach.reminder.suggested" }, - meta: { cadenceId: cadence.id, metric: cadence.measurementType }, + action: { name: "coach.keyvalues.parse_failed" }, + meta: { + kept: sentinel.keyValues.length, + reason: "empty_prose_after_strip", + promptVersion: PROMPT_VERSION, + }, }); - } else { + return { ok: false, code: "coach.provider.empty" }; + } + // v1.18.1 (Workstream C) — strip the optional `---SUGGEST-REMINDER---` + // block out of the prose-after-keyvalues. The model proposes a cadence; + // the gate decides whether it actually surfaces (module-toggle + opt-out + // + dismissal memory + cooldown + dedup against a live COACH reminder). + // A suppressed proposal leaves the prose unchanged and emits no card. + const suggestParse = parseSuggestReminder(proseAfterStrip); + let replyText = suggestParse.prose.trim() || proseAfterStrip; + + // v1.22 (B2) — strip the optional `---REMEMBER---` block and capture the + // reminder INLINE on this turn (not in the >20-turn memory worker), so a + // casual "remind me about X" in a SHORT chat is no longer lost. The note is + // the user's OWN explicit request → captured silently as `active` (no confirm + // gate); the model acknowledges it in prose. A missing note / invalid `when` + // drops the block (the user never sees the raw marker). Fire-and-forget: the + // capture write must never break the chat turn. + const rememberParse = parseRememberSentinel(replyText, new Date()); + replyText = rememberParse.prose.trim() || replyText; + if (rememberParse.reminder) { + const capture = rememberParse.reminder; + void captureReminderFromSentinel({ + userId, + conversationId: workingConversationId, + parsed: capture, + }).catch(() => { + // Reminder capture is best-effort; never sink the turn. + }); + } else if (rememberParse.malformed) { annotate({ - action: { name: "coach.reminder.suppressed" }, - meta: { cadenceId: suggestParse.cadence.id, reason: decision.reason }, + action: { name: "coach.reminder.capture_malformed" }, + meta: { conversationId: workingConversationId }, }); } - } - // v1.22 (F6) — surface the confirm-card action when the turn was not blocked. - // Additive: the prose already stands alone; the card only offers the one-tap - // confirm. Nothing is created server-side until the user taps it. - let surfacedAction: CoachSuggestedAction | null = null; - if (!outbound.block && actionParse.action) { - surfacedAction = actionParse.action; - annotate({ - action: { name: "coach.action.suggested" }, - meta: { actionType: actionParse.action.actionType }, - }); - } - const enrichedProvenance: typeof snapshot.provenance = { - ...snapshot.provenance, - ...(surfacedAction ? { suggestedAction: surfacedAction } : {}), - // v1.18.10 (HIGH-2) — a blocked turn carries the fallback prose, so the - // key-values from the discarded reply must not ride along as provenance. - ...(!outbound.block && sentinel.keyValues.length > 0 - ? { keyValues: sentinel.keyValues } - : {}), - ...(surfacedSuggestion ? { suggestion: surfacedSuggestion } : {}), - // v1.20.0 (F1) — persist the retrieval-tool trace (which tools ran + - // whether each found data) so a reload can show "what I looked at" and the - // audit can replay grounding. Metadata only. - ...(toolTrace.length > 0 - ? { - toolCalls: toolTrace.map((t) => ({ - name: t.name, - present: t.present, - })), - } - : {}), - }; - if (sentinel.malformed) { - // Graceful degrade: log so ops can spot a provider whose - // sentinel format has drifted, but pass the prose through - // unchanged. v1.4.23 H1 — split the annotation: - // - parse_partial: at least one row parsed AND at least one - // row failed (mixed-format drift on a single reply) - // - parse_failed: the whole block was unusable - // Both annotations carry the per-line `reasons` array so an ops - // dashboard can attribute the failure cause without re-running - // the parser. - const reasons = sentinel.malformedEntries.map((entry) => entry.reason); - const annotationName = - sentinel.keyValues.length > 0 && sentinel.malformedEntries.length > 0 - ? "coach.keyvalues.parse_partial" - : "coach.keyvalues.parse_failed"; + // v1.22 (F6) — strip the optional `---SUGGEST-ACTION---` block (the + // generalised confirm→apply moat). The model names ONE action from the closed + // allowlist (`checkup.create` / `reminder.note`); the card surfaces additively + // and NOTHING is created until the user taps confirm (the entity is built + // server-side, field-by-field, by `POST /api/coach/suggested-actions`). + const actionParse = parseSuggestAction(replyText); + replyText = actionParse.prose.trim() || replyText; + + // v1.18.10 (HIGH-2) — OUTBOUND safety screen on the assembled assistant + // reply, before persistence and streaming. The inbound `detectRefusal` + // guards the user's message; this guards the model's reply for a + // dose-prescription or a fabricated clinical risk score that slipped past + // the system-prompt GLP-1/grounding contracts. On a trip the turn is + // replaced with a calm, grounded fallback and any reminder suggestion / + // key-value provenance is dropped — the user never sees the unsafe text. + const outbound = screenCoachReply(replyText); + if (outbound.block && outbound.reason) { + replyText = coachOutboundFallback(outbound.reason, locale); + annotate({ + action: { name: "insights.coach.outbound_blocked" }, + meta: { reason: outbound.reason, promptVersion: PROMPT_VERSION }, + }); + await auditLog("insights.coach.outbound_blocked", { + userId, + details: { + conversationId: workingConversationId, + reason: outbound.reason, + }, + }); + } + + // v1.21.0 (P6 / C2-5) — post-hoc numeric verifier on the Coach prose. Cross- + // check every number the model cited against this turn's authoritative figure + // set; an unmatched number (transcription / paraphrase drift) is soft-stripped + // to "[unverified]" and annotated. Cheap, non-blocking, and a no-op when there + // is no authoritative set — the prompt-level grounding rule remains the + // backstop, exactly like the briefing's "no signals → skip". A blocked turn + // already carries canned fallback prose, so skip it. + // + // v1.21.2 (A8) — the authoritative set is the figures the tools returned on the + // tool path, and the SNAPSHOT the model was shown on the no-tools/local path + // (`snapshot.sections`, which already carries the correlations-snapshot block). + // Exactly one is populated per turn; the grading, tolerance, and exemptions are + // identical, so a number the model invents is flagged the same way on both. + const authoritativePayloads = + toolResultPayloads.length > 0 + ? toolResultPayloads + : noToolsSnapshotPayloads; + if (!outbound.block && authoritativePayloads.length > 0) { + const unverified = findUnverifiedCoachNumbers( + replyText, + authoritativePayloads, + ); + if (unverified.length > 0) { + const { prose: corrected, stripped } = stripUnverifiedNumbers( + replyText, + unverified, + ); + replyText = corrected; + annotate({ + action: { name: "coach.prose.number_unverified" }, + meta: { + flagged: unverified.length, + stripped, + // No raw values — just the count + truncated tokens for ops triage. + tokens: unverified.slice(0, 6).map((u) => u.source), + promptVersion: PROMPT_VERSION, + }, + }); + } + } + + // v1.21.0 (NEW-C C-3) — Learn-link post-filter. The prompt instructs the + // model to only link a published `/learn/`, but that is guidance, not + // enforcement: a fabricated `/learn/` would otherwise ship as + // a dead link. Scrub any reference whose slug is not in the catalog (a real + // one is kept verbatim). A blocked turn carries canned fallback prose with no + // links, so skip it. + if (!outbound.block && replyText.includes("/learn/")) { + const scrubbed = scrubUnknownLearnLinks(replyText); + if (scrubbed.dropped.length > 0) { + replyText = scrubbed.text; + annotate({ + action: { name: "coach.learn.link_dropped" }, + meta: { + dropped: scrubbed.dropped.length, + // Truncated slug tokens for ops triage — no user content. + slugs: scrubbed.dropped.slice(0, 6), + promptVersion: PROMPT_VERSION, + }, + }); + } + } + + let surfacedSuggestion: CoachSuggestion | null = null; + if (!outbound.block && suggestParse.cadence) { + const decision = await gateSuggestion({ + prisma, + userId, + cadence: suggestParse.cadence, + prefs: + coachPrefs.reminderSuggestions ?? DEFAULT_REMINDER_SUGGESTION_PREFS, + }); + if (decision.surface) { + const cadence = suggestParse.cadence; + surfacedSuggestion = { + cadenceId: cadence.id, + measurementType: cadence.measurementType, + label: cadence.labelKey, + }; + // Stamp the cooldown anchor (frequency cap) onto the prefs blob. + const nextSuggestionPrefs = { + ...(coachPrefs.reminderSuggestions ?? + DEFAULT_REMINDER_SUGGESTION_PREFS), + lastSuggestedAt: new Date().toISOString(), + }; + void prisma.user + .update({ + where: { id: userId }, + data: { + coachPrefsJson: { + ...coachPrefs, + reminderSuggestions: nextSuggestionPrefs, + }, + }, + }) + .catch(() => { + // Cooldown stamp is best-effort: a write failure at worst lets a + // second suggestion through sooner, never breaks the chat turn. + }); + annotate({ + action: { name: "coach.reminder.suggested" }, + meta: { cadenceId: cadence.id, metric: cadence.measurementType }, + }); + } else { + annotate({ + action: { name: "coach.reminder.suppressed" }, + meta: { cadenceId: suggestParse.cadence.id, reason: decision.reason }, + }); + } + } + // v1.22 (F6) — surface the confirm-card action when the turn was not blocked. + // Additive: the prose already stands alone; the card only offers the one-tap + // confirm. Nothing is created server-side until the user taps it. + let surfacedAction: CoachSuggestedAction | null = null; + if (!outbound.block && actionParse.action) { + surfacedAction = actionParse.action; + annotate({ + action: { name: "coach.action.suggested" }, + meta: { actionType: actionParse.action.actionType }, + }); + } + + const enrichedProvenance: typeof snapshot.provenance = { + ...snapshot.provenance, + ...(surfacedAction ? { suggestedAction: surfacedAction } : {}), + // v1.18.10 (HIGH-2) — a blocked turn carries the fallback prose, so the + // key-values from the discarded reply must not ride along as provenance. + ...(!outbound.block && sentinel.keyValues.length > 0 + ? { keyValues: sentinel.keyValues } + : {}), + ...(surfacedSuggestion ? { suggestion: surfacedSuggestion } : {}), + // v1.20.0 (F1) — persist the retrieval-tool trace (which tools ran + + // whether each found data) so a reload can show "what I looked at" and the + // audit can replay grounding. Metadata only. + ...(toolTrace.length > 0 + ? { + toolCalls: toolTrace.map((t) => ({ + name: t.name, + present: t.present, + })), + } + : {}), + }; + if (sentinel.malformed) { + // Graceful degrade: log so ops can spot a provider whose + // sentinel format has drifted, but pass the prose through + // unchanged. v1.4.23 H1 — split the annotation: + // - parse_partial: at least one row parsed AND at least one + // row failed (mixed-format drift on a single reply) + // - parse_failed: the whole block was unusable + // Both annotations carry the per-line `reasons` array so an ops + // dashboard can attribute the failure cause without re-running + // the parser. + const reasons = sentinel.malformedEntries.map((entry) => entry.reason); + const annotationName = + sentinel.keyValues.length > 0 && sentinel.malformedEntries.length > 0 + ? "coach.keyvalues.parse_partial" + : "coach.keyvalues.parse_failed"; + annotate({ + action: { name: annotationName }, + meta: { + kept: sentinel.keyValues.length, + malformedCount: sentinel.malformedEntries.length, + reasons, + promptVersion: PROMPT_VERSION, + }, + }); + } + + // Persist the assistant message BEFORE we begin streaming; if the + // client disconnects we still have the canonical row. + const assistantMessage = await appendMessage({ + conversationId: workingConversationId, + role: "assistant", + content: replyText, + metricSource: enrichedProvenance, + providerType: workingProviderType, + promptVersion: PROMPT_VERSION, + // v1.18.9 — persist the per-turn token count + model so the quiet + // token footer survives a conversation reload. The live turn paints + // from the `done.usage` SSE frame below; reloads read these columns. + // v1.20.0 (F1) — the summed cost across every tool round, so the footer + // reflects the true turn cost on the tool path too. + tokensUsed: totalTokensSpent || null, + model: result.model ?? null, + }); + + // v1.18.7 — the day's spend was already reconciled against the + // reservation immediately after the provider returned (above), so there is + // no post-persistence ledger bump here. The reservation guarantees the + // tokens are counted even if persistence or streaming fails afterwards. + annotate({ - action: { name: annotationName }, + action: { name: "insights.coach.replied" }, meta: { - kept: sentinel.keyValues.length, - malformedCount: sentinel.malformedEntries.length, - reasons, + provider: workingProviderType, + // v1.20.0 (F1) — summed tokens across every tool round (the loop) or the + // single call (no-tools path), so the dashboards see the true turn cost. + tokens: totalTokensSpent, promptVersion: PROMPT_VERSION, + conversationId: workingConversationId, + historyTurns: window.length, + // v1.20.0 (F1) — whether this turn ran the tool-retrieval path, and how + // many tools it fetched, so the dashboards can correlate the token delta + // with the new path vs the legacy snapshot path. + toolMode, + toolsCalled: toolTrace.length, + // v1.19.1 (C4) — whether the full SNAPSHOT block rode this turn (the + // expensive prefix) vs the cheap pointer. On the tool path the snapshot + // never rides the prompt, so this is the legacy-path signal only. + snapshotSent: !toolMode && includeFullSnapshot, + promptChars: userPrompt.length, + // v1.7.0 — count of provenance metrics the snapshot surfaced + // this turn (a proxy for cluster breadth) so the dashboards can + // correlate reply shape with cluster activation. + clusterCount: snapshot.provenance.metrics.length, }, }); - } - // Persist the assistant message BEFORE we begin streaming; if the - // client disconnects we still have the canonical row. - const assistantMessage = await appendMessage({ - conversationId: workingConversationId, - role: "assistant", - content: replyText, - metricSource: enrichedProvenance, - providerType: workingProviderType, - promptVersion: PROMPT_VERSION, - // v1.18.9 — persist the per-turn token count + model so the quiet - // token footer survives a conversation reload. The live turn paints - // from the `done.usage` SSE frame below; reloads read these columns. - // v1.20.0 (F1) — the summed cost across every tool round, so the footer - // reflects the true turn cost on the tool path too. - tokensUsed: totalTokensSpent || null, - model: result.model ?? null, - }); - - // v1.18.7 — the day's spend was already reconciled against the - // reservation immediately after the provider returned (above), so there is - // no post-persistence ledger bump here. The reservation guarantees the - // tokens are counted even if persistence or streaming fails afterwards. - - annotate({ - action: { name: "insights.coach.replied" }, - meta: { - provider: workingProviderType, - // v1.20.0 (F1) — summed tokens across every tool round (the loop) or the - // single call (no-tools path), so the dashboards see the true turn cost. - tokens: totalTokensSpent, - promptVersion: PROMPT_VERSION, - conversationId: workingConversationId, - historyTurns: window.length, - // v1.20.0 (F1) — whether this turn ran the tool-retrieval path, and how - // many tools it fetched, so the dashboards can correlate the token delta - // with the new path vs the legacy snapshot path. - toolMode, - toolsCalled: toolTrace.length, - // v1.19.1 (C4) — whether the full SNAPSHOT block rode this turn (the - // expensive prefix) vs the cheap pointer. On the tool path the snapshot - // never rides the prompt, so this is the legacy-path signal only. - snapshotSent: !toolMode && includeFullSnapshot, - promptChars: userPrompt.length, - // v1.7.0 — count of provenance metrics the snapshot surfaced - // this turn (a proxy for cluster breadth) so the dashboards can - // correlate reply shape with cluster activation. - clusterCount: snapshot.provenance.metrics.length, - }, - }); + // produceReply success — hand the fully-guarded reply + provenance back to + // the heartbeat-fronted stream below for client-visible emission. + return { + ok: true, + replyText, + provenance: enrichedProvenance, + suggestion: surfacedSuggestion, + action: surfacedAction, + messageId: assistantMessage.id, + totalTokens: totalTokensSpent, + model: result.model ?? null, + }; + } // end produceReply // ── Stream the body to the client ──────────────────────────── - // v1.12.0 — yield to the event loop between token frames so each one - // flushes as its own network chunk. The provider clients return the - // full reply in one shot; without the yield the whole tokenised body - // was enqueued synchronously inside `start()` and the runtime - // coalesced every frame into a single read, so the client painted the - // answer all at once despite the per-token render path in `use-coach`. - // A zero-delay yield is enough to land each frame on its own tick — - // the visible cadence reads ChatGPT/Claude-style without a contrived - // sleep. The refusal + error paths stay single-frame (nothing to - // pace). + // v1.22 (#89) — the provider call + every guard + persistence run INSIDE the + // stream (via `produceReply`) so the HTTP headers + a keepalive heartbeat + // flush immediately and keep the reverse proxy from idle-dropping the + // connection while a slow local backend is still loading the model / + // generating. Only once the FULLY-GUARDED reply is ready do client-visible + // token frames go out. + // + // v1.12.0 — yield to the event loop between token frames so each one flushes + // as its own network chunk; the visible cadence reads ChatGPT/Claude-style. const stream = createSseStream(async (controller) => { - for (const tok of tokeniseForStreaming(replyText)) { - // v1.18.10 (A-2) — stop tokenising the moment the client disconnects; - // the cancel handler flips this signal, so we don't pace frames into a - // closed connection. + // Keepalive: an SSE comment frame (`: ka`) the client parser ignores, + // flushed on an interval through the pre-first-token (prompt-processing) + // and generation phases. + const heartbeat = setInterval(() => { + controller.enqueue(HEARTBEAT_FRAME); + }, HEARTBEAT_MS); + + let outcome: ReplyOutcome; + try { + outcome = await produceReply(); + } catch (err) { + clearInterval(heartbeat); + // The stream is already open, so we cannot 500: a tagged provider error + // surfaces a graceful `coach.provider.*` frame; anything else degrades to + // a generic unavailable frame. A genuine defect is still annotated. + const providerError = classifyBubblingProviderError(err); + annotate({ + action: { name: "insights.coach.streamError" }, + meta: { + unwrapped: Boolean(providerError), + firstStatus: providerError?.httpStatus ?? null, + }, + }); + if (!controller.signal.aborted) { + const code = providerError + ? `coach.provider.${providerError.code}` + : "coach.provider.unavailable"; + controller.enqueue(encodeFrame({ type: "error", code, message: code })); + } + return; + } + clearInterval(heartbeat); + + if (!outcome.ok) { + if (!controller.signal.aborted) { + controller.enqueue( + encodeFrame({ + type: "error", + code: outcome.code, + message: outcome.code, + }), + ); + } + return; + } + + for (const tok of tokeniseForStreaming(outcome.replyText)) { + // v1.18.10 (A-2) — stop tokenising the moment the client disconnects. if (controller.signal.aborted) return; controller.enqueue(encodeFrame({ type: "token", token: tok })); await flushTick(); } if (controller.signal.aborted) return; controller.enqueue( - encodeFrame({ - type: "provenance", - metricSource: enrichedProvenance, - }), + encodeFrame({ type: "provenance", metricSource: outcome.provenance }), ); - // v1.18.1 (Workstream C) — additive `suggestion` frame. Older clients - // ignore it; newer ones render the one-tap action card. - if (surfacedSuggestion) { + // v1.18.1 (Workstream C) — additive `suggestion` frame. + if (outcome.suggestion) { controller.enqueue( - encodeFrame({ type: "suggestion", suggestion: surfacedSuggestion }), + encodeFrame({ type: "suggestion", suggestion: outcome.suggestion }), ); } - // v1.22 (F6) — additive `suggestedAction` frame. Older clients ignore it; - // newer ones render the generic one-tap confirm card. - if (surfacedAction) { + // v1.22 (F6) — additive `suggestedAction` frame. + if (outcome.action) { controller.enqueue( encodeFrame({ type: "suggestedAction", - suggestedAction: surfacedAction, + suggestedAction: outcome.action, }), ); } - // v1.18.9 — additive `usage` envelope on the `done` frame so the client - // can paint the quiet per-message token footer the instant the stream - // closes. Server-authoritative: the client renders these numbers, never - // recomputes them. Older clients drop the unknown key. + // v1.18.9 — additive `usage` envelope on the `done` frame. controller.enqueue( encodeFrame({ type: "done", conversationId: workingConversationId, - messageId: assistantMessage.id, + messageId: outcome.messageId, usage: { - // v1.20.0 (F1) — summed across every tool round. - totalTokens: totalTokensSpent || null, - model: result.model ?? null, + totalTokens: outcome.totalTokens || null, + model: outcome.model, }, }), ); diff --git a/src/app/api/user/ai-provider/__tests__/route.test.ts b/src/app/api/user/ai-provider/__tests__/route.test.ts index 0c58d877b..6f2250e96 100644 --- a/src/app/api/user/ai-provider/__tests__/route.test.ts +++ b/src/app/api/user/ai-provider/__tests__/route.test.ts @@ -18,7 +18,7 @@ vi.mock("@/lib/api-handler", () => ({ vi.mock("@/lib/db", () => ({ prisma: { - user: { findUnique: vi.fn() }, + user: { findUnique: vi.fn(), update: vi.fn() }, }, })); @@ -35,14 +35,26 @@ vi.mock("@/lib/validations/notifications", () => ({ isPublicUrl: vi.fn(() => true), })); +vi.mock("@/lib/ai/local-host-allowlist", () => ({ + isLocalAiHostAllowed: vi.fn(() => false), +})); + vi.mock("@/lib/logging/context", () => ({ annotate: vi.fn(), })); -import { GET } from "../route"; +import { GET, PATCH } from "../route"; import { prisma } from "@/lib/db"; import { resolveProviderAvailability } from "@/lib/ai/provider"; +function patchRequest(body: unknown): Request { + return new Request("http://localhost/api/user/ai-provider", { + method: "PATCH", + headers: { "content-type": "application/json" }, + body: JSON.stringify(body), + }); +} + interface AiProviderResponse { data: { provider: string | null; @@ -126,4 +138,101 @@ describe("GET /api/user/ai-provider availability", () => { expect(body.data.aiAvailable).toBe(false); expect(body.data.managedBy).toBeNull(); }); + + it("returns the v1.22 timeout + OCR fields", async () => { + vi.mocked(prisma.user.findUnique).mockResolvedValue({ + aiProvider: "OPENAI", + aiModel: "gpt-4o", + aiBaseUrl: null, + aiAnthropicKeyEncrypted: null, + aiLocalKeyEncrypted: null, + aiOpenaiKeyEncrypted: null, + aiResponseTimeoutSeconds: 240, + aiOcrEnabled: true, + aiOcrProvider: "ANTHROPIC", + aiOcrModel: "claude-sonnet-4-6", + aiOcrBaseUrl: null, + aiOcrKeyEncrypted: "enc-ocr-1234", + } as never); + vi.mocked(resolveProviderAvailability).mockResolvedValue({ + aiAvailable: true, + managedBy: "user", + }); + + const res = await (GET as () => Promise)(); + const body = (await res.json()) as { + data: { + responseTimeoutSeconds: number | null; + ocrEnabled: boolean; + ocrProvider: string | null; + hasOcrKey: boolean; + ocrKeyPreview: string | null; + }; + }; + + expect(body.data.responseTimeoutSeconds).toBe(240); + expect(body.data.ocrEnabled).toBe(true); + expect(body.data.ocrProvider).toBe("ANTHROPIC"); + expect(body.data.hasOcrKey).toBe(true); + // Preview is `...` + last 4 of the decrypted value (decrypt → `dec:...`). + expect(body.data.ocrKeyPreview).toMatch(/^\.\.\./); + }); +}); + +describe("PATCH /api/user/ai-provider — v1.22 fields", () => { + const patch = PATCH as (req: Request) => Promise; + + it("persists a valid response timeout", async () => { + const res = await patch(patchRequest({ responseTimeoutSeconds: 120 })); + expect(res.status).toBe(200); + expect(vi.mocked(prisma.user.update)).toHaveBeenCalledWith( + expect.objectContaining({ + data: expect.objectContaining({ aiResponseTimeoutSeconds: 120 }), + }), + ); + }); + + it("rejects an out-of-range response timeout", async () => { + await expect( + patch(patchRequest({ responseTimeoutSeconds: 5 })), + ).rejects.toThrow(/between 10 and 600/); + expect(vi.mocked(prisma.user.update)).not.toHaveBeenCalled(); + }); + + it("clears the timeout when null is sent", async () => { + await patch(patchRequest({ responseTimeoutSeconds: null })); + expect(vi.mocked(prisma.user.update)).toHaveBeenCalledWith( + expect.objectContaining({ + data: expect.objectContaining({ aiResponseTimeoutSeconds: null }), + }), + ); + }); + + it("persists OCR provider config and encrypts the key", async () => { + const res = await patch( + patchRequest({ + ocrEnabled: true, + ocrProvider: "OPENAI", + ocrModel: "gpt-4o", + ocrKey: "sk-ocr-secret", + }), + ); + expect(res.status).toBe(200); + expect(vi.mocked(prisma.user.update)).toHaveBeenCalledWith( + expect.objectContaining({ + data: expect.objectContaining({ + aiOcrEnabled: true, + aiOcrProvider: "OPENAI", + aiOcrModel: "gpt-4o", + aiOcrKeyEncrypted: "enc:sk-ocr-secret", + }), + }), + ); + }); + + it("rejects an invalid OCR provider", async () => { + await expect( + patch(patchRequest({ ocrProvider: "CHATGPT_OAUTH" })), + ).rejects.toThrow(/Invalid OCR provider/); + }); }); diff --git a/src/app/api/user/ai-provider/route.ts b/src/app/api/user/ai-provider/route.ts index 885a66a32..2fa312548 100644 --- a/src/app/api/user/ai-provider/route.ts +++ b/src/app/api/user/ai-provider/route.ts @@ -11,6 +11,9 @@ import { annotate } from "@/lib/logging/context"; export const dynamic = "force-dynamic"; const ALLOWED = new Set(["OPENAI", "ANTHROPIC", "LOCAL", "CHATGPT_OAUTH"]); +// v1.22 (#90) — the dedicated document-scan provider has no OAuth path, so the +// allowlist is the three keyed/self-hosted providers only (no CHATGPT_OAUTH). +const OCR_PROVIDERS = new Set(["OPENAI", "ANTHROPIC", "LOCAL"]); export const GET = apiHandler(async () => { const { user } = await requireAuth(); @@ -25,6 +28,13 @@ export const GET = apiHandler(async () => { aiAnthropicKeyEncrypted: true, aiLocalKeyEncrypted: true, aiOpenaiKeyEncrypted: true, + // v1.22 (#89 + #90) + aiResponseTimeoutSeconds: true, + aiOcrEnabled: true, + aiOcrProvider: true, + aiOcrModel: true, + aiOcrBaseUrl: true, + aiOcrKeyEncrypted: true, }, }); @@ -50,6 +60,18 @@ export const GET = apiHandler(async () => { openaiKeyPreview: u?.aiOpenaiKeyEncrypted ? `...${decrypt(u.aiOpenaiKeyEncrypted).slice(-4)}` : null, + // v1.22 (#89) — per-user response timeout, in seconds (null = default). + responseTimeoutSeconds: u?.aiResponseTimeoutSeconds ?? null, + // v1.22 (#90) — dedicated document-scan (Lab-OCR) provider config. Key + // material is presence + 4-char preview only, like every other provider. + ocrEnabled: Boolean(u?.aiOcrEnabled), + ocrProvider: u?.aiOcrProvider ?? null, + ocrModel: u?.aiOcrModel ?? null, + ocrBaseUrl: u?.aiOcrBaseUrl ?? null, + hasOcrKey: Boolean(u?.aiOcrKeyEncrypted), + ocrKeyPreview: u?.aiOcrKeyEncrypted + ? `...${decrypt(u.aiOcrKeyEncrypted).slice(-4)}` + : null, }); }); @@ -132,6 +154,87 @@ export const PATCH = apiHandler(async (request: NextRequest) => { } } + // ── v1.22 (#89) — response timeout (seconds) ────────────────── + if (body.responseTimeoutSeconds !== undefined) { + if (body.responseTimeoutSeconds === null) { + updates.aiResponseTimeoutSeconds = null; + } else if ( + typeof body.responseTimeoutSeconds === "number" && + Number.isInteger(body.responseTimeoutSeconds) + ) { + // Sane bounds: 10 s floor (a stricter value just trips the timeout) and + // 600 s ceiling (a generous cap for a slow self-hosted backend). + if ( + body.responseTimeoutSeconds < 10 || + body.responseTimeoutSeconds > 600 + ) { + throw new HttpError( + 422, + "Response timeout must be between 10 and 600 seconds", + ); + } + updates.aiResponseTimeoutSeconds = body.responseTimeoutSeconds; + } else { + throw new HttpError(422, "Invalid response timeout"); + } + } + + // ── v1.22 (#90) — dedicated document-scan (Lab-OCR) provider ── + if (body.ocrEnabled !== undefined) { + if (typeof body.ocrEnabled === "boolean") { + updates.aiOcrEnabled = body.ocrEnabled; + } else { + throw new HttpError(422, "Invalid ocrEnabled"); + } + } + + if (body.ocrProvider !== undefined) { + if (body.ocrProvider === null || body.ocrProvider === "") { + updates.aiOcrProvider = null; + } else if ( + typeof body.ocrProvider === "string" && + OCR_PROVIDERS.has(body.ocrProvider) + ) { + updates.aiOcrProvider = body.ocrProvider; + } else { + throw new HttpError(422, "Invalid OCR provider"); + } + } + + if (body.ocrModel !== undefined) { + if (body.ocrModel === null || body.ocrModel === "") { + updates.aiOcrModel = null; + } else if (typeof body.ocrModel === "string") { + updates.aiOcrModel = body.ocrModel.trim() || null; + } + } + + if (body.ocrBaseUrl !== undefined) { + if (body.ocrBaseUrl === null || body.ocrBaseUrl === "") { + updates.aiOcrBaseUrl = null; + } else if (typeof body.ocrBaseUrl === "string") { + const trimmed = body.ocrBaseUrl.trim(); + // Same SSRF guard as the main LOCAL base URL: reject private/internal + // hosts unless the operator opted in via ALLOW_LOCAL_AI_PRIVATE_HOSTS. + const allowPrivate = isLocalAiHostAllowed(trimmed); + if (!allowPrivate && !isPublicUrl(trimmed)) { + return apiError( + "OCR base URL points to an internal/private host. Ops must allow it on this instance via ALLOW_LOCAL_AI_PRIVATE_HOSTS.", + 422, + ); + } + updates.aiOcrBaseUrl = trimmed; + } + } + + if (body.ocrKey !== undefined) { + if (body.ocrKey === null || body.ocrKey === "") { + updates.aiOcrKeyEncrypted = null; + } else if (typeof body.ocrKey === "string") { + updates.aiOcrKeyEncrypted = encrypt(body.ocrKey.trim()); + } + } + if (Object.keys(updates).length === 0) { throw new HttpError(422, "No valid fields"); } diff --git a/src/components/settings/ai/ai-insights-card.tsx b/src/components/settings/ai/ai-insights-card.tsx index d761bc7da..2768c96e6 100644 --- a/src/components/settings/ai/ai-insights-card.tsx +++ b/src/components/settings/ai/ai-insights-card.tsx @@ -31,9 +31,11 @@ import { queryKeys } from "@/lib/query-keys"; import { AdminOpenAIProviderForm } from "./admin-openai-provider-form"; import { AnthropicProviderForm } from "./anthropic-provider-form"; import { CodexProviderForm } from "./codex-provider-form"; +import { DocumentScanCard } from "./document-scan-card"; import { FallbackChainCard } from "./fallback-chain-card"; import { LocalProviderForm } from "./local-provider-form"; import { OpenAIProviderForm } from "./openai-provider-form"; +import { ResponseTimeoutCard } from "./response-timeout-card"; import { RuntimeActionsRow } from "./runtime-actions-row"; import { PROVIDER_TYPES, @@ -171,6 +173,13 @@ export function AiInsightsCard({ onSelect={pickProvider} /> + {/* v1.22 (#89) — per-user response timeout (mainly for slow + local/self-hosted backends). */} + + + {/* v1.22 (#90) — dedicated document-scan (Lab-OCR) provider. */} + + ("OPENAI"); + const [model, setModel] = useState(""); + const [baseUrl, setBaseUrl] = useState(""); + const [apiKey, setApiKey] = useState(""); + const [msg, setMsg] = useState(null); + const [ok, setOk] = useState(false); + + const seededKey = + userProvider != null + ? `${userProvider.ocrEnabled}|${userProvider.ocrProvider ?? ""}|${userProvider.ocrModel ?? ""}|${userProvider.ocrBaseUrl ?? ""}` + : null; + const [previousSeed, setPreviousSeed] = useState(null); + if (seededKey != null && seededKey !== previousSeed) { + setPreviousSeed(seededKey); + setEnabled(userProvider?.ocrEnabled ?? false); + setProvider(userProvider?.ocrProvider ?? "OPENAI"); + setModel(userProvider?.ocrModel ?? ""); + setBaseUrl(userProvider?.ocrBaseUrl ?? ""); + } + + // Vision-capability hint: warn when a model is chosen that can't read images + // (local is trust-by-default; the warning surfaces only for cloud providers). + const visionOk = + !enabled || + model.trim() === "" || + supportsVisionForConfig(toVisionType(provider), model.trim()); + + const saveMutation = useMutation({ + mutationFn: async () => { + const body: Record = { + ocrEnabled: enabled, + ocrProvider: provider, + ocrModel: model.trim() || null, + ocrBaseUrl: provider === "LOCAL" ? baseUrl.trim() || null : null, + }; + if (apiKey.trim()) body.ocrKey = apiKey.trim(); + await apiPatch("/api/user/ai-provider", body); + }, + onSuccess: () => { + setOk(true); + setMsg(t("settings.ai.saved")); + setApiKey(""); + queryClient.invalidateQueries({ queryKey: queryKeys.userAiProvider() }); + }, + onError: (e) => { + setOk(false); + setMsg(e instanceof Error ? e.message : t("settings.ai.errorGeneric")); + }, + }); + + return ( +
+
+
+

{t("settings.ai.ocrHeading")}

+

+ {t("settings.ai.ocrBody")} +

+
+ +
+ + {enabled && ( +
+
+ + setProvider(e.target.value)} + className="mt-1 sm:max-w-md" + > + {OCR_PROVIDER_TYPES.map((p) => ( + + ))} + +
+ +
+ + setModel(e.target.value)} + placeholder={t("settings.ai.ocrModelPlaceholder")} + className="mt-1" + /> + {!visionOk && ( +

+ {t("settings.ai.ocrVisionHint")} +

+ )} +
+ + {provider === "LOCAL" && ( +
+ + setBaseUrl(e.target.value)} + placeholder="http://localhost:11434/v1" + className="mt-1" + /> +
+ )} + +
+ + setApiKey(e.target.value)} + placeholder={ + userProvider?.hasOcrKey ? t("settings.ai.savedShort") : "" + } + className="mt-1" + /> +
+
+ )} + +
+ +
+ + {msg && ( +

+ {msg} +

+ )} +
+ ); +} diff --git a/src/components/settings/ai/response-timeout-card.tsx b/src/components/settings/ai/response-timeout-card.tsx new file mode 100644 index 000000000..c87f64e2d --- /dev/null +++ b/src/components/settings/ai/response-timeout-card.tsx @@ -0,0 +1,126 @@ +"use client"; + +/* ──────────────────────────────────────────────────────────────── + * v1.22 (#89) — Response timeout (seconds). + * + * A per-user upstream timeout for AI generation, threaded onto the provider + * call (`CompletionParams.timeoutMs`). Surfaced here mainly for local / + * self-hosted backends: an MLX/exo server can take >60 s on the first request + * while it loads the model, which the legacy 60 s default timed out before the + * first token landed. Empty = the generous built-in default (~180 s). + * ──────────────────────────────────────────────────────────────── */ + +import { useState } from "react"; +import { useMutation, useQueryClient } from "@tanstack/react-query"; +import { Loader2, Save } from "lucide-react"; + +import { Button } from "@/components/ui/button"; +import { Input } from "@/components/ui/input"; +import { Label } from "@/components/ui/label"; +import { apiPatch } from "@/lib/api/api-fetch"; +import { useTranslations } from "@/lib/i18n/context"; +import { queryKeys } from "@/lib/query-keys"; + +import type { UserAIProvider } from "./shared"; + +export function ResponseTimeoutCard({ + userProvider, +}: { + userProvider: UserAIProvider | null | undefined; +}) { + const { t } = useTranslations(); + const queryClient = useQueryClient(); + + const [value, setValue] = useState(""); + const [msg, setMsg] = useState(null); + const [ok, setOk] = useState(false); + + // Seed from the persisted value once it arrives (seed-on-data pattern). + const seededKey = + userProvider != null + ? `${userProvider.responseTimeoutSeconds ?? ""}` + : null; + const [previousSeed, setPreviousSeed] = useState(null); + if (seededKey != null && seededKey !== previousSeed) { + setPreviousSeed(seededKey); + setValue( + userProvider?.responseTimeoutSeconds != null + ? String(userProvider.responseTimeoutSeconds) + : "", + ); + } + + const saveMutation = useMutation({ + mutationFn: async () => { + const trimmed = value.trim(); + const responseTimeoutSeconds = trimmed === "" ? null : Number(trimmed); + if ( + responseTimeoutSeconds !== null && + (!Number.isInteger(responseTimeoutSeconds) || + responseTimeoutSeconds < 10 || + responseTimeoutSeconds > 600) + ) { + throw new Error(t("settings.ai.responseTimeoutInvalid")); + } + await apiPatch("/api/user/ai-provider", { responseTimeoutSeconds }); + }, + onSuccess: () => { + setOk(true); + setMsg(t("settings.ai.saved")); + queryClient.invalidateQueries({ queryKey: queryKeys.userAiProvider() }); + }, + onError: (e) => { + setOk(false); + setMsg(e instanceof Error ? e.message : t("settings.ai.errorGeneric")); + }, + }); + + return ( +
+
+

+ {t("settings.ai.responseTimeoutHeading")} +

+

+ {t("settings.ai.responseTimeoutBody")} +

+
+
+ + setValue(e.target.value)} + placeholder={t("settings.ai.responseTimeoutPlaceholder")} + className="mt-1 sm:max-w-xs" + /> +
+
+ +
+ {msg && ( +

+ {msg} +

+ )} +
+ ); +} diff --git a/src/components/settings/ai/shared.ts b/src/components/settings/ai/shared.ts index 4e1e796ce..5d723fbbe 100644 --- a/src/components/settings/ai/shared.ts +++ b/src/components/settings/ai/shared.ts @@ -27,8 +27,21 @@ export interface UserAIProvider { hasLocalKey: boolean; hasOpenaiKey: boolean; openaiKeyPreview: string | null; + // v1.22 (#89) — per-user response timeout, in seconds (null = default). + responseTimeoutSeconds: number | null; + // v1.22 (#90) — dedicated document-scan (Lab-OCR) provider config. + ocrEnabled: boolean; + ocrProvider: string | null; + ocrModel: string | null; + ocrBaseUrl: string | null; + hasOcrKey: boolean; + ocrKeyPreview: string | null; } +/** v1.22 (#90) — providers the dedicated document-scan path accepts. */ +export const OCR_PROVIDER_TYPES = ["OPENAI", "ANTHROPIC", "LOCAL"] as const; +export type OcrProviderType = (typeof OCR_PROVIDER_TYPES)[number]; + /** * v1.4.16 phase B2 — provider tags exposed to the UI. Mirrors * `PROVIDER_CHAIN_TYPES` server-side; kept as a const-array here to diff --git a/src/lib/ai/__tests__/local-client.test.ts b/src/lib/ai/__tests__/local-client.test.ts index 8e6c7b341..8bc0a1a62 100644 --- a/src/lib/ai/__tests__/local-client.test.ts +++ b/src/lib/ai/__tests__/local-client.test.ts @@ -200,6 +200,135 @@ describe("LocalOpenAICompatibleClient", () => { ).rejects.toThrow("Local AI returned empty content"); }); + // ── v1.22 (#89) — true token streaming ────────────────────────── + function sseStreamResponse(chunks: string[]) { + const encoder = new TextEncoder(); + const body = new ReadableStream({ + start(controller) { + for (const c of chunks) controller.enqueue(encoder.encode(c)); + controller.close(); + }, + }); + return { + ok: true, + status: 200, + headers: { + get: (k: string) => (k === "content-type" ? "text/event-stream" : null), + }, + body, + }; + } + + it("streams token deltas over stream:true and assembles the full reply", async () => { + const mockFetch = vi + .fn() + .mockResolvedValue( + sseStreamResponse([ + 'data: {"choices":[{"delta":{"content":"Hel"}}]}\n\n', + 'data: {"choices":[{"delta":{"content":"lo"}}]}\n\n', + 'data: {"choices":[{"delta":{"content":" world"}}]}\n\n', + 'data: {"choices":[{"delta":{}}],"usage":{"total_tokens":42}}\n\n', + "data: [DONE]\n\n", + ]), + ); + vi.stubGlobal("fetch", mockFetch); + + const client = new LocalOpenAICompatibleClient({ + apiKey: null, + model: "llama3:8b", + baseUrl: "http://localhost:11434/v1", + }); + + const deltas: string[] = []; + const result = await client.generateCompletionStream( + singleUserTurn({ system: "s", user: "u" }), + (d) => deltas.push(d), + ); + + expect(deltas).toEqual(["Hel", "lo", " world"]); + expect(result.content).toBe("Hello world"); + expect(result.tokensUsed).toBe(42); + expect(result.providerType).toBe("local"); + + // The request opted into streaming. + const body = JSON.parse(mockFetch.mock.calls[0][1].body); + expect(body.stream).toBe(true); + expect(body.stream_options).toEqual({ include_usage: true }); + }); + + it("falls back to the buffered body when the server ignores stream:true", async () => { + // 200 OK but a normal JSON completion (not event-stream). + const mockFetch = vi.fn().mockResolvedValue({ + ok: true, + status: 200, + headers: { get: () => "application/json" }, + body: null, + json: () => + Promise.resolve({ + choices: [{ message: { content: "buffered reply" } }], + usage: { total_tokens: 7 }, + }), + }); + vi.stubGlobal("fetch", mockFetch); + + const client = new LocalOpenAICompatibleClient({ + apiKey: null, + model: "llama3", + baseUrl: "http://localhost:11434/v1", + }); + + const deltas: string[] = []; + const result = await client.generateCompletionStream( + singleUserTurn({ system: "s", user: "u" }), + (d) => deltas.push(d), + ); + + expect(result.content).toBe("buffered reply"); + expect(result.tokensUsed).toBe(7); + // The whole buffered body is emitted as one delta. + expect(deltas).toEqual(["buffered reply"]); + }); + + it("falls back to non-streaming generateCompletion when streaming is rejected", async () => { + const mockFetch = vi + .fn() + // 1st call: the streaming attempt is rejected (server 400s on stream). + .mockResolvedValueOnce({ + ok: false, + status: 400, + headers: { get: () => "application/json" }, + text: () => Promise.resolve("stream not supported"), + }) + // 2nd call: the buffered retry succeeds. + .mockResolvedValueOnce({ + ok: true, + status: 200, + json: () => + Promise.resolve({ + choices: [{ message: { content: "non-stream reply" } }], + usage: { total_tokens: 5 }, + }), + }); + vi.stubGlobal("fetch", mockFetch); + + const client = new LocalOpenAICompatibleClient({ + apiKey: null, + model: "llama3", + baseUrl: "http://localhost:11434/v1", + }); + + const result = await client.generateCompletionStream( + singleUserTurn({ system: "s", user: "u" }), + () => {}, + ); + + expect(result.content).toBe("non-stream reply"); + expect(mockFetch).toHaveBeenCalledTimes(2); + // The retry was the buffered (non-streaming) request shape. + const retryBody = JSON.parse(mockFetch.mock.calls[1][1].body); + expect(retryBody).not.toHaveProperty("stream"); + }); + it("degrades tools silently — never forwards a `tools` field", async () => { const mockFetch = vi.fn().mockResolvedValue({ ok: true, diff --git a/src/lib/ai/__tests__/ocr-provider-resolution.test.ts b/src/lib/ai/__tests__/ocr-provider-resolution.test.ts new file mode 100644 index 000000000..64ffcd95d --- /dev/null +++ b/src/lib/ai/__tests__/ocr-provider-resolution.test.ts @@ -0,0 +1,120 @@ +import { describe, it, expect, vi, beforeEach } from "vitest"; + +// v1.22 (#90) — dedicated document-scan provider resolution. + +vi.mock("@/lib/db", () => ({ + prisma: { + user: { findUnique: vi.fn() }, + appSettings: { findUnique: vi.fn() }, + }, +})); +vi.mock("@/lib/crypto", () => ({ + decrypt: vi.fn((v: string) => `decrypted:${v}`), + encrypt: vi.fn((v: string) => `encrypted:${v}`), +})); +vi.mock("@/lib/ai/codex-oauth", () => ({ + refreshDeviceTokens: vi.fn(), + encryptCodexCreds: vi.fn(), + decryptCodexCreds: vi.fn(), +})); + +import { resolveOcrProviderChain } from "../provider"; +import { prisma } from "@/lib/db"; + +describe("resolveOcrProviderChain", () => { + beforeEach(() => { + vi.resetAllMocks(); + }); + + it("uses the dedicated OCR provider when enabled with a key", async () => { + // Single read: the OCR columns. The dedicated provider builds, so the main + // chain is never resolved. + vi.mocked(prisma.user.findUnique).mockResolvedValueOnce({ + aiOcrEnabled: true, + aiOcrProvider: "OPENAI", + aiOcrModel: "gpt-4o", + aiOcrBaseUrl: null, + aiOcrKeyEncrypted: "enc-key", + } as never); + + const res = await resolveOcrProviderChain("u1"); + + expect(res.dedicated).toBe(true); + expect(res.ocrModelOverride).toBe("gpt-4o"); + expect(res.chain).toHaveLength(1); + expect(res.chain[0].providerType).toBe("openai"); + // Only the OCR-column read happened — no main-chain resolution. + expect(prisma.user.findUnique).toHaveBeenCalledTimes(1); + }); + + it("maps LOCAL OCR provider to the local chain tag via base URL", async () => { + vi.mocked(prisma.user.findUnique).mockResolvedValueOnce({ + aiOcrEnabled: true, + aiOcrProvider: "LOCAL", + aiOcrModel: "llava", + aiOcrBaseUrl: "http://localhost:11434/v1", + aiOcrKeyEncrypted: null, + } as never); + + const res = await resolveOcrProviderChain("u1"); + + expect(res.dedicated).toBe(true); + expect(res.chain[0].providerType).toBe("local"); + }); + + it("falls back to the main chain when OCR is disabled", async () => { + // First read: OCR columns (disabled). Second read: the main provider chain + // resolution (no enabled chain / creds) → empty main chain. + vi.mocked(prisma.user.findUnique) + .mockResolvedValueOnce({ + aiOcrEnabled: false, + aiOcrProvider: null, + aiOcrModel: null, + aiOcrBaseUrl: null, + aiOcrKeyEncrypted: null, + } as never) + .mockResolvedValueOnce({ + aiProvider: null, + aiModel: null, + aiBaseUrl: null, + aiAnthropicKeyEncrypted: null, + aiLocalKeyEncrypted: null, + aiOpenaiKeyEncrypted: null, + aiProviderChain: null, + } as never); + + const res = await resolveOcrProviderChain("u1"); + + expect(res.dedicated).toBe(false); + expect(res.ocrModelOverride).toBeNull(); + // The OCR-column read plus the main-chain resolution both ran (the + // fallback path), so more than the single dedicated-path read happened. + expect( + vi.mocked(prisma.user.findUnique).mock.calls.length, + ).toBeGreaterThanOrEqual(2); + }); + + it("falls back to the main chain when enabled but the key is missing", async () => { + vi.mocked(prisma.user.findUnique) + .mockResolvedValueOnce({ + aiOcrEnabled: true, + aiOcrProvider: "OPENAI", + aiOcrModel: "gpt-4o", + aiOcrBaseUrl: null, + aiOcrKeyEncrypted: null, // missing credential → cannot build + } as never) + .mockResolvedValueOnce({ + aiProvider: null, + aiModel: null, + aiBaseUrl: null, + aiAnthropicKeyEncrypted: null, + aiLocalKeyEncrypted: null, + aiOpenaiKeyEncrypted: null, + aiProviderChain: null, + } as never); + + const res = await resolveOcrProviderChain("u1"); + + expect(res.dedicated).toBe(false); + }); +}); diff --git a/src/lib/ai/__tests__/provider-runner.test.ts b/src/lib/ai/__tests__/provider-runner.test.ts index b58243dff..46a13fc93 100644 --- a/src/lib/ai/__tests__/provider-runner.test.ts +++ b/src/lib/ai/__tests__/provider-runner.test.ts @@ -32,6 +32,7 @@ import { getLastWorkingProvider, isHardProviderFailure, runRawCompletionWithFallback, + runStreamingRawCompletionWithFallback, runWithFallback, } from "../provider-runner"; import { @@ -720,3 +721,96 @@ describe("AllProvidersFailedError — primaryCredentialExpired", () => { expect(e.primaryCredentialExpired).toBe(false); }); }); + +// ── v1.22 (#89) — streaming runner ─────────────────────────────────── +/** A provider that streams: emits each char of `text` through onDelta. */ +class StreamingProvider implements AIProvider { + readonly type: AIProvider["type"] = "local"; + streamCalls = 0; + bufferedCalls = 0; + lastStreamParams: CompletionParams | null = null; + constructor(private readonly text: string) {} + async generateCompletion(): Promise { + this.bufferedCalls += 1; + return { + content: this.text, + tokensUsed: 1, + model: "stream-model", + providerType: this.type, + }; + } + async generateCompletionStream( + params: CompletionParams, + onDelta: (d: string) => void, + ): Promise { + this.streamCalls += 1; + this.lastStreamParams = params; + for (const ch of this.text) onDelta(ch); + return { + content: this.text, + tokensUsed: 3, + model: "stream-model", + providerType: this.type, + }; + } +} + +describe("runStreamingRawCompletionWithFallback", () => { + it("drives a streaming provider via generateCompletionStream + onDelta and forwards timeoutMs", async () => { + const ledger = createInMemoryProviderHealthLedger(); + const provider = new StreamingProvider("Hi"); + const deltas: string[] = []; + const result = await runStreamingRawCompletionWithFallback({ + userId: "u-stream", + providers: [{ providerType: "local", instance: provider }], + params: singleUserTurn({ system: "s", user: "u", timeoutMs: 12345 }), + onDelta: (d) => deltas.push(d), + ledger, + }); + expect(deltas).toEqual(["H", "i"]); + expect(result.result.content).toBe("Hi"); + expect(provider.streamCalls).toBe(1); + expect(provider.bufferedCalls).toBe(0); + // Per-request timeout is threaded through to the provider call. + expect(provider.lastStreamParams?.timeoutMs).toBe(12345); + }); + + it("falls back to generateCompletion for a provider that does not stream", async () => { + const ledger = createInMemoryProviderHealthLedger(); + // ScriptedProvider has no generateCompletionStream → buffered path. + const provider = new ScriptedProvider({ + script: [{ ok: true, content: "x" }], + }); + const deltas: string[] = []; + const result = await runStreamingRawCompletionWithFallback({ + userId: "u-nostream", + providers: [{ providerType: "local", instance: provider }], + params: singleUserTurn({ system: "s", user: "u" }), + onDelta: (d) => deltas.push(d), + ledger, + }); + expect(result.result.content).toBe("x"); + expect(provider.callCount).toBe(1); + expect(deltas).toEqual([]); // no streaming → no deltas + }); + + it("cascades past a hard failure to the next (streaming) provider", async () => { + const ledger = createInMemoryProviderHealthLedger(); + const broken = new ScriptedProvider({ + script: [{ ok: false, error: err(503) }], + }); + const good = new StreamingProvider("ok"); + const result = await runStreamingRawCompletionWithFallback({ + userId: "u-cascade", + providers: [ + { providerType: "openai", instance: broken }, + { providerType: "local", instance: good }, + ], + params: singleUserTurn({ system: "s", user: "u" }), + onDelta: () => {}, + ledger, + }); + expect(result.workingProvider.providerType).toBe("local"); + expect(result.fallbackHops.map((h) => h.providerType)).toEqual(["openai"]); + }); +}); diff --git a/src/lib/ai/coach/tools/loop.ts b/src/lib/ai/coach/tools/loop.ts index ba7b52be3..a961086d9 100644 --- a/src/lib/ai/coach/tools/loop.ts +++ b/src/lib/ai/coach/tools/loop.ts @@ -96,6 +96,12 @@ export async function runCoachToolLoop(args: { ledger?: ProviderHealthLedger; /** Aborts the per-round provider calls on client disconnect. */ signal?: AbortSignal; + /** + * v1.22 (#89) — per-user upstream timeout (ms) for each tool-round provider + * call. Forwarded to `CompletionParams.timeoutMs`; unset → the client's + * shared default holds. + */ + timeoutMs?: number; }): Promise { const { userId, @@ -108,6 +114,7 @@ export async function runCoachToolLoop(args: { sharedScope, ledger, signal, + timeoutMs, } = args; const messages: AiMessage[] = [...args.messages]; @@ -135,6 +142,7 @@ export async function runCoachToolLoop(args: { temperature, maxTokens, signal, + timeoutMs, ...(offerTools ? { tools, toolChoice: "auto" as const } : { toolChoice: "none" as const }), diff --git a/src/lib/ai/local-client.ts b/src/lib/ai/local-client.ts index 784e53337..924793c3a 100644 --- a/src/lib/ai/local-client.ts +++ b/src/lib/ai/local-client.ts @@ -17,6 +17,14 @@ interface LocalClientConfig { const STRICT_JSON_PREFIX = "Return strict JSON only, no markdown, no commentary.\n\n"; +/** + * v1.22 (#89) — absolute backstop for the streaming path. The real ceiling is + * the per-idle-gap timer (`CompletionParams.timeoutMs`); this only stops a + * server that streams forever from pinning a worker. Ten minutes is far beyond + * any legitimate single Coach turn. + */ +const STREAM_ABSOLUTE_CEILING_MS = 10 * 60_000; + /** * Prepend the strict-JSON instruction to the FIRST user turn's text. Local * model templates respect an in-message instruction better than the @@ -64,9 +72,16 @@ export class LocalOpenAICompatibleClient implements AIProvider { this.config = config; } - async generateCompletion( + /** + * Build the shared request shape (url + headers + body) for both the + * buffered and the streaming paths. `stream` adds the OpenAI-compatible + * `stream: true` + `stream_options.include_usage` so the final SSE chunk + * carries the token total. + */ + private buildRequest( params: CompletionParams, - ): Promise { + stream: boolean, + ): { url: string; headers: Record; body: string } { const url = `${this.config.baseUrl.replace(/\/$/, "")}/chat/completions`; const headers: Record = { @@ -89,6 +104,36 @@ export class LocalOpenAICompatibleClient implements AIProvider { withStrictJsonPrefix(params.messages), ); + const body = JSON.stringify({ + model: this.config.model, + messages, + temperature: params.temperature ?? 0.3, + max_tokens: params.maxTokens ?? 1000, + // Deterministic seed for reproducible reference output. Ollama + + // most OpenAI-compatible local servers honour it; servers that + // ignore it simply disregard the field. Omitted when unset. + ...(params.seed !== undefined ? { seed: params.seed } : {}), + // v1.18.7 — Ollama's native structured-output switch. The JSON + // surfaces opt in via `responseFormat: "json"`; it cuts the + // first-pass JSON-failure rate that the fence-stripping safety net + // otherwise has to catch. The Coach (prose) leaves it unset. + ...(params.responseFormat === "json" ? { format: "json" } : {}), + // v1.22 (#89) — streaming opt-in. `stream_options.include_usage` asks + // OpenAI-compatible servers (vLLM / LM Studio / exo) to append a final + // chunk carrying the usage block so the token footer survives streaming. + ...(stream + ? { stream: true, stream_options: { include_usage: true } } + : {}), + }); + + return { url, headers, body }; + } + + async generateCompletion( + params: CompletionParams, + ): Promise { + const { url, headers, body } = this.buildRequest(params, false); + // safeFetch defaults: no redirect-follow (the local endpoint is the // most exploitable on this surface — a user-controlled baseUrl that // 302s to 169.254.169.254 would otherwise leak the bearer on the @@ -109,21 +154,7 @@ export class LocalOpenAICompatibleClient implements AIProvider { { method: "POST", headers, - body: JSON.stringify({ - model: this.config.model, - messages, - temperature: params.temperature ?? 0.3, - max_tokens: params.maxTokens ?? 1000, - // Deterministic seed for reproducible reference output. Ollama + - // most OpenAI-compatible local servers honour it; servers that - // ignore it simply disregard the field. Omitted when unset. - ...(params.seed !== undefined ? { seed: params.seed } : {}), - // v1.18.7 — Ollama's native structured-output switch. The JSON - // surfaces opt in via `responseFormat: "json"`; it cuts the - // first-pass JSON-failure rate that the fence-stripping safety net - // otherwise has to catch. The Coach (prose) leaves it unset. - ...(params.responseFormat === "json" ? { format: "json" } : {}), - }), + body, }, // v1.20.1 — compose the caller's cancel signal (Coach SSE disconnect) so // a mid-generation abort tears the upstream call down early. @@ -181,4 +212,182 @@ export class LocalOpenAICompatibleClient implements AIProvider { providerType: "local", }; } + + /** + * v1.22 (#89) — true token streaming over the OpenAI-compatible + * `stream: true` wire. POSTs with `stream: true`, reads the SSE response body + * incrementally, and emits each `choices[0].delta.content` chunk through + * `onDelta` as it arrives. The returned `CompletionResult` carries the FULL + * assembled reply so every downstream guard still runs on the complete text. + * + * Timeout model: per-IDLE-gap, not whole-call. An idle timer (`timeoutMs`, + * default 60 s) is armed before the request and reset on every chunk, so a + * slow first token (an MLX/exo server loading the model) and a long total + * generation are both fine as long as bytes keep flowing. A genuine stall + * (no byte for the idle window) aborts the upstream. A generous absolute + * ceiling backstops a server that streams forever. + * + * Graceful fallback: a server that rejects `stream: true` (non-2xx), or + * ignores it and returns a buffered JSON body, or has no readable stream + * body, degrades to the non-streaming {@link generateCompletion} path. The + * Coach gets real tokens where the backend supports streaming and identical + * behaviour where it does not. + */ + async generateCompletionStream( + params: CompletionParams, + onDelta: (delta: string) => void, + ): Promise { + const { url, headers, body } = this.buildRequest(params, true); + const allowPrivate = isLocalAiHostAllowed(url); + + // Idle controller: aborts the upstream when no chunk arrives within the + // per-idle window. Composed with the caller's cancel signal so a client + // disconnect still tears the call down immediately. + const idle = new AbortController(); + const idleMs = params.timeoutMs ?? 60_000; + const signal = params.signal + ? AbortSignal.any([params.signal, idle.signal]) + : idle.signal; + + let res: Response; + try { + res = await safeFetch( + url, + { method: "POST", headers, body }, + { + // Absolute backstop only — the per-idle timer below is the real + // ceiling. A server that streams forever still cannot pin a worker. + timeoutMs: STREAM_ABSOLUTE_CEILING_MS, + requirePublicHost: !allowPrivate, + signal, + }, + ); + } catch { + // Network / private-host / abort before the stream opened → fall back to + // the buffered path, which surfaces the structured error the chain runner + // understands (or succeeds on a server that simply doesn't stream). + return this.generateCompletion(params); + } + + if (!res.ok) { + // The server rejected the streaming request (some servers 400 on + // `stream`/`stream_options`). Retry once on the buffered path so a + // non-streaming server still works and surfaces its real status. + return this.generateCompletion(params); + } + + const contentType = res.headers.get("content-type") ?? ""; + if (!contentType.includes("text/event-stream") || !res.body) { + // The server ignored the stream flag and returned a buffered completion + // (or there is no readable body). Parse it like the non-streaming path + // rather than paying a second round-trip. + const json = (await res.json().catch(() => null)) as { + choices?: Array<{ message?: { content?: string } }>; + usage?: { total_tokens?: number }; + } | null; + const buffered = json?.choices?.[0]?.message?.content; + if (!buffered) { + const err = new Error("Local AI returned empty content"); + Object.assign(err, { + httpStatus: 0, + kind: "empty_response", + upstream: "local", + }); + throw err; + } + onDelta(buffered); + return { + content: buffered, + tokensUsed: json?.usage?.total_tokens ?? null, + model: this.config.model, + providerType: "local", + }; + } + + const reader = res.body.getReader(); + const decoder = new TextDecoder(); + let sseBuffer = ""; + let content = ""; + let tokensUsed: number | null = null; + + let idleTimer: ReturnType | null = null; + const armIdle = () => { + if (idleTimer) clearTimeout(idleTimer); + idleTimer = setTimeout(() => idle.abort(), idleMs); + }; + armIdle(); + + try { + for (;;) { + const { done, value } = await reader.read(); + if (done) break; + armIdle(); + sseBuffer += decoder.decode(value, { stream: true }); + // SSE frames are separated by a blank line; each frame is one or more + // `data:` lines. Split on newlines and process complete lines, keeping + // any partial trailing line in the buffer. + let nl: number; + while ((nl = sseBuffer.indexOf("\n")) !== -1) { + const line = sseBuffer.slice(0, nl).trim(); + sseBuffer = sseBuffer.slice(nl + 1); + if (!line.startsWith("data:")) continue; + const payload = line.slice("data:".length).trim(); + if (!payload || payload === "[DONE]") continue; + let chunk: { + choices?: Array<{ delta?: { content?: string } }>; + usage?: { total_tokens?: number }; + }; + try { + chunk = JSON.parse(payload); + } catch { + // A malformed chunk line is skipped, not fatal — keep reading. + continue; + } + const delta = chunk.choices?.[0]?.delta?.content; + if (delta) { + content += delta; + onDelta(delta); + } + if (typeof chunk.usage?.total_tokens === "number") { + tokensUsed = chunk.usage.total_tokens; + } + } + } + } catch (err) { + // An abort (idle stall or client disconnect) mid-stream: if we already + // assembled usable prose, return it — the user got a real (if clipped) + // answer and every guard still runs on it. With nothing assembled, this + // is a hard transport failure for the chain runner to cascade on. + if (!content) { + const e = new Error("Local AI stream failed before any content"); + Object.assign(e, { + httpStatus: 0, + kind: "stream_aborted", + upstream: "local", + cause: err, + }); + throw e; + } + } finally { + if (idleTimer) clearTimeout(idleTimer); + reader.cancel().catch(() => {}); + } + + if (!content) { + const err = new Error("Local AI returned empty content"); + Object.assign(err, { + httpStatus: 0, + kind: "empty_response", + upstream: "local", + }); + throw err; + } + + return { + content, + tokensUsed, + model: this.config.model, + providerType: "local", + }; + } } diff --git a/src/lib/ai/provider-runner.ts b/src/lib/ai/provider-runner.ts index 5344190cd..b3dfe6437 100644 --- a/src/lib/ai/provider-runner.ts +++ b/src/lib/ai/provider-runner.ts @@ -424,6 +424,92 @@ export async function runWithFallback( throw new AllProvidersFailedError(hops); } +/** + * v1.22 (#89) — streaming sibling of {@link runRawCompletionWithFallback}. + * Identical chain / fallback / health-ledger semantics, but each candidate is + * driven through its `generateCompletionStream` (emitting `onDelta` per token) + * when the provider implements one, and through the buffered + * `generateCompletion` otherwise. The caller (the Coach SSE route) uses the + * delta callback to keep the connection warm and — once safety guards have run + * on the full reply — to confirm real tokens flowed. The returned shape matches + * the buffered runner so the route is path-agnostic. + */ +export async function runStreamingRawCompletionWithFallback(args: { + userId: string; + providers: ProviderChainResolved[]; + params: CompletionParams; + /** Called once per streamed token/chunk for providers that stream. */ + onDelta: (delta: string) => void; + /** See `RunWithFallbackParams.ledger`. */ + ledger?: ProviderHealthLedger; +}): Promise { + const { userId, providers, params, onDelta } = args; + const ledger = args.ledger ?? postgresProviderHealthLedger; + + if (providers.length === 0) { + throw new AllProvidersFailedError([]); + } + + const ordered = await resolveChainOrder(userId, providers, ledger); + const hops: FallbackHop[] = []; + + for (let i = 0; i < ordered.length; i += 1) { + const candidate = ordered[i]; + try { + // Prefer the provider's streaming wire; fall back to the buffered call + // for providers (cloud clients today) that do not implement streaming. + const result = candidate.instance.generateCompletionStream + ? await candidate.instance.generateCompletionStream(params, onDelta) + : await candidate.instance.generateCompletion(params); + rememberWorkingProvider(userId, candidate.providerType); + void ledger.recordSuccess(userId, candidate.providerType); + annotate({ + meta: { + ai_chain_working_provider: candidate.providerType, + ai_chain_fallback_count: hops.length, + ai_chain_streamed: Boolean( + candidate.instance.generateCompletionStream, + ), + }, + }); + return { + result, + workingProvider: candidate, + fallbackHops: hops, + }; + } catch (error) { + if (!isHardProviderFailure(error)) { + throw error; + } + const summary = summariseError(error); + void ledger.recordFailure(userId, candidate.providerType, summary.status); + const hop: FallbackHop = { + providerType: candidate.providerType, + attempt: i + 1, + failureReason: summary.reason, + httpStatus: summary.status, + }; + hops.push(hop); + annotate({ + meta: { + [`ai_chain_hop_${i + 1}_provider`]: candidate.providerType, + [`ai_chain_hop_${i + 1}_status`]: summary.status, + [`ai_chain_hop_${i + 1}_reason`]: summary.reason.slice(0, 240), + [`ai_chain_hop_${i + 1}_body`]: summary.bodyExcerpt, + }, + }); + } + } + + annotate({ + meta: { + ai_chain_outcome: "all-failed", + ai_chain_fallback_count: hops.length, + }, + }); + throw new AllProvidersFailedError(hops); +} + export interface RunRawWithFallbackResult { /** Raw provider response — the legacy `/api/insights/generate` route * consumes the rich shape from `result.content`, so the runner does diff --git a/src/lib/ai/provider.ts b/src/lib/ai/provider.ts index 5ddd2a93e..1f9415b6b 100644 --- a/src/lib/ai/provider.ts +++ b/src/lib/ai/provider.ts @@ -314,6 +314,114 @@ export async function resolveProviderChain( return resolved; } +/** + * v1.22 (#90) — dedicated document-scan (Lab-OCR) provider columns. + */ +type UserOcrRow = { + aiOcrEnabled: boolean | null; + aiOcrProvider: string | null; + aiOcrModel: string | null; + aiOcrBaseUrl: string | null; + aiOcrKeyEncrypted: string | null; +}; + +/** + * Build the dedicated OCR provider instance from the per-user `aiOcr*` columns. + * Returns the constructed instance paired with the chain tag the consent gate + + * vision-capability resolver understand, or null when the OCR provider is + * unconfigured / missing its credential (the caller then falls back to the main + * provider chain so document scanning is never silently lost). + */ +function buildOcrProvider( + row: UserOcrRow, +): { providerType: ProviderChainType; instance: AIProvider } | null { + if (!row.aiOcrEnabled) return null; + const choice = row.aiOcrProvider?.toUpperCase(); + switch (choice) { + case "ANTHROPIC": { + if (!row.aiOcrKeyEncrypted) return null; + return { + providerType: "anthropic", + instance: new AnthropicClient({ + apiKey: decrypt(row.aiOcrKeyEncrypted), + model: row.aiOcrModel ?? "claude-sonnet-4-6", + }), + }; + } + case "OPENAI": { + if (!row.aiOcrKeyEncrypted) return null; + return { + providerType: "openai", + instance: new OpenAIClient({ + apiKey: decrypt(row.aiOcrKeyEncrypted), + model: row.aiOcrModel ?? "gpt-4o", + baseUrl: "https://api.openai.com/v1", + }), + }; + } + case "LOCAL": { + if (!row.aiOcrBaseUrl) return null; + return { + providerType: "local", + instance: new LocalOpenAICompatibleClient({ + apiKey: row.aiOcrKeyEncrypted ? decrypt(row.aiOcrKeyEncrypted) : null, + model: row.aiOcrModel ?? "local-model", + baseUrl: row.aiOcrBaseUrl, + }), + }; + } + default: + return null; + } +} + +/** + * v1.22 (#90) — resolve the provider chain for the Lab-OCR ingestion path. + * + * When the user enabled a dedicated document-scan provider (Settings → AI → + * Document scanning), this returns a single-entry chain built from the `aiOcr*` + * columns plus the OCR model that drives the vision-capability allowlist. + * Otherwise (the default) it returns the MAIN provider chain unchanged, so the + * OCR path behaves exactly as before the feature existed. A dedicated provider + * that fails to build (missing key / base URL) also falls through to the main + * chain so scanning is never silently dropped. + */ +export async function resolveOcrProviderChain(userId: string): Promise<{ + chain: ProviderChainResolved[]; + /** The model driving the dedicated provider's vision allowlist, else null. */ + ocrModelOverride: string | null; + /** True when the dedicated document-scan provider served this chain. */ + dedicated: boolean; +}> { + const row = await prisma.user.findUnique({ + where: { id: userId }, + select: { + aiOcrEnabled: true, + aiOcrProvider: true, + aiOcrModel: true, + aiOcrBaseUrl: true, + aiOcrKeyEncrypted: true, + }, + }); + + if (row?.aiOcrEnabled) { + const built = buildOcrProvider(row); + if (built) { + return { + chain: [built], + ocrModelOverride: row.aiOcrModel ?? null, + dedicated: true, + }; + } + } + + return { + chain: await resolveProviderChain(userId), + ocrModelOverride: null, + dedicated: false, + }; +} + /** * Credential-presence subset of the `User` row that * `userRowHasProviderCredential` evaluates. Mirrors the columns diff --git a/src/lib/ai/types.ts b/src/lib/ai/types.ts index 0837fbd01..06d5f98e5 100644 --- a/src/lib/ai/types.ts +++ b/src/lib/ai/types.ts @@ -121,6 +121,24 @@ export interface AIProvider { */ supportsTools?: boolean; generateCompletion(params: CompletionParams): Promise; + /** + * v1.22 (#89) — optional true-streaming variant. A provider that can emit + * tokens incrementally (the local OpenAI-compatible client over `stream:true`) + * implements this so a slow self-hosted backend (e.g. an MLX/exo server that + * takes >60 s to load the model before the first token) produces output as it + * arrives rather than buffering the whole reply behind a single fetch — which + * is what tripped the legacy total-request timeout and the reverse-proxy idle + * cut. `onDelta` is called once per token/chunk as it arrives; the returned + * `CompletionResult` carries the FULL assembled reply so every downstream + * guard (grounding, refusal, sentinel-stripping) still runs on the complete + * text. The timeout becomes a per-idle-gap ceiling (time-to-first-token + + * inter-token gap), not a whole-call ceiling. Providers without a streaming + * wire omit this; the runner falls back to {@link generateCompletion}. + */ + generateCompletionStream?( + params: CompletionParams, + onDelta: (delta: string) => void, + ): Promise; } /** diff --git a/src/lib/crypto/encrypted-columns.ts b/src/lib/crypto/encrypted-columns.ts index 0d9d97a0b..d408bdd7d 100644 --- a/src/lib/crypto/encrypted-columns.ts +++ b/src/lib/crypto/encrypted-columns.ts @@ -68,6 +68,8 @@ export const ENCRYPTED_COLUMNS: readonly EncryptedColumn[] = [ { model: "User", field: "aiAnthropicKeyEncrypted", kind: "string" }, { model: "User", field: "aiLocalKeyEncrypted", kind: "string" }, { model: "User", field: "aiOpenaiKeyEncrypted", kind: "string" }, + // v1.22 (#90) — dedicated document-scan (Lab-OCR) provider key. + { model: "User", field: "aiOcrKeyEncrypted", kind: "string" }, { model: "User", field: "insuranceNumberEncrypted", kind: "string" }, // ───── OAuth token tables (`*_token`-convention columns) ───── diff --git a/src/lib/labs/ocr-capability.ts b/src/lib/labs/ocr-capability.ts index 400fc0187..5991932c2 100644 --- a/src/lib/labs/ocr-capability.ts +++ b/src/lib/labs/ocr-capability.ts @@ -10,7 +10,7 @@ * vision allowlist. */ import type { ProviderChainResolved } from "@/lib/ai/provider-runner"; -import { resolveProvider, resolveProviderChain } from "@/lib/ai/provider"; +import { resolveProvider, resolveOcrProviderChain } from "@/lib/ai/provider"; import { resolveCodexVisionSlug } from "@/lib/ai/codex-client"; import { prisma } from "@/lib/db"; import { @@ -70,14 +70,21 @@ export async function resolveVisionProvider( where: { id: "singleton" }, select: { adminAiModel: true }, }); - const ctx: ModelContext = { - userModel: userRow?.aiModel ?? null, - adminModel: settings?.adminAiModel ?? null, - }; const localOcrEnabled = userRow?.labsLocalOcrEnabled ?? false; - const chain = await resolveProviderChain(userId); + // v1.22 (#90) — resolve the OCR chain: the dedicated document-scan provider + // when the user enabled one, else the main provider chain unchanged. When a + // dedicated provider is active, ITS model drives the vision allowlist. + const ocr = await resolveOcrProviderChain(userId); + const ctx: ModelContext = { + userModel: ocr.dedicated + ? (ocr.ocrModelOverride ?? userRow?.aiModel ?? null) + : (userRow?.aiModel ?? null), + adminModel: settings?.adminAiModel ?? null, + }; + + const chain = ocr.chain; if (chain.length === 0) { // Mirror the Coach: fall back to the legacy single-provider resolution and // tag it as the admin-managed entry the consent gate recognises. @@ -116,7 +123,9 @@ export async function resolveTextProvider(userId: string): Promise<{ chain: ProviderChainResolved[]; pick: { entry: ProviderChainResolved; providerType: string } | null; }> { - const chain = await resolveProviderChain(userId); + // v1.22 (#90) — use the dedicated document-scan provider when enabled, else + // the main chain (the text-mode structuring pass needs no vision). + const chain = (await resolveOcrProviderChain(userId)).chain; if (chain.length === 0) { const legacy = await resolveProvider(userId); if (legacy.type !== "none") { diff --git a/src/lib/openapi/routes/profile.ts b/src/lib/openapi/routes/profile.ts index 941e53245..d987f1fc7 100644 --- a/src/lib/openapi/routes/profile.ts +++ b/src/lib/openapi/routes/profile.ts @@ -334,6 +334,34 @@ const aiProviderResponse = z .string() .nullable() .describe("`...` + last 4 chars of the stored OpenAI key, or null."), + responseTimeoutSeconds: z + .number() + .int() + .nullable() + .describe( + "v1.22 (#89) — per-user response timeout for AI generation, in seconds (10–600). Null = the generous built-in default (~180 s). Mainly for slow local/self-hosted backends.", + ), + ocrEnabled: z + .boolean() + .describe( + "v1.22 (#90) — whether a dedicated document-scan (Lab-OCR) provider is active. When false the OCR path uses the main provider chain.", + ), + ocrProvider: z + .enum(["OPENAI", "ANTHROPIC", "LOCAL"]) + .nullable() + .describe("The dedicated document-scan provider, or null when unset."), + ocrModel: z.string().nullable(), + ocrBaseUrl: z + .string() + .nullable() + .describe("Custom base URL (LOCAL OCR provider only); null otherwise."), + hasOcrKey: z.boolean(), + ocrKeyPreview: z + .string() + .nullable() + .describe( + "`...` + last 4 chars of the stored document-scan provider key, or null.", + ), }) .meta({ id: "AiProviderResponse",