merge: AI settings (coach timeout, local streaming, OCR provider) into v1.22

This commit is contained in:
Marc-André Bombeck committed 2026-06-27 15:22:55 +02:00
commit 7ff14864b9
30 files changed
+2281 -569

No files matched your search

@@ -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<unknown> }));
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>,
) => {
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<Response>;
async function postAndDrain(body: Record<string, unknown>): Promise<Response> {
const res = await post(chatReq(body));
await sse.done;
return res;
}
function chatReq(body: Record<string, unknown>): Request {
return new Request("http://localhost/api/insights/chat", {
method: "POST",
@@ -167,7 +174,8 @@ function chatReq(body: Record<string, unknown>): 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);
@@ -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<unknown> }));
vi.mock("@/lib/sse/create-stream", () => ({
createSseStream: (
producer: (c: {
@@ -201,8 +205,9 @@ vi.mock("@/lib/sse/create-stream", () => ({
enqueue: () => void;
}) => void | Promise<void>,
) => {
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<Response>;
/** Run a POST and wait for the in-stream producer to finish. */
async function postAndDrain(body: Record<string, unknown>): Promise<Response> {
const res = await post(chatReq(body));
await sse.done;
return res;
}
function chatReq(body: Record<string, unknown>): 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<typeof vi.fn>).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<typeof vi.fn>).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;
File diff suppressed because it is too large. Load diff
@@ -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<Response>)();
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<Response>;
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/);
});
});
+103
View File
@@ -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");
}
@@ -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). */}
<ResponseTimeoutCard userProvider={userProvider} />
{/* v1.22 (#90) — dedicated document-scan (Lab-OCR) provider. */}
<DocumentScanCard userProvider={userProvider} />
<RuntimeActionsRow
provider={selectedProvider}
userProvider={userProvider}
@@ -0,0 +1,212 @@
"use client";
/* ────────────────────────────────────────────────────────────────
* v1.22 (#90) — Document scanning (Lab reports).
*
* Opt-in dedicated provider for the Lab-OCR ingestion path. OFF by default:
* scanning uses the main provider chain. When enabled, document/lab-report
* processing routes to THIS provider/model/key instead — so a user can keep a
* cheap text Coach on one provider and send expensive vision OCR to a
* vision-capable one. Surfaces a hint when the chosen model can't read images.
* ──────────────────────────────────────────────────────────────── */
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 { NativeSelect } from "@/components/ui/native-select";
import { PasswordInput } from "@/components/ui/password-input";
import { Switch } from "@/components/ui/switch";
import { apiPatch } from "@/lib/api/api-fetch";
import {
supportsVisionForConfig,
type VisionProviderType,
} from "@/lib/ai/vision-capability";
import { useTranslations } from "@/lib/i18n/context";
import { queryKeys } from "@/lib/query-keys";
import { OCR_PROVIDER_TYPES, type UserAIProvider } from "./shared";
/** Map the OCR provider enum to the vision-capability tag. */
function toVisionType(provider: string): VisionProviderType {
switch (provider) {
case "ANTHROPIC":
return "anthropic";
case "OPENAI":
return "openai";
case "LOCAL":
return "local";
default:
return "none";
}
}
export function DocumentScanCard({
userProvider,
}: {
userProvider: UserAIProvider | null | undefined;
}) {
const { t } = useTranslations();
const queryClient = useQueryClient();
const [enabled, setEnabled] = useState(false);
const [provider, setProvider] = useState<string>("OPENAI");
const [model, setModel] = useState("");
const [baseUrl, setBaseUrl] = useState("");
const [apiKey, setApiKey] = useState("");
const [msg, setMsg] = useState<string | null>(null);
const [ok, setOk] = useState(false);
const seededKey =
userProvider != null
? `${userProvider.ocrEnabled}|${userProvider.ocrProvider ?? ""}|${userProvider.ocrModel ?? ""}|${userProvider.ocrBaseUrl ?? ""}`
: null;
const [previousSeed, setPreviousSeed] = useState<string | null>(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<string, unknown> = {
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 (
<div className="bg-muted/50 space-y-4 rounded-lg p-4">
<div className="flex items-start justify-between gap-3">
<div>
<p className="text-sm font-medium">{t("settings.ai.ocrHeading")}</p>
<p className="text-muted-foreground text-xs">
{t("settings.ai.ocrBody")}
</p>
</div>
<Switch
checked={enabled}
onCheckedChange={setEnabled}
aria-label={t("settings.ai.ocrEnableLabel")}
data-testid="ai-ocr-enable"
/>
</div>
{enabled && (
<div className="space-y-4">
<div>
<Label htmlFor="ai-ocr-provider">
{t("settings.ai.ocrProviderLabel")}
</Label>
<NativeSelect
id="ai-ocr-provider"
value={provider}
onChange={(e) => setProvider(e.target.value)}
className="mt-1 sm:max-w-md"
>
{OCR_PROVIDER_TYPES.map((p) => (
<option key={p} value={p}>
{t(`settings.ai.ocrProviderOption.${p}` as const)}
</option>
))}
</NativeSelect>
</div>
<div>
<Label htmlFor="ai-ocr-model">{t("settings.ai.modelLabel")}</Label>
<Input
id="ai-ocr-model"
value={model}
onChange={(e) => setModel(e.target.value)}
placeholder={t("settings.ai.ocrModelPlaceholder")}
className="mt-1"
/>
{!visionOk && (
<p className="text-warning mt-1 text-xs">
{t("settings.ai.ocrVisionHint")}
</p>
)}
</div>
{provider === "LOCAL" && (
<div>
<Label htmlFor="ai-ocr-base-url">
{t("settings.ai.baseUrlLabel")}
</Label>
<Input
id="ai-ocr-base-url"
value={baseUrl}
onChange={(e) => setBaseUrl(e.target.value)}
placeholder="http://localhost:11434/v1"
className="mt-1"
/>
</div>
)}
<div>
<Label htmlFor="ai-ocr-key">{t("settings.ai.ocrKeyLabel")}</Label>
<PasswordInput
id="ai-ocr-key"
value={apiKey}
onChange={(e) => setApiKey(e.target.value)}
placeholder={
userProvider?.hasOcrKey ? t("settings.ai.savedShort") : ""
}
className="mt-1"
/>
</div>
</div>
)}
<div>
<Button
size="sm"
className="min-h-11 sm:min-h-9"
onClick={() => saveMutation.mutate()}
disabled={saveMutation.isPending}
>
{saveMutation.isPending ? (
<Loader2 className="h-4 w-4 animate-spin motion-reduce:animate-none" />
) : (
<Save className="h-4 w-4" />
)}
{t("settings.ai.saveCta")}
</Button>
</div>
{msg && (
<p className={`text-xs ${ok ? "text-success" : "text-destructive"}`}>
{msg}
</p>
)}
</div>
);
}
@@ -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<string | null>(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<string | null>(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 (
<div className="bg-muted/50 space-y-3 rounded-lg p-4">
<div>
<p className="text-sm font-medium">
{t("settings.ai.responseTimeoutHeading")}
</p>
<p className="text-muted-foreground text-xs">
{t("settings.ai.responseTimeoutBody")}
</p>
</div>
<div>
<Label htmlFor="ai-response-timeout">
{t("settings.ai.responseTimeoutLabel")}
</Label>
<Input
id="ai-response-timeout"
type="number"
inputMode="numeric"
min={10}
max={600}
value={value}
onChange={(e) => setValue(e.target.value)}
placeholder={t("settings.ai.responseTimeoutPlaceholder")}
className="mt-1 sm:max-w-xs"
/>
</div>
<div>
<Button
size="sm"
className="min-h-11 sm:min-h-9"
onClick={() => saveMutation.mutate()}
disabled={saveMutation.isPending}
>
{saveMutation.isPending ? (
<Loader2 className="h-4 w-4 animate-spin motion-reduce:animate-none" />
) : (
<Save className="h-4 w-4" />
)}
{t("settings.ai.saveCta")}
</Button>
</div>
{msg && (
<p className={`text-xs ${ok ? "text-success" : "text-destructive"}`}>
{msg}
</p>
)}
</div>
);
}
+13
View File
@@ -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
+129
View File
@@ -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<Uint8Array>({
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,
@@ -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);
});
});
@@ -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<CompletionResult> {
this.bufferedCalls += 1;
return {
content: this.text,
tokensUsed: 1,
model: "stream-model",
providerType: this.type,
};
}
async generateCompletionStream(
params: CompletionParams,
onDelta: (d: string) => void,
): Promise<CompletionResult> {
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"]);
});
});
+8
View File
@@ -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<CoachToolLoopResult> {
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 }),
+226 -17
View File
@@ -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<CompletionResult> {
stream: boolean,
): { url: string; headers: Record<string, string>; body: string } {
const url = `${this.config.baseUrl.replace(/\/$/, "")}/chat/completions`;
const headers: Record<string, string> = {
@@ -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<CompletionResult> {
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<CompletionResult> {
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<typeof setTimeout> | 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",
};
}
}
+86
View File
@@ -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<RunRawWithFallbackResult> {
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
+108
View File
@@ -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
+18
View File
@@ -121,6 +121,24 @@ export interface AIProvider {
*/
supportsTools?: boolean;
generateCompletion(params: CompletionParams): Promise<CompletionResult>;
/**
* 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<CompletionResult>;
}
/**
+2
View File
@@ -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) ─────
+16 -7
View File
@@ -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") {
+28
View File
@@ -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",