From fe53804c4cf6bf1ca0acdbf75d7211ce36c8d86b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sun, 27 Sep 2026 10:14:07 +0200 Subject: [PATCH] refactor(coach): split the chat turn into a pipeline under lib/ai/coach/turn The route keeps the front door: auth, the coach capability, the body read and schema, the rate limit, the locale and the inbound refusal screen. Everything after it moves into one module per step: conversation, context, chain, budget, model, reply-guards, cards, provenance, persist, sse, errors and pipeline. The capability re-check at the egress site stays owned by the route and is handed to the pipeline, so the capability inventory still reads both calls there. The stream producer now writes through a TurnEmitter created inside createSseStream and passed to the model step, which is where live progress frames will come from. Nothing writes to it early yet. No behaviour change: the golden transcripts pass byte-identical, the existing route tests pass unmodified, and every annotate, audit and error-code name appears as often as before. Guards follow the code. The fence call-site check reads turn/context.ts, the German-default locale ban covers every file in turn/, and the fenced module-graph test now asserts that turn/sse.ts and turn/errors.ts reach neither the tool registry nor the snapshot builder. That test named lib/ai/coach/tools.ts, which stopped existing when the registry became a directory, so its negative check could not fail; it points at tools/index.ts now, and a new sanity check fails if the tool route can no longer reach the path it names. --- src/app/api/insights/chat/route.ts | 1557 +---------------- .../coach-locale-no-german-default.test.ts | 19 +- .../__tests__/data-field-injection.test.ts | 6 +- src/lib/ai/coach/turn/budget.ts | 146 ++ src/lib/ai/coach/turn/cards.ts | 89 + src/lib/ai/coach/turn/chain.ts | 54 + src/lib/ai/coach/turn/context.ts | 223 +++ src/lib/ai/coach/turn/conversation.ts | 159 ++ src/lib/ai/coach/turn/errors.ts | 130 ++ src/lib/ai/coach/turn/model.ts | 322 ++++ src/lib/ai/coach/turn/persist.ts | 76 + src/lib/ai/coach/turn/pipeline.ts | 176 ++ src/lib/ai/coach/turn/provenance.ts | 88 + src/lib/ai/coach/turn/reply-guards.ts | 338 ++++ src/lib/ai/coach/turn/sse.ts | 218 +++ src/lib/ai/coach/turn/types.ts | 87 + .../fenced-chat-module-graph.test.ts | 35 +- 17 files changed, 2191 insertions(+), 1532 deletions(-) create mode 100644 src/lib/ai/coach/turn/budget.ts create mode 100644 src/lib/ai/coach/turn/cards.ts create mode 100644 src/lib/ai/coach/turn/chain.ts create mode 100644 src/lib/ai/coach/turn/context.ts create mode 100644 src/lib/ai/coach/turn/conversation.ts create mode 100644 src/lib/ai/coach/turn/errors.ts create mode 100644 src/lib/ai/coach/turn/model.ts create mode 100644 src/lib/ai/coach/turn/persist.ts create mode 100644 src/lib/ai/coach/turn/pipeline.ts create mode 100644 src/lib/ai/coach/turn/provenance.ts create mode 100644 src/lib/ai/coach/turn/reply-guards.ts create mode 100644 src/lib/ai/coach/turn/sse.ts create mode 100644 src/lib/ai/coach/turn/types.ts diff --git a/src/app/api/insights/chat/route.ts b/src/app/api/insights/chat/route.ts index ab0a26722..985da05aa 100644 --- a/src/app/api/insights/chat/route.ts +++ b/src/app/api/insights/chat/route.ts @@ -6,31 +6,19 @@ * single `provenance` frame describing what the assistant could see, * and a closing `done` frame carrying the persisted message ids. * - * Behaviour: - * 1. requireAuth() — cookie session OR bearer token (iOS app), then the - * `coach` AI capability (operator switch, the Coach module and the - * `disableCoach` opt-out, provider presence, consent). Refused with the - * capability envelope, except `no_provider`, which keeps its - * `coach.provider.none` SSE frame (now with `reason`). - * 2. Validate body with `coachChatRequestSchema`. - * 3. reserveBudget(resolveDailyCap(chain)) — 429 with - * `coach.budget.exceeded` when the day's token cap for the - * credential that would pay for the call is already burned. - * 4. detectRefusal() — pattern-based prompt-injection + - * off-topic guard. Refusal emits a single `token` frame with - * the localised refusal copy and a `done` frame; never hits a - * provider. - * 5. Idempotency: only when `conversationId` is absent (= the user - * is creating a new conversation). withIdempotency() caches the - * streamed body so a retry under the same Idempotency-Key - * replays the original assistant message instead of double- - * creating the conversation. - * 6. Provider chain — runRawCompletionWithFallback() walks the - * user's configured providers; on AllProvidersFailedError emit - * an `error` frame and persist nothing. - * 7. Persist user message + assistant message (encrypted) and bump - * the day's CoachUsage token ledger. + * This file owns the front door: auth, the `coach` AI capability + * (refused with the capability envelope, except `no_provider`, which + * keeps its `coach.provider.none` SSE frame with `reason`), the body + * read and `coachChatRequestSchema`, the per-user rate limit, the + * locale, and the inbound refusal screen (`detectRefusal` — a refusal + * streams the localised copy and never reaches a provider). + * + * Everything after that — conversation, context, provider chain and + * consent, the capability re-check at the egress site, the budget + * reservation, the model, the reply guards, persistence and the + * streamed frames — is the turn pipeline in `src/lib/ai/coach/turn/`. */ + import { type NextRequest } from "next/server"; import { apiHandler, requireAuth, HttpError } from "@/lib/api-handler"; @@ -40,158 +28,16 @@ import { apiValidationError, sanitiseZodIssues, } from "@/lib/api-response"; -import { annotate, getEvent } from "@/lib/logging/context"; -import { redactOptional, redactSecrets } from "@/lib/logging/redact"; -import { auditLog } from "@/lib/auth/audit"; -import { prisma } from "@/lib/db"; +import { annotate } from "@/lib/logging/context"; import { checkRateLimit } from "@/lib/rate-limit"; -import { isModuleEnabled } from "@/lib/modules/gate"; import { requireAiCapability } from "@/lib/ai/capabilities/gate"; import { AiUnavailableError } from "@/lib/ai/capabilities/refusal"; -import type { AiUnavailableReason } from "@/lib/ai/capabilities/types"; - import { resolveServerLocale } from "@/lib/i18n/server-locale"; -import { getServerTranslator } from "@/lib/i18n/server-translator"; -import { localeLanguageNames as LANGUAGE_NAMES } from "@/lib/i18n/config"; -import { - AllProvidersFailedError, - runStreamingRawCompletionWithFallback, -} from "@/lib/ai/provider-runner"; -import { resolveProviderChain, resolveProvider } from "@/lib/ai/provider"; -import type { ProviderChainType } from "@/lib/ai/provider-chain"; -import { assertConsentForChain } from "@/lib/ai/consent-guard"; -import { singleUserTurn, type CompletionResult } from "@/lib/ai/types"; -import { PROMPT_VERSION } from "@/lib/ai/prompts/insight-generator"; -import { AI_BUDGETS } from "@/lib/ai/ai-budgets"; - -import { - coachChatRequestSchema, - type CoachStreamEvent, -} from "@/lib/ai/coach/types"; -import { - appendMessage, - createConversation, - fetchConversationWithMessages, - listConversations, -} from "@/lib/ai/coach/persistence"; -import { enqueueCoachMemoryRefresh } from "@/lib/ai/coach/coach-memory-shared"; -import { instructionLocale } from "@/lib/ai/prompts/output-language"; -import { storeDeterministicFacts } from "@/lib/ai/coach/facts"; -import { - buildDateKey, - reserveBudget, - reconcileSpend, - resolveCostOwner, - resolveDailyCap, -} from "@/lib/ai/coach/budget"; +import { coachChatRequestSchema } from "@/lib/ai/coach/types"; +import { listConversations } from "@/lib/ai/coach/persistence"; import { detectRefusal } from "@/lib/ai/coach/refusal"; -import { - screenCoachReply, - coachOutboundFallback, -} from "@/lib/ai/coach/outbound-guard"; -import { getCoachSystemPrompt } from "@/lib/ai/coach/system-prompt"; -import { - openerArchetypeHint, - shouldUseNameForTurn, - firstNameFromDisplayName, -} from "@/lib/ai/prompts/opener-archetype"; -import { getSelfContextTextForUser } from "@/lib/ai/coach/about-me"; -import { buildCoachSnapshot } from "@/lib/ai/coach/snapshot"; -import { - buildCoachProviderPrompts, - buildCoachToolRequest, - buildCoachTurnContext, - type CoachTurn, -} from "@/lib/ai/coach/chat-request-builder"; -import { buildWorkoutEvidenceSection } from "@/lib/ai/coach/workout-evidence-builder"; -import { - COACH_TOOL_DEFS, - buildCoachDataInventory, - renderDataInventory, - renderFocusHint, - buildToolModeAddendum, - runCoachToolLoop, - MAX_ROUNDS, - type CoachToolTrace, -} from "@/lib/ai/coach/tools"; -import { parseKeyValuesSentinel } from "@/lib/ai/coach/keyvalues"; -import { - findUnverifiedCoachNumbersInLedger, - stripUnverifiedNumbers, -} from "@/lib/ai/coach/coach-prose-grounding"; -import { - buildGroundingLedger, - figuresForPersistence, -} from "@/lib/ai/coach/grounding-ledger"; -import { getScheduledDoseValues } from "@/lib/medications/scheduled-doses"; -import { scrubUnknownLearnLinks } from "@/lib/ai/coach/learn-link-guard"; -import { parseSuggestReminder } from "@/lib/ai/coach/suggest-reminder"; -import { gateSuggestion } from "@/lib/ai/coach/suggest-gate"; -import { - parseRememberSentinel, - captureReminderFromSentinel, - buildRememberAddendum, -} from "@/lib/ai/coach/reminders"; -import { - parseSuggestAction, - buildSuggestActionAddendum, - type CoachSuggestedAction, -} from "@/lib/ai/coach/suggest-action"; -import { - parseCoachPrefs, - DEFAULT_REMINDER_SUGGESTION_PREFS, -} from "@/lib/validations/coach-prefs"; -import type { CoachSuggestion } from "@/lib/ai/coach/types"; -import { createSseStream } from "@/lib/sse/create-stream"; - -const SSE_HEADERS: Record = { - "Content-Type": "text/event-stream; charset=utf-8", - "Cache-Control": "no-cache, no-transform", - Connection: "keep-alive", - "X-Accel-Buffering": "no", -}; - -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 - * returned the body in one shot. - */ -function tokeniseForStreaming(content: string): string[] { - if (!content) return []; - // Split on whitespace boundaries while preserving the spaces — keeps - // word boundaries intact and avoids the UI having to glue tokens. - const matches = content.match(/\S+\s*/g); - return matches ?? [content]; -} - -/** - * v1.12.0 — yield control back to the event loop for one tick so the - * stream controller flushes the just-enqueued frame before the next one - * is produced. `setTimeout(0)` (rather than a bare `Promise.resolve()` - * microtask) hands the turn back to the platform's stream pump so each - * SSE frame lands in its own network chunk; a microtask would drain - * before the runtime gets a chance to flush. The delay is intentionally - * zero — we want incremental delivery, not an artificial typewriter - * pause. - */ -function flushTick(): Promise { - return new Promise((resolve) => setTimeout(resolve, 0)); -} +import { runCoachTurn } from "@/lib/ai/coach/turn/pipeline"; +import { streamProviderError, streamRefusal } from "@/lib/ai/coach/turn/sse"; /** * The `coach` capability gate for a turn. Every unavailable reason throws the @@ -282,9 +128,9 @@ async function handleChatRequest(request: NextRequest): Promise { } // v1.18.7 (SENIOR-DEV HIGH) — the daily token cap is enforced atomically - // by reserving budget BEFORE the provider call (see the reservation right - // before `runRawCompletionWithFallback`), not by a read-then-write here. - // The old read-before-call gate let concurrent requests all pass the cap. + // by reserving budget right before the provider call (`reserveTurnBudget` + // in the turn pipeline), not by a read-then-write here. The old + // read-before-call gate let concurrent requests all pass the cap. const locale = await resolveServerLocale({ request, @@ -314,1364 +160,19 @@ async function handleChatRequest(request: NextRequest): Promise { }); } - // ── Conversation resolution ────────────────────────────────── - let workingConversationId: string; - let priorTurns: CoachTurn[] = []; - // v1.32.9 (Coach Guard II / G2) — cross-turn Grounding Ledger sources. Prior - // USER messages (numbers the user themselves stated) and the persisted - // per-turn tool figures of prior turns (`groundedFigures` — server-computed - // magnitudes, NEVER the assistant's prose, per D3). Both feed reconciliation - // so a figure recalled a turn later is not stripped, without a laundering - // path through assistant narration. - let priorUserMessages: string[] = []; - let priorToolFigures: number[] = []; - // v1.11.1 — rolling summary of the elided older turns, read stale-while- - // revalidate; null for a fresh conversation or when none is on file. - let priorSummary: string | null = null; - - if (conversationId) { - // v1.29.x (S7) — the load-bearing prompt-injection fence, now a DUAL - // predicate. Untrusted document text may only enter an LLM prompt on the - // fenced pipeline (no tools, no snapshot). This tool route must never load a - // fenced conversation: - // PRIMARY — `documentScoped: false` in the WHERE. The sticky flag is set - // at fenced-creation / first-attach and NEVER cleared, so a - // conversation that has EVER held a document 404s here forever. - // BACKSTOP — even so, assert zero live attachments. `documentScoped: false` - // with an attachment row present is flag/join drift (an - // invariant broke somewhere): fail closed, loudly. - // A doc turn is routed by the client to `/api/insights/chat/fenced` (or the - // single-doc sheet endpoint). Should a fenced id ever reach THIS route the - // fetch returns nothing and the turn 404s rather than running an injected - // instruction against the coach's write tools. Do not relax. - const existing = await fetchConversationWithMessages( - userId, - conversationId, - { documentScoped: false }, - ); - if (!existing) { - // 404, not 403 — never reveal cross-user / cross-mode existence - throw new HttpError(404, "coach.conversation.notFound"); - } - if (existing.attachmentCount > 0) { - // Drift alarm — telemetry AND an audit row (not routine telemetry). - annotate({ - action: { name: "insights.coach.fence_drift" }, - meta: { conversationId: existing.id }, - }); - await auditLog("insights.coach.fence_drift", { - userId, - details: { conversationId: existing.id }, - }); - throw new HttpError(404, "coach.conversation.notFound"); - } - workingConversationId = existing.id; - priorSummary = existing.summary ?? null; - // #781 — cancelled-turn markers (`providerType: "cancelled"`, empty body) - // are UI rows, not conversation content: keep them out of the provider - // transcript so an interrupted turn adds no empty assistant line to the - // prompt. They stay in the DTO the client renders. - priorTurns = existing.messages - .filter((m) => m.providerType !== "cancelled") - .map((m) => ({ - role: m.role, - content: m.content, - })); - // Ledger cross-turn sources (D3-safe): user-authored numbers + the - // persisted tool figures of prior turns. Assistant PROSE is never read. - priorUserMessages = existing.messages - .filter((m) => m.role === "user") - .map((m) => m.content); - priorToolFigures = existing.messages.flatMap((m) => - m.metricSource?.groundedFigures - ? [...m.metricSource.groundedFigures] - : [], - ); - - // v1.4.43 W13 M-3 — replay-injection guard. `detectRefusal` runs - // only on the inbound `message` per turn, so an injection that - // slipped past the regex bank on a previous turn would re-enter - // the prompt every reply. Re-run the detector against every - // user-turn re-loaded from DB; on a hit, short-circuit the SSE - // with a refusal AND drop an `insights.coach.replay_injection` - // row so the failure case is observable. The audit row carries - // the conversation id (server-owned), the turn index (no PII) - // and the matched reason — never the message content. v1.4.43 - // W10 simplifier-L-2 — action name follows the `.` - // convention (no `audit.` prefix; no `.replay-injection` dash). - for (let i = 0; i < priorTurns.length; i++) { - const turn = priorTurns[i]; - if (turn.role !== "user") continue; - const replayed = detectRefusal({ message: turn.content, locale }); - if (!replayed.refuse) continue; - annotate({ - action: { name: "insights.coach.replay_injection" }, - meta: { reason: replayed.reason, turnIndex: i }, - }); - await auditLog("insights.coach.replay_injection", { - userId, - details: { - conversationId: existing.id, - turnIndex: i, - reason: replayed.reason, - }, - }); - return streamRefusal({ - userId, - conversationId: existing.id, - message, - refusalText: - replayed.message ?? - getServerTranslator(locale).t("coach.refusal.conversationPoisoned"), - }); - } - } else { - const created = await createConversation({ userId, title: message }); - workingConversationId = created.id; - } - - // Persist the user's turn first so it's safely on disk regardless - // of whether the provider call succeeds. - await appendMessage({ - conversationId: workingConversationId, - role: "user", - content: message, - }); - - // Build the prompt: system + (optional) snapshot + recent history + - // the new user message. v1.4.23 H4 — fold per-user prefs into the - // system-prompt prefix; the snapshot builder reads the same prefs - // separately so excluded metrics never even leave the DB. - // - // v1.4.25 W5 — `coachPrefs.defaultWindow` is the user's saved - // analysis-window preference. Merge it into the snapshot scope when - // the client didn't supply a per-conversation override; the override - // (header pill / sources rail) always wins. Keep the merge cheap so - // we don't accidentally widen narrow per-call scopes. - const prefsRow = await prisma.user.findUnique({ - where: { id: userId }, - 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 } - : scope; - const snapshot = await buildCoachSnapshot(userId, effectiveScope); - // v1.15.20 — the user-authored "about me" self-description (Settings → - // AI) rides the system prompt as a delimited, user-provided context - // block. Fail-open: a missing / undecryptable text yields null and the - // prompt is byte-identical to the pre-feature one. - // v1.16.0 — composed self-context: structured questionnaire fields - // plus age/gender merged in from the User profile. - const aboutMe = await getSelfContextTextForUser(userId, locale); - // v1.32.9 (Coach Guard II / G3) — the user's active medication doses, for the - // Grounding Ledger's `schedule` source AND the schedule-gated dose - // continuation exemption. Fail-open: a read failure yields an empty set, which - // keeps the Guard I phrase-anchored dose behaviour. - const scheduleDoses = await getScheduledDoseValues(userId).catch(() => []); - // v1.22 (B2/F6) — the canonical system-prompt module is owned elsewhere, so - // the two memory/action clauses are appended at assembly time here: one - // teaches the model to emit `---REMEMBER---` (durable "remind me" capture), - // one teaches the closed `---SUGGEST-ACTION---` confirm-card allowlist. Both - // are provider-neutral sentinels stripped from the prose before it streams. - // v1.22 (W6) — per-turn personalization: a sparse, hash-gated first name and - // an opener-archetype hint so multi-turn sessions vary. The turn index is the - // count of prior turns, so the name surfaces on ~1-in-3 turns, varied and - // never on a fixed cadence; both omit cleanly when no display name is set. - const turnIndex = priorTurns.length; - const firstName = firstNameFromDisplayName(prefsRow?.displayName ?? null); - const coachPersonalization = { - firstName, - mayUseName: - firstName != null && shouldUseNameForTurn(`${userId}:${turnIndex}`), - openerHint: openerArchetypeHint(`${userId}:${turnIndex}`, locale), - }; - const baseSystemPrompt = getCoachSystemPrompt( + // Conversation, context, chain, egress re-check, budget, and the streamed + // reply: `src/lib/ai/coach/turn/`. + return runCoachTurn({ + userId, locale, - coachPrefs, - aboutMe, - coachPersonalization, - ); - const turnContext = buildCoachTurnContext({ - priorTurns, - priorSummary, + signal: request.signal, + conversationId, message, + scope, guidedQuestion, + workoutId, + recheckCapability: coachCapabilityRefusal, }); - const { window, isFirstTurn, includeFullSnapshot } = turnContext; - // v1.11.1 — once a conversation grows past the history cap, refresh the - // rolling summary + extract durable facts off the request path. Fire-and- - // forget: this turn uses whatever summary is already on disk; the refresh - // makes the next long turn fresh. No-ops without an embedded worker. - // v1.16.1 — always-remember categories (allergies, intolerances, explicit - // self-reported diagnoses) must not wait for the >TURN_CAP memory refresh: - // a health-critical statement in the second message of a short chat used - // to never reach the fact store. The deterministic pattern pass is - // provider-free and deduped, so it fires on every user turn. - void storeDeterministicFacts({ - conversationId: workingConversationId, - userId, - message, - locale, - }).catch(() => { - // Fact capture must never break the chat turn; the >TURN_CAP LLM - // extraction remains as the catch-all on long conversations. - }); - if (turnContext.historyElided) { - void enqueueCoachMemoryRefresh({ - conversationId: workingConversationId, - userId, - // Coach memory prose is composed in de/en only — it is MODEL-FACING - // context (a rolling conversation summary + extracted durable facts), - // not user-facing prose, so English is the correct target for every - // locale without a reviewed body. The former `=== "en" ? "en" : "de"` - // binary composed and keyed a French account's memory in German. - locale: instructionLocale(locale), - }); - } - - // A workout launch is an optional narrowing of Coach, not permission to - // bypass the workouts module. Disabled modules contribute no read and no - // provider payload; the generic conversation still proceeds. - const workoutsEnabled = - !workoutId || (await isModuleEnabled(userId, "workouts")); - const workoutEvidence = - isFirstTurn && workoutId && workoutsEnabled - ? await buildWorkoutEvidenceSection(userId, workoutId) - : null; - if (workoutId) { - annotate({ - action: { name: "coach.launch.scoped" }, - meta: { - source: "workout", - // Whether the narrow actually resolved. A foreign / stale id finds - // nothing and the conversation simply proceeds unscoped. - resolved: workoutEvidence !== null, - firstTurn: isFirstTurn, - }, - }); - } - const { systemPrompt, userPrompt } = buildCoachProviderPrompts({ - baseSystemPrompt, - rememberAddendum: buildRememberAddendum(locale), - suggestActionAddendum: buildSuggestActionAddendum(locale), - languageName: LANGUAGE_NAMES[locale], - snapshotJson: snapshot.snapshotJson, - referenceGrounding: snapshot.referenceGrounding, - workoutEvidence, - turnContext, - }); - - // ── Provider chain ────────────────────────────────────────── - // v1.20.0 (F1) — provider capabilities select tool retrieval or the legacy - // snapshot-stuffing path. - const chain = await resolveProviderChain(userId); - if (chain.length === 0) { - const legacy = await resolveProvider(userId); - if (legacy.type === "none") { - annotate({ - action: { name: "insights.coach.noProvider" }, - }); - return streamProviderError({ - code: "coach.provider.none", - reason: "no_provider", - }); - } - chain.push({ providerType: "admin-openai", instance: legacy }); - } - - // v1.12.1 — consent gate before any server-managed external egress. When - // the chain could egress via the operator's global key, require an active - // `ai_coach` (or master `ai_full`) receipt. BYOK / local / ChatGPT-OAuth - // chains are the user's own egress and stay ungated. Throws - // ConsentRequiredError → apiHandler returns 403 + `consent.ai.required`. - await assertConsentForChain({ userId, chain, surface: "coach" }); - - // v1.20.0 (F1) — tool mode is on only when EVERY provider in the chain - // supports tool-calling, so whichever hop the fallback runner lands on can - // still serve tools. A chain that includes a no-tools provider (local / - // Ollama) falls back to the legacy snapshot-stuffing path verbatim, exactly - // as before — the snapshot builder stays alive as the no-tools floor. - const toolMode = chain.every((c) => c.instance.supportsTools !== false); - - // The last check before the budget reservation and the provider call: the - // `coach` capability again, at the egress site, so the call site names its - // capability whatever happens to the gate at the top. Inside one request - // the capability inputs are memoised, so this answers from the same - // resolution as the top gate; a fresh read at the wire belongs to the - // provider chokepoints, not to this route. - const refusedAtEgress = await coachCapabilityRefusal(); - if (refusedAtEgress) return refusedAtEgress; - - // v1.18.7 (SENIOR-DEV HIGH) — atomically RESERVE the day's budget before - // the provider call. The reservation increments the day's total by the - // per-call ceiling (`maxTokens`) in one upsert and returns the new total; - // concurrent requests serialise on the row so they cannot all pass the cap. - // Over-cap → 429 refusal frame, reservation already refunded. The actual - // token count is reconciled against this reservation after the call, - // including on empty / sentinel replies (their tokens were still burned). - // - // v1.20.0 (F1) — the tool loop makes up to MAX_ROUNDS provider round-trips, - // so reserve the per-call ceiling × the round count up front and reconcile - // the SUMMED actual tokens afterwards. The atomic reserve/reconcile - // primitives are unchanged; only the reserved amount scales. - // v1.21.0 (F1) — the daily ceiling is the OPERATOR's cost cap only when the - // chain egresses via the operator's own key (`admin-openai` primary). A - // ChatGPT-OAuth/Codex or BYOK chain runs on the user's OWN plan/key and costs - // the operator nothing, so it gets the generous user-plan ceiling — gating it - // on the operator-cost cap would lock the user out of a plan they pay for. - const dailyCap = resolveDailyCap(chain); - const reqDateKey = buildDateKey(); - const reservation = await reserveBudget( - userId, - toolMode - ? AI_BUDGETS.coach.maxTokens * MAX_ROUNDS - : AI_BUDGETS.coach.maxTokens, - reqDateKey, - dailyCap, - resolveCostOwner(chain), - "coach", - ); - if (!reservation.allowed) { - // v1.38.19 — say WHICH ceiling refused and WHOSE. - // `totalAfter` alone is the day's mixed total; on an operator refusal that - // is not the counter that tripped, and reading `totalAfter: 1200000` - // against a 200 000 ceiling is what turned the 06:42Z incident into a - // production log dig. Integers and two small enums — no key, host or model - // name is added. - annotate({ - action: { name: "coach.budget.exceeded" }, - meta: { - owner: reservation.owner, - surface: "coach", - limit: reservation.limit, - cap: dailyCap, - totalAfter: reservation.totalAfter, - operatorAfter: reservation.operatorAfter, - }, - }); - return streamProviderError({ code: "coach.budget.exceeded" }); - } - - // 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; - // v1.38.19 — typed as the chain's provider union, not a bare - // string: the budget reconcile attributes the turn's tokens to the cost - // owner of the hop named here. - let workingProviderType: ProviderChainType; - 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[] = []; - // The DATA INVENTORY manifest (per-domain sample counts) that rode this - // turn's tool-mode prompt. Hoisted out of the tool branch because the - // all-missed activation below needs it as a WIDENER: the counts were in - // front of the model, so a plain count restatement ("you've logged 42 BP - // readings") must stay grounded even on a turn whose every tool missed. - // Never an ACTIVATOR on its own — see the v1.32.1 note below. - let inventoryPayloads: 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 toolRequest = buildCoachToolRequest({ - systemPrompt, - toolModeAddendum: buildToolModeAddendum(locale), - focusHint: renderFocusHint(effectiveScope?.sources), - workoutEvidence, - dataInventory: renderDataInventory(inventory), - guidedBlock: turnContext.guidedBlock, - transcript: turnContext.transcript, - languageName: LANGUAGE_NAMES[locale], - }); - const loop = await runCoachToolLoop({ - userId, - providers: chain, - system: toolRequest.system, - messages: toolRequest.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, - // v1.22 (#89) — per-user response timeout for each tool-round call. - timeoutMs: aiResponseTimeoutMs, - }); - result = loop.result; - // The loop reports the hop it landed on as a bare string; it is - // assigned from `workingProvider.providerType` one frame up. - workingProviderType = loop.workingProviderType as ProviderChainType; - toolTrace = loop.toolTrace; - // v1.32.1 — the numeric verifier ACTIVATES only when this turn actually - // delivered figures the model was told to ground against: a pinned - // workout-evidence block or a present tool result. The DATA INVENTORY - // manifest (sample counts per domain) is NOT an activator — it rides - // every tool-mode prompt even on a turn where the model answered - // without calling a tool, and on that turn the base prompt deliberately - // carries no pre-computed figures (the model must fetch them), so the - // verifier must stay dormant and leave the prompt-level grounding rule - // as the backstop, exactly as on `main`. Activating it off the - // counts-only inventory would flag a snapshot figure the model cited - // without a fresh tool call as ungrounded (a real regression caught by - // the integration suite). When the turn IS active, the inventory counts - // still WIDEN the authoritative set so a plain count restatement - // ("you've logged 42 BP readings") stays grounded. - const presentToolPayloads = [ - ...(workoutEvidence === null ? [] : [workoutEvidence]), - // A miss carries no `data` but may carry `available` — the bounded - // out-of-window aggregate rule 3 lets the model cite. Ground it. - ...(loop.toolResults ?? []).map((r) => r.data ?? r.available), - ]; - inventoryPayloads = [inventory.entries]; - toolResultPayloads = - presentToolPayloads.length > 0 - ? [...presentToolPayloads, inventory.entries] - : []; - 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({ - surface: "coach", - userId, - providers: chain, - onDelta: () => { - streamedDeltas += 1; - }, - // v1.20.0 — the no-tools path still builds one assembled user turn (the - // transcript-flattening includes the current authoritative snapshot - // on every stateless no-tools request), so it ships as a single user - // message. The stable persona rides `system` and is 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 grades only when it delivered the - // full SNAPSHOT. Stateless provider requests always do; retaining the - // conditional keeps the builder contract explicit and testable. - if (includeFullSnapshot) { - noToolsSnapshotPayloads = [ - snapshot.sections, - ...(workoutEvidence === null ? [] : [workoutEvidence]), - ]; - } - } - } 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, 0, { - servedBy: null, - reservedOwner: reservation.owner, - }).catch(() => {}); - // #781 — the client walked away mid-generation. The request's abort - // signal is threaded into every provider call, so the teardown surfaces - // here as an abort-shaped failure with `request.signal` already flipped. - // Close the turn honestly instead of letting the user message dangle - // unanswered: persist an EMPTY assistant marker tagged - // `providerType: "cancelled"` (the same channel "refusal" and "nudge" - // already use for non-provider rows) so the thread shows the turn as - // interrupted and offers a retry on reload. The content is empty by - // construction — nothing guarded was produced, and anything the client - // ever SAW was persisted before the first token frame left, so no - // unguarded partial text is ever written. The reservation was refunded - // above; a retry pays exactly what a fresh turn pays. - if (request.signal.aborted) { - await appendMessage({ - conversationId: workingConversationId, - role: "assistant", - content: "", - providerType: "cancelled", - promptVersion: PROMPT_VERSION, - }).catch(() => { - // Marker persistence is best-effort — a failure leaves the - // dangling user turn, which is exactly the pre-#781 state. - }); - annotate({ - action: { name: "insights.coach.cancelled" }, - meta: { conversationId: workingConversationId }, - }); - return { ok: false, code: "coach.cancelled" }; - } - 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, - { servedBy: workingProviderType, reservedOwner: reservation.owner }, - ).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.keyvalues.parse_failed" }, - meta: { - kept: sentinel.keyValues.length, - reason: "empty_prose_after_strip", - promptVersion: PROMPT_VERSION, - }, - }); - 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. 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 turn. - // - // v1.30.25 — the capture writes `proposed`, not `active`. The block is - // model output, and the model reads a prompt carrying document-sourced - // text, so the write needs the same propose-then-confirm moat every other - // model-driven write already has. See `captureReminderFromSentinel`. - 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, locale, scheduleDoses); - 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 the editorial elision mark `[…]` and annotated (v1.32.14 — the withheld - // count also rides the provenance envelope to drive the per-message notice). - // 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. - // v1.32.9 (Coach Guard II / G2+G3) — grade the prose against the typed - // Grounding Ledger rather than a per-turn magnitude bag. Activation is - // unchanged from Guard I: the ledger is graded ONLY when THIS turn delivered - // fresh figures the model was told to ground against (a present tool result - // / pinned workout, or the full snapshot on the no-tools path). The - // cross-turn, memory, schedule, reference, and guided sources only WIDEN an - // active grading — they never ACTIVATE it, so a snapshot figure the model - // cited on a no-tool turn is still left alone (the v1.32.1 regression guard - // holds). Assistant prose is never a ledger source (D3). - // - // The turn that CALLED tools and got nothing back is graded too. It used to - // be the one turn the verifier sat out: a pure miss carries no `data` and no - // `available`, `runCoachToolLoop` drops that shape from `toolResults`, and - // the no-tools snapshot fallback below is populated only in the non-tool - // branch — so `activatingPayloads` came out empty and every figure in the - // reply shipped unchecked. That is backwards. A turn whose every tool - // reported `{ present: false }` is not an absence of evidence about the - // reply, it is evidence that the record holds nothing to cite, which is - // exactly when a fabricated figure is both likeliest and most harmful. - // - // `toolTrace` is the discriminant, because it records every tool that ran - // INCLUDING the pure misses. It separates the two cases the old condition - // conflated: tools ran and found nothing (grade — the model was told the - // record is empty and answered with numbers anyway), versus the model - // answered without calling a tool at all (stay dormant — the v1.32.1 - // regression, where the base prompt deliberately carries no pre-computed - // figures and flagging a legitimately recalled one was a real defect). - const missedEveryTool = - toolMode && toolTrace.length > 0 && toolResultPayloads.length === 0; - const activatingPayloads = - toolResultPayloads.length > 0 - ? toolResultPayloads - : missedEveryTool - ? inventoryPayloads - : noToolsSnapshotPayloads; - let groundedFigures: number[] = []; - // v1.32.14 — count of figures withheld from THIS reply (each rewritten to the - // elision mark). Hoisted out of the guard block so it can ride the provenance - // envelope and drive the quiet per-message notice. Stays 0 on a blocked turn - // (the guard is skipped, its fallback prose carries no figures). - let unverifiedStripped = 0; - // True when the guard stripped a figure on a turn whose every tool missed — - // i.e. the model put a number on a record that holds none. Drives the - // honest-replacement below. - let fabricatedOnEmptyRecord = false; - if (activatingPayloads.length > 0 || missedEveryTool) { - const ledger = buildGroundingLedger({ - toolPayloads: activatingPayloads, - priorToolFigures, - priorUserMessages, - // The self-context (goals / about-me) rides the system prompt on both - // paths — the ledger's `memory` source. - memoryTexts: aboutMe ? [aboutMe] : [], - scheduleDoses, - // Reference grounding rides the prompt only on the no-tools - // full-snapshot path; the guided block rides both paths. - referenceGrounding: - !toolMode && includeFullSnapshot ? snapshot.referenceGrounding : null, - guidedBlock: turnContext.guidedBlock, - }); - // Persist THIS turn's tool figures (bare magnitudes) so a LATER turn can - // recall them via the ledger — D3: the tool trace, never the prose. - groundedFigures = figuresForPersistence(ledger); - if (!outbound.block) { - const unverified = findUnverifiedCoachNumbersInLedger( - replyText, - ledger, - locale, - // On an all-missed turn an EMPTY ledger is the finding, not a reason - // to skip: nothing was retrievable, so nothing can reconcile. - { gradeAgainstEmptyLedger: missedEveryTool }, - ); - if (unverified.length > 0) { - const { prose: corrected, stripped } = stripUnverifiedNumbers( - replyText, - unverified, - ); - replyText = corrected; - unverifiedStripped = stripped; - fabricatedOnEmptyRecord = missedEveryTool && stripped > 0; - 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, - }, - }); - } - } - } - - // A stripped reply on an empty record is not worth sending. Eliding the - // digits leaves the sentences that carried them — "your sleep averaged […] - // minutes, up from […]" still asserts a series, an average and a direction - // for a metric the tools just reported has no readings at all. The false - // claim survives the strip; only its precision goes. So replace the turn - // with the honest answer instead of shipping the elided mush. - // - // Scoped tightly: this fires only when every tool missed AND the guard - // actually stripped something. A reply that cited a legitimately grounded - // figure (a prior turn's tool result, an inventory count) reconciles - // against the ledger, strips nothing, and is left alone. - // - // REPLACE, not withhold — the same policy the outbound screen uses on this - // surface, for the same reason: the user is waiting on a synchronous answer - // and silence reads as a failure. - // - // Accepted trade-off: a reply mixing one grounded recall with one - // fabrication is replaced wholesale, losing the good half. That is the - // right way round — the alternative leaves an invented figure's sentence - // on screen. - if (fabricatedOnEmptyRecord) { - replyText = getServerTranslator(locale).t("coach.noRecordedData"); - // The reply now carries no figures at all, so the withheld-figure notice - // would be describing prose the reader can no longer see. The replacement - // copy states the same thing in plain words. - unverifiedStripped = 0; - annotate({ - action: { name: "coach.prose.empty_record_replaced" }, - meta: { promptVersion: PROMPT_VERSION }, - }); - await auditLog("insights.coach.empty_record_replaced", { - userId, - details: { conversationId: workingConversationId }, - }); - } - - // 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, - })), - } - : {}), - // v1.32.9 (Coach Guard II / G2) — persist THIS turn's tool figures so a - // later turn's Grounding Ledger can recall them (D3-safe: tool trace, not - // prose). Dropped on a blocked turn (its figures rode the discarded reply). - ...(!outbound.block && groundedFigures.length > 0 - ? { groundedFigures } - : {}), - // v1.32.14 — the count of figures the grounding guard withheld this turn, - // so the quiet "some figures couldn't be checked" notice renders under the - // bubble and survives a conversation reload. Count only, no values. Omitted - // when nothing was withheld. - ...(unverifiedStripped > 0 - ? { unverifiedFigures: unverifiedStripped } - : {}), - }; - 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: "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.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) => { - // 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); - // #781 — a client disconnect can also surface as an abort-shaped error - // on a path `produceReply`'s own catch does not own (e.g. a teardown - // rejection out of persistence). That is not a defect and nobody is - // listening: skip the GlitchTip forward and the error frame. - if (request.signal.aborted) { - annotate({ - action: { name: "insights.coach.cancelled" }, - meta: { conversationId: workingConversationId, lateAbort: true }, - }); - return; - } - // 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); - if (!providerError) { - // Not a tagged provider failure — a genuine server defect raised - // inside the open producer (e.g. a DB error in the post-stream - // `appendMessage` persistence). Record it on the wide event and - // forward it to GlitchTip so it stays visible instead of reading - // as a provider outage; the stream is open, so we still emit an - // error frame below rather than a 500. - getEvent()?.setError( - err instanceof Error ? err : new Error(String(err)), - ); - void reportCoachStreamDefect(err); - } - annotate({ - action: { name: "insights.coach.streamError" }, - meta: { - unwrapped: Boolean(providerError), - reported: !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: outcome.provenance }), - ); - // v1.18.1 (Workstream C) — additive `suggestion` frame. - if (outcome.suggestion) { - controller.enqueue( - encodeFrame({ type: "suggestion", suggestion: outcome.suggestion }), - ); - } - // v1.22 (F6) — additive `suggestedAction` frame. - if (outcome.action) { - controller.enqueue( - encodeFrame({ - type: "suggestedAction", - suggestedAction: outcome.action, - }), - ); - } - // v1.18.9 — additive `usage` envelope on the `done` frame. - controller.enqueue( - encodeFrame({ - type: "done", - conversationId: workingConversationId, - messageId: outcome.messageId, - usage: { - totalTokens: outcome.totalTokens || null, - model: outcome.model, - }, - }), - ); - }); - - return new Response(stream, { status: 200, headers: SSE_HEADERS }); -} - -/** - * Emit a refusal as a single `token` frame followed by `done`. No - * provider call, no persisted assistant message — the user message is - * still kept on disk so the rail shows the conversation history - * accurately. The user message landing on disk is a deliberate choice; - * the rail otherwise wouldn't show the user's attempt at all. - */ -async function streamRefusal(args: { - userId: string; - conversationId: string | undefined; - message: string; - refusalText: string; -}): Promise { - let conversationId = args.conversationId; - if (!conversationId) { - const created = await createConversation({ - userId: args.userId, - title: args.message, - }); - conversationId = created.id; - } else { - const owned = await prisma.coachConversation.findFirst({ - where: { id: conversationId, userId: args.userId }, - select: { id: true }, - }); - if (!owned) { - throw new HttpError(404, "coach.conversation.notFound"); - } - } - - await appendMessage({ - conversationId, - role: "user", - content: args.message, - }); - const refusalMessage = await appendMessage({ - conversationId, - role: "assistant", - content: args.refusalText, - metricSource: { windows: [], metrics: ["general"] }, - providerType: "refusal", - promptVersion: PROMPT_VERSION, - }); - - const stream = createSseStream((controller) => { - controller.enqueue(encodeFrame({ type: "token", token: args.refusalText })); - controller.enqueue( - encodeFrame({ - type: "done", - conversationId, - messageId: refusalMessage.id, - }), - ); - }); - - return new Response(stream, { status: 200, headers: SSE_HEADERS }); -} - -/** - * v1.21.3 — classify a provider error that bubbled out of the chain runner - * un-wrapped (i.e. not an `AllProvidersFailedError`). The provider clients tag - * their thrown errors with `upstream` + `httpStatus`; a tagged error is a - * provider failure that must surface a graceful `coach.provider.*` frame rather - * than rethrow into an HTTP 500. Returns `null` for anything that is NOT a - * recognisable provider error (a real server bug), so those keep the 500 + - * GlitchTip path. Status mapping mirrors `AllProvidersFailedError`: 401/403 → - * credential_expired, 429 → rate_limited, everything else → unavailable. - */ -function classifyBubblingProviderError(err: unknown): { - code: "credential_expired" | "rate_limited" | "unavailable"; - httpStatus: number | null; -} | null { - if (err === null || typeof err !== "object") return null; - const e = err as { upstream?: unknown; httpStatus?: unknown }; - const hasUpstreamTag = typeof e.upstream === "string"; - const status = typeof e.httpStatus === "number" ? e.httpStatus : null; - // Require the wire tag — a bare `{ httpStatus }` from unrelated code must not - // be swallowed as a provider outage. - if (!hasUpstreamTag) return null; - if (status === 401 || status === 403) { - return { code: "credential_expired", httpStatus: status }; - } - if (status === 429) return { code: "rate_limited", httpStatus: status }; - return { code: "unavailable", httpStatus: status }; -} - -/** - * Fire-and-forget GlitchTip forward for a genuine server defect that - * surfaces INSIDE the open SSE producer — e.g. a Prisma failure in the - * post-stream `appendMessage` persistence, which runs after the provider - * call and therefore outside its try/catch. Once the stream is open the - * route can no longer return a 500, so without this such a defect would - * read as a generic provider outage in error tracking rather than the - * server bug it is. Mirrors the api-handler / worker forwarders: dynamic - * import (no cycle, no startup cost), redacted message + stack, and it - * NEVER throws so a sink failure cannot mask the original error. - */ -async function reportCoachStreamDefect(err: unknown): Promise { - const e = - err instanceof Error ? err : new Error("Unknown coach stream error"); - const message = redactSecrets(`[insights.coach.stream] ${e.message}`); - console.error("[coach-stream]", message, e); - try { - const [{ getGlitchtipSettings }, { sendGlitchtipEvent }] = - await Promise.all([ - import("@/lib/monitoring-settings"), - import("@/lib/monitoring/glitchtip"), - ]); - const settings = await getGlitchtipSettings(); - if (!settings.glitchtipEnabled || !settings.glitchtipDsn) return; - await sendGlitchtipEvent({ - dsn: settings.glitchtipDsn, - input: { - environment: settings.glitchtipEnvironment || "production", - message, - level: "error", - type: e.name || "Error", - stack: redactOptional(e.stack), - sourceTag: "healthlog-api-handler", - }, - }); - } catch { - /* the reporter must never throw */ - } -} - -function streamProviderError(args: { - code: string; - /** The capability reason, when the refusal came from one. */ - reason?: AiUnavailableReason; -}): Response { - const stream = createSseStream((controller) => { - controller.enqueue( - encodeFrame({ - type: "error", - code: args.code, - message: args.code, - ...(args.reason ? { reason: args.reason } : {}), - }), - ); - }); - // Status 200 so the streaming client reads the SSE body and parses - // the structured `error` frame (HTTP-status branches drop the - // structured code on the floor). - return new Response(stream, { status: 200, headers: SSE_HEADERS }); } // Idempotency is intentionally NOT applied to this SSE-streaming route. diff --git a/src/lib/ai/coach/__tests__/coach-locale-no-german-default.test.ts b/src/lib/ai/coach/__tests__/coach-locale-no-german-default.test.ts index 0572940cf..c295304c1 100644 --- a/src/lib/ai/coach/__tests__/coach-locale-no-german-default.test.ts +++ b/src/lib/ai/coach/__tests__/coach-locale-no-german-default.test.ts @@ -13,13 +13,24 @@ * which is precisely what makes it survive a refactor of the logic around it. */ import { describe, it, expect } from "vitest"; -import { readFileSync } from "node:fs"; +import { readdirSync, readFileSync } from "node:fs"; import { resolve } from "node:path"; const ROOT = resolve(__dirname, "../../../.."); +/** + * The chat route hands each turn to the pipeline under `lib/ai/coach/turn/`, + * so every module there is guarded — read from the directory, so a step added + * later is covered without editing this list. + */ +const TURN_PIPELINE_FILES = readdirSync(resolve(ROOT, "lib/ai/coach/turn")) + .filter((name) => name.endsWith(".ts")) + .sort() + .map((name) => `lib/ai/coach/turn/${name}`); + const GUARDED_FILES = [ "app/api/insights/chat/route.ts", + ...TURN_PIPELINE_FILES, "app/api/insights/narrative/route.ts", "lib/ai/coach/snapshot.ts", "lib/ai/coach/memory-snapshot.ts", @@ -62,6 +73,12 @@ describe("no German-default locale binary", () => { expect(readCode(rel)).not.toMatch(DE_DEFAULT_NULLISH); }); + it("finds the turn pipeline it guards", () => { + // An empty directory read would leave the pipeline unguarded and green. + expect(TURN_PIPELINE_FILES).toContain("lib/ai/coach/turn/context.ts"); + expect(TURN_PIPELINE_FILES).toContain("lib/ai/coach/turn/pipeline.ts"); + }); + it("recognises the shape it bans", () => { // Guard the guard: the patterns must actually match the bug they describe, // so a typo cannot turn this file into a silent pass. diff --git a/src/lib/ai/coach/__tests__/data-field-injection.test.ts b/src/lib/ai/coach/__tests__/data-field-injection.test.ts index c86c5ead6..2c71111c9 100644 --- a/src/lib/ai/coach/__tests__/data-field-injection.test.ts +++ b/src/lib/ai/coach/__tests__/data-field-injection.test.ts @@ -225,9 +225,11 @@ describe("fence escape — a data field cannot close its container", () => { * assemblies actually route through the fence. */ describe("fence call sites", () => { - it("the Coach chat route delegates SNAPSHOT assembly to the fenced builder", () => { + it("the Coach chat turn delegates SNAPSHOT assembly to the fenced builder", () => { + // The chat route hands the turn to the pipeline under + // `src/lib/ai/coach/turn/`; the prompt is assembled in its context step. const source = readFileSync( - resolve(process.cwd(), "src/app/api/insights/chat/route.ts"), + resolve(process.cwd(), "src/lib/ai/coach/turn/context.ts"), "utf-8", ); expect(source).toContain("buildCoachProviderPrompts({"); diff --git a/src/lib/ai/coach/turn/budget.ts b/src/lib/ai/coach/turn/budget.ts new file mode 100644 index 000000000..0d9430cce --- /dev/null +++ b/src/lib/ai/coach/turn/budget.ts @@ -0,0 +1,146 @@ +/** + * The day's token budget for one Coach turn: reserve before the provider + * call, settle against the actual count after it, refund on failure. + */ +import { annotate } from "@/lib/logging/context"; +import type { ProviderChainType } from "@/lib/ai/provider-chain"; +import { AI_BUDGETS } from "@/lib/ai/ai-budgets"; +import { + buildDateKey, + reserveBudget, + reconcileSpend, + resolveCostOwner, + resolveDailyCap, +} from "@/lib/ai/coach/budget"; +import { MAX_ROUNDS } from "@/lib/ai/coach/tools"; + +import type { TurnChain } from "./chain"; +import { streamProviderError } from "./sse"; + +export interface TurnReservation { + reserved: number; + owner: Awaited>["owner"]; + dateKey: string; +} + +export async function reserveTurnBudget(args: { + userId: string; + chain: TurnChain; + toolMode: boolean; +}): Promise< + { ok: true; reservation: TurnReservation } | { ok: false; response: Response } +> { + const { userId, chain, toolMode } = args; + // v1.18.7 (SENIOR-DEV HIGH) — atomically RESERVE the day's budget before + // the provider call. The reservation increments the day's total by the + // per-call ceiling (`maxTokens`) in one upsert and returns the new total; + // concurrent requests serialise on the row so they cannot all pass the cap. + // Over-cap → 429 refusal frame, reservation already refunded. The actual + // token count is reconciled against this reservation after the call, + // including on empty / sentinel replies (their tokens were still burned). + // + // v1.20.0 (F1) — the tool loop makes up to MAX_ROUNDS provider round-trips, + // so reserve the per-call ceiling × the round count up front and reconcile + // the SUMMED actual tokens afterwards. The atomic reserve/reconcile + // primitives are unchanged; only the reserved amount scales. + // v1.21.0 (F1) — the daily ceiling is the OPERATOR's cost cap only when the + // chain egresses via the operator's own key (`admin-openai` primary). A + // ChatGPT-OAuth/Codex or BYOK chain runs on the user's OWN plan/key and costs + // the operator nothing, so it gets the generous user-plan ceiling — gating it + // on the operator-cost cap would lock the user out of a plan they pay for. + const dailyCap = resolveDailyCap(chain); + const reqDateKey = buildDateKey(); + const reservation = await reserveBudget( + userId, + toolMode + ? AI_BUDGETS.coach.maxTokens * MAX_ROUNDS + : AI_BUDGETS.coach.maxTokens, + reqDateKey, + dailyCap, + resolveCostOwner(chain), + "coach", + ); + if (!reservation.allowed) { + // v1.38.19 — say WHICH ceiling refused and WHOSE. + // `totalAfter` alone is the day's mixed total; on an operator refusal that + // is not the counter that tripped, and reading `totalAfter: 1200000` + // against a 200 000 ceiling is what turned the 06:42Z incident into a + // production log dig. Integers and two small enums — no key, host or model + // name is added. + annotate({ + action: { name: "coach.budget.exceeded" }, + meta: { + owner: reservation.owner, + surface: "coach", + limit: reservation.limit, + cap: dailyCap, + totalAfter: reservation.totalAfter, + operatorAfter: reservation.operatorAfter, + }, + }); + return { + ok: false, + response: streamProviderError({ code: "coach.budget.exceeded" }), + }; + } + return { + ok: true, + reservation: { + reserved: reservation.reserved, + owner: reservation.owner, + dateKey: reqDateKey, + }, + }; +} + +/** + * The provider chain failed outright — no tokens were billed, so refund the + * full reservation before surfacing the error frame. + */ +export async function refundReservation( + userId: string, + reservation: TurnReservation, +): Promise { + await reconcileSpend( + userId, + reservation.reserved, + 0, + reservation.dateKey, + 0, + { + servedBy: null, + reservedOwner: reservation.owner, + }, + ).catch(() => {}); +} + +/** + * 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. + */ +export async function settleReservation( + userId: string, + reservation: TurnReservation, + spent: { + totalTokens: number; + cachedTokens: number; + servedBy: ProviderChainType; + }, +): Promise { + await reconcileSpend( + userId, + reservation.reserved, + spent.totalTokens, + reservation.dateKey, + spent.cachedTokens, + { servedBy: spent.servedBy, reservedOwner: reservation.owner }, + ).catch(() => { + // Ledger reconcile is best-effort; a failure leaves the conservative + // reservation in place (never an undercount) and never breaks the turn. + }); +} diff --git a/src/lib/ai/coach/turn/cards.ts b/src/lib/ai/coach/turn/cards.ts new file mode 100644 index 000000000..3c5a4b54d --- /dev/null +++ b/src/lib/ai/coach/turn/cards.ts @@ -0,0 +1,89 @@ +/** + * The cards a Coach reply may carry beside its prose: a gated cadence + * suggestion (with its cooldown stamp) and a confirm-to-apply action. + * Both are dropped when the outbound screen replaced the reply. + */ +import { prisma } from "@/lib/db"; +import { annotate } from "@/lib/logging/context"; +import type { CoachSuggestion } from "@/lib/ai/coach/types"; +import { gateSuggestion } from "@/lib/ai/coach/suggest-gate"; +import type { CoachSuggestedAction } from "@/lib/ai/coach/suggest-action"; +import { + DEFAULT_REMINDER_SUGGESTION_PREFS, + type CoachPrefs, +} from "@/lib/validations/coach-prefs"; + +import type { GuardedReply } from "./reply-guards"; + +export async function surfaceCards(args: { + userId: string; + coachPrefs: CoachPrefs; + reply: GuardedReply; +}): Promise<{ + suggestion: CoachSuggestion | null; + action: CoachSuggestedAction | null; +}> { + const { userId, coachPrefs, reply } = args; + const { suggestParse, actionParse, outboundBlocked } = reply; + + let surfacedSuggestion: CoachSuggestion | null = null; + if (!outboundBlocked && 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 (!outboundBlocked && actionParse.action) { + surfacedAction = actionParse.action; + annotate({ + action: { name: "coach.action.suggested" }, + meta: { actionType: actionParse.action.actionType }, + }); + } + + return { suggestion: surfacedSuggestion, action: surfacedAction }; +} diff --git a/src/lib/ai/coach/turn/chain.ts b/src/lib/ai/coach/turn/chain.ts new file mode 100644 index 000000000..e2bb7741b --- /dev/null +++ b/src/lib/ai/coach/turn/chain.ts @@ -0,0 +1,54 @@ +/** + * Resolve the provider chain a Coach turn will run on, gate it on consent, + * and decide between tool retrieval and the legacy snapshot path. + */ +import { annotate } from "@/lib/logging/context"; +import { resolveProviderChain, resolveProvider } from "@/lib/ai/provider"; +import { assertConsentForChain } from "@/lib/ai/consent-guard"; + +import { streamProviderError } from "./sse"; + +export type TurnChain = Awaited>; + +export async function resolveTurnChain( + userId: string, +): Promise< + | { ok: true; chain: TurnChain; toolMode: boolean } + | { ok: false; response: Response } +> { + // v1.20.0 (F1) — provider capabilities select tool retrieval or the legacy + // snapshot-stuffing path. + const chain = await resolveProviderChain(userId); + if (chain.length === 0) { + const legacy = await resolveProvider(userId); + if (legacy.type === "none") { + annotate({ + action: { name: "insights.coach.noProvider" }, + }); + return { + ok: false, + response: streamProviderError({ + code: "coach.provider.none", + reason: "no_provider", + }), + }; + } + chain.push({ providerType: "admin-openai", instance: legacy }); + } + + // v1.12.1 — consent gate before any server-managed external egress. When + // the chain could egress via the operator's global key, require an active + // `ai_coach` (or master `ai_full`) receipt. BYOK / local / ChatGPT-OAuth + // chains are the user's own egress and stay ungated. Throws + // ConsentRequiredError → apiHandler returns 403 + `consent.ai.required`. + await assertConsentForChain({ userId, chain, surface: "coach" }); + + // v1.20.0 (F1) — tool mode is on only when EVERY provider in the chain + // supports tool-calling, so whichever hop the fallback runner lands on can + // still serve tools. A chain that includes a no-tools provider (local / + // Ollama) falls back to the legacy snapshot-stuffing path verbatim, exactly + // as before — the snapshot builder stays alive as the no-tools floor. + const toolMode = chain.every((c) => c.instance.supportsTools !== false); + + return { ok: true, chain, toolMode }; +} diff --git a/src/lib/ai/coach/turn/context.ts b/src/lib/ai/coach/turn/context.ts new file mode 100644 index 000000000..a085e2d24 --- /dev/null +++ b/src/lib/ai/coach/turn/context.ts @@ -0,0 +1,223 @@ +/** + * Assemble everything a Coach turn tells the model: the user's prefs, the + * snapshot, the self-context, the scheduled doses, the system prompt, the + * conversation window, an optional pinned workout, and the fenced provider + * prompts. Also starts the per-turn memory capture off the request path. + */ +import { prisma } from "@/lib/db"; +import { annotate } from "@/lib/logging/context"; +import { isModuleEnabled } from "@/lib/modules/gate"; +import type { Locale } from "@/lib/i18n/config"; +import { localeLanguageNames as LANGUAGE_NAMES } from "@/lib/i18n/config"; +import type { CoachScope } from "@/lib/ai/coach/types"; +import { enqueueCoachMemoryRefresh } from "@/lib/ai/coach/coach-memory-shared"; +import { instructionLocale } from "@/lib/ai/prompts/output-language"; +import { storeDeterministicFacts } from "@/lib/ai/coach/facts"; +import { getCoachSystemPrompt } from "@/lib/ai/coach/system-prompt"; +import { + openerArchetypeHint, + shouldUseNameForTurn, + firstNameFromDisplayName, +} from "@/lib/ai/prompts/opener-archetype"; +import { getSelfContextTextForUser } from "@/lib/ai/coach/about-me"; +import { + buildCoachSnapshot, + type CoachSnapshotResult, +} from "@/lib/ai/coach/snapshot"; +import { + buildCoachProviderPrompts, + buildCoachTurnContext, + type CoachTurnContext, +} from "@/lib/ai/coach/chat-request-builder"; +import { buildWorkoutEvidenceSection } from "@/lib/ai/coach/workout-evidence-builder"; +import { getScheduledDoseValues } from "@/lib/medications/scheduled-doses"; +import { buildRememberAddendum } from "@/lib/ai/coach/reminders"; +import { buildSuggestActionAddendum } from "@/lib/ai/coach/suggest-action"; +import { + parseCoachPrefs, + type CoachPrefs, +} from "@/lib/validations/coach-prefs"; + +import type { TurnConversation } from "./types"; + +export interface TurnContext { + coachPrefs: CoachPrefs; + /** + * 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. + */ + aiResponseTimeoutMs: number; + effectiveScope: CoachScope | undefined; + snapshot: CoachSnapshotResult; + aboutMe: string | null; + scheduleDoses: Awaited>; + turnContext: CoachTurnContext; + workoutEvidence: Record | null; + systemPrompt: string; + userPrompt: string; +} + +export async function assembleTurnContext(args: { + userId: string; + locale: Locale; + message: string; + scope: CoachScope | undefined; + guidedQuestion: string | undefined; + workoutId: string | undefined; + conversation: TurnConversation; +}): Promise { + const { userId, locale, message, scope, guidedQuestion, workoutId } = args; + const { conversationId, priorTurns, priorSummary } = args.conversation; + + // Build the prompt: system + (optional) snapshot + recent history + + // the new user message. v1.4.23 H4 — fold per-user prefs into the + // system-prompt prefix; the snapshot builder reads the same prefs + // separately so excluded metrics never even leave the DB. + // + // v1.4.25 W5 — `coachPrefs.defaultWindow` is the user's saved + // analysis-window preference. Merge it into the snapshot scope when + // the client didn't supply a per-conversation override; the override + // (header pill / sources rail) always wins. Keep the merge cheap so + // we don't accidentally widen narrow per-call scopes. + const prefsRow = await prisma.user.findUnique({ + where: { id: userId }, + 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. + aiResponseTimeoutSeconds: true, + }, + }); + const coachPrefs = parseCoachPrefs(prefsRow?.coachPrefsJson); + // v1.22 (#89) — 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 } + : scope; + const snapshot = await buildCoachSnapshot(userId, effectiveScope); + // v1.15.20 — the user-authored "about me" self-description (Settings → + // AI) rides the system prompt as a delimited, user-provided context + // block. Fail-open: a missing / undecryptable text yields null and the + // prompt is byte-identical to the pre-feature one. + // v1.16.0 — composed self-context: structured questionnaire fields + // plus age/gender merged in from the User profile. + const aboutMe = await getSelfContextTextForUser(userId, locale); + // v1.32.9 (Coach Guard II / G3) — the user's active medication doses, for the + // Grounding Ledger's `schedule` source AND the schedule-gated dose + // continuation exemption. Fail-open: a read failure yields an empty set, which + // keeps the Guard I phrase-anchored dose behaviour. + const scheduleDoses = await getScheduledDoseValues(userId).catch(() => []); + // v1.22 (B2/F6) — the canonical system-prompt module is owned elsewhere, so + // the two memory/action clauses are appended at assembly time here: one + // teaches the model to emit `---REMEMBER---` (durable "remind me" capture), + // one teaches the closed `---SUGGEST-ACTION---` confirm-card allowlist. Both + // are provider-neutral sentinels stripped from the prose before it streams. + // v1.22 (W6) — per-turn personalization: a sparse, hash-gated first name and + // an opener-archetype hint so multi-turn sessions vary. The turn index is the + // count of prior turns, so the name surfaces on ~1-in-3 turns, varied and + // never on a fixed cadence; both omit cleanly when no display name is set. + const turnIndex = priorTurns.length; + const firstName = firstNameFromDisplayName(prefsRow?.displayName ?? null); + const coachPersonalization = { + firstName, + mayUseName: + firstName != null && shouldUseNameForTurn(`${userId}:${turnIndex}`), + openerHint: openerArchetypeHint(`${userId}:${turnIndex}`, locale), + }; + const baseSystemPrompt = getCoachSystemPrompt( + locale, + coachPrefs, + aboutMe, + coachPersonalization, + ); + const turnContext = buildCoachTurnContext({ + priorTurns, + priorSummary, + message, + guidedQuestion, + }); + const { isFirstTurn } = turnContext; + // v1.11.1 — once a conversation grows past the history cap, refresh the + // rolling summary + extract durable facts off the request path. Fire-and- + // forget: this turn uses whatever summary is already on disk; the refresh + // makes the next long turn fresh. No-ops without an embedded worker. + // v1.16.1 — always-remember categories (allergies, intolerances, explicit + // self-reported diagnoses) must not wait for the >TURN_CAP memory refresh: + // a health-critical statement in the second message of a short chat used + // to never reach the fact store. The deterministic pattern pass is + // provider-free and deduped, so it fires on every user turn. + void storeDeterministicFacts({ + conversationId, + userId, + message, + locale, + }).catch(() => { + // Fact capture must never break the chat turn; the >TURN_CAP LLM + // extraction remains as the catch-all on long conversations. + }); + if (turnContext.historyElided) { + void enqueueCoachMemoryRefresh({ + conversationId, + userId, + // Coach memory prose is composed in de/en only — it is MODEL-FACING + // context (a rolling conversation summary + extracted durable facts), + // not user-facing prose, so English is the correct target for every + // locale without a reviewed body. The former `=== "en" ? "en" : "de"` + // binary composed and keyed a French account's memory in German. + locale: instructionLocale(locale), + }); + } + + // A workout launch is an optional narrowing of Coach, not permission to + // bypass the workouts module. Disabled modules contribute no read and no + // provider payload; the generic conversation still proceeds. + const workoutsEnabled = + !workoutId || (await isModuleEnabled(userId, "workouts")); + const workoutEvidence = + isFirstTurn && workoutId && workoutsEnabled + ? await buildWorkoutEvidenceSection(userId, workoutId) + : null; + if (workoutId) { + annotate({ + action: { name: "coach.launch.scoped" }, + meta: { + source: "workout", + // Whether the narrow actually resolved. A foreign / stale id finds + // nothing and the conversation simply proceeds unscoped. + resolved: workoutEvidence !== null, + firstTurn: isFirstTurn, + }, + }); + } + const { systemPrompt, userPrompt } = buildCoachProviderPrompts({ + baseSystemPrompt, + rememberAddendum: buildRememberAddendum(locale), + suggestActionAddendum: buildSuggestActionAddendum(locale), + languageName: LANGUAGE_NAMES[locale], + snapshotJson: snapshot.snapshotJson, + referenceGrounding: snapshot.referenceGrounding, + workoutEvidence, + turnContext, + }); + + return { + coachPrefs, + aiResponseTimeoutMs, + effectiveScope, + snapshot, + aboutMe, + scheduleDoses, + turnContext, + workoutEvidence, + systemPrompt, + userPrompt, + }; +} diff --git a/src/lib/ai/coach/turn/conversation.ts b/src/lib/ai/coach/turn/conversation.ts new file mode 100644 index 000000000..ef511d9cd --- /dev/null +++ b/src/lib/ai/coach/turn/conversation.ts @@ -0,0 +1,159 @@ +/** + * Resolve the conversation a Coach turn belongs to: load (and fence-check) an + * existing one, re-screen its stored user turns for a replayed injection, or + * create a new one. Then record the user's turn. + */ +import { HttpError } from "@/lib/api-handler"; +import { annotate } from "@/lib/logging/context"; +import { auditLog } from "@/lib/auth/audit"; +import type { Locale } from "@/lib/i18n/config"; +import { getServerTranslator } from "@/lib/i18n/server-translator"; +import { + appendMessage, + createConversation, + fetchConversationWithMessages, +} from "@/lib/ai/coach/persistence"; +import { detectRefusal } from "@/lib/ai/coach/refusal"; +import type { CoachTurn } from "@/lib/ai/coach/chat-request-builder"; + +import { streamRefusal } from "./sse"; +import type { TurnConversation } from "./types"; + +export async function resolveTurnConversation(args: { + userId: string; + conversationId: string | undefined; + message: string; + locale: Locale; +}): Promise<{ conversation: TurnConversation } | { refusal: Response }> { + const { userId, conversationId, message, locale } = args; + + if (!conversationId) { + const created = await createConversation({ userId, title: message }); + return { + conversation: { + conversationId: created.id, + priorTurns: [], + priorUserMessages: [], + priorToolFigures: [], + priorSummary: null, + }, + }; + } + + // v1.29.x (S7) — the load-bearing prompt-injection fence, now a DUAL + // predicate. Untrusted document text may only enter an LLM prompt on the + // fenced pipeline (no tools, no snapshot). This tool route must never load a + // fenced conversation: + // PRIMARY — `documentScoped: false` in the WHERE. The sticky flag is set + // at fenced-creation / first-attach and NEVER cleared, so a + // conversation that has EVER held a document 404s here forever. + // BACKSTOP — even so, assert zero live attachments. `documentScoped: false` + // with an attachment row present is flag/join drift (an + // invariant broke somewhere): fail closed, loudly. + // A doc turn is routed by the client to `/api/insights/chat/fenced` (or the + // single-doc sheet endpoint). Should a fenced id ever reach THIS route the + // fetch returns nothing and the turn 404s rather than running an injected + // instruction against the coach's write tools. Do not relax. + const existing = await fetchConversationWithMessages(userId, conversationId, { + documentScoped: false, + }); + if (!existing) { + // 404, not 403 — never reveal cross-user / cross-mode existence + throw new HttpError(404, "coach.conversation.notFound"); + } + if (existing.attachmentCount > 0) { + // Drift alarm — telemetry AND an audit row (not routine telemetry). + annotate({ + action: { name: "insights.coach.fence_drift" }, + meta: { conversationId: existing.id }, + }); + await auditLog("insights.coach.fence_drift", { + userId, + details: { conversationId: existing.id }, + }); + throw new HttpError(404, "coach.conversation.notFound"); + } + // #781 — cancelled-turn markers (`providerType: "cancelled"`, empty body) + // are UI rows, not conversation content: keep them out of the provider + // transcript so an interrupted turn adds no empty assistant line to the + // prompt. They stay in the DTO the client renders. + const priorTurns: CoachTurn[] = existing.messages + .filter((m) => m.providerType !== "cancelled") + .map((m) => ({ + role: m.role, + content: m.content, + })); + // Ledger cross-turn sources (D3-safe): user-authored numbers + the + // persisted tool figures of prior turns. Assistant PROSE is never read. + const priorUserMessages = existing.messages + .filter((m) => m.role === "user") + .map((m) => m.content); + const priorToolFigures = existing.messages.flatMap((m) => + m.metricSource?.groundedFigures ? [...m.metricSource.groundedFigures] : [], + ); + + // v1.4.43 W13 M-3 — replay-injection guard. `detectRefusal` runs + // only on the inbound `message` per turn, so an injection that + // slipped past the regex bank on a previous turn would re-enter + // the prompt every reply. Re-run the detector against every + // user-turn re-loaded from DB; on a hit, short-circuit the SSE + // with a refusal AND drop an `insights.coach.replay_injection` + // row so the failure case is observable. The audit row carries + // the conversation id (server-owned), the turn index (no PII) + // and the matched reason — never the message content. v1.4.43 + // W10 simplifier-L-2 — action name follows the `.` + // convention (no `audit.` prefix; no `.replay-injection` dash). + for (let i = 0; i < priorTurns.length; i++) { + const turn = priorTurns[i]; + if (turn.role !== "user") continue; + const replayed = detectRefusal({ message: turn.content, locale }); + if (!replayed.refuse) continue; + annotate({ + action: { name: "insights.coach.replay_injection" }, + meta: { reason: replayed.reason, turnIndex: i }, + }); + await auditLog("insights.coach.replay_injection", { + userId, + details: { + conversationId: existing.id, + turnIndex: i, + reason: replayed.reason, + }, + }); + return { + refusal: await streamRefusal({ + userId, + conversationId: existing.id, + message, + refusalText: + replayed.message ?? + getServerTranslator(locale).t("coach.refusal.conversationPoisoned"), + }), + }; + } + + return { + conversation: { + conversationId: existing.id, + priorTurns, + priorUserMessages, + priorToolFigures, + priorSummary: existing.summary ?? null, + }, + }; +} + +/** + * Persist the user's turn first so it's safely on disk regardless of + * whether the provider call succeeds. + */ +export async function persistUserTurn( + conversationId: string, + message: string, +): Promise { + await appendMessage({ + conversationId, + role: "user", + content: message, + }); +} diff --git a/src/lib/ai/coach/turn/errors.ts b/src/lib/ai/coach/turn/errors.ts new file mode 100644 index 000000000..3a60c07c6 --- /dev/null +++ b/src/lib/ai/coach/turn/errors.ts @@ -0,0 +1,130 @@ +/** + * Error classification for a Coach chat turn: which thrown errors are + * provider failures (a graceful `coach.provider.*` frame) and which are + * server defects (forwarded to GlitchTip and still answered with a frame, + * because the stream is already open). + * + * Imports nothing from the tool registry or the snapshot builder. + */ +import { annotate, getEvent } from "@/lib/logging/context"; +import { redactOptional, redactSecrets } from "@/lib/logging/redact"; + +import type { TurnEmitter } from "./types"; + +/** + * v1.21.3 — classify a provider error that bubbled out of the chain runner + * un-wrapped (i.e. not an `AllProvidersFailedError`). The provider clients tag + * their thrown errors with `upstream` + `httpStatus`; a tagged error is a + * provider failure that must surface a graceful `coach.provider.*` frame rather + * than rethrow into an HTTP 500. Returns `null` for anything that is NOT a + * recognisable provider error (a real server bug), so those keep the 500 + + * GlitchTip path. Status mapping mirrors `AllProvidersFailedError`: 401/403 → + * credential_expired, 429 → rate_limited, everything else → unavailable. + */ +export function classifyBubblingProviderError(err: unknown): { + code: "credential_expired" | "rate_limited" | "unavailable"; + httpStatus: number | null; +} | null { + if (err === null || typeof err !== "object") return null; + const e = err as { upstream?: unknown; httpStatus?: unknown }; + const hasUpstreamTag = typeof e.upstream === "string"; + const status = typeof e.httpStatus === "number" ? e.httpStatus : null; + // Require the wire tag — a bare `{ httpStatus }` from unrelated code must not + // be swallowed as a provider outage. + if (!hasUpstreamTag) return null; + if (status === 401 || status === 403) { + return { code: "credential_expired", httpStatus: status }; + } + if (status === 429) return { code: "rate_limited", httpStatus: status }; + return { code: "unavailable", httpStatus: status }; +} + +/** + * Fire-and-forget GlitchTip forward for a genuine server defect that + * surfaces INSIDE the open SSE producer — e.g. a Prisma failure in the + * post-stream `appendMessage` persistence, which runs after the provider + * call and therefore outside its try/catch. Once the stream is open the + * route can no longer return a 500, so without this such a defect would + * read as a generic provider outage in error tracking rather than the + * server bug it is. Mirrors the api-handler / worker forwarders: dynamic + * import (no cycle, no startup cost), redacted message + stack, and it + * NEVER throws so a sink failure cannot mask the original error. + */ +async function reportCoachStreamDefect(err: unknown): Promise { + const e = + err instanceof Error ? err : new Error("Unknown coach stream error"); + const message = redactSecrets(`[insights.coach.stream] ${e.message}`); + console.error("[coach-stream]", message, e); + try { + const [{ getGlitchtipSettings }, { sendGlitchtipEvent }] = + await Promise.all([ + import("@/lib/monitoring-settings"), + import("@/lib/monitoring/glitchtip"), + ]); + const settings = await getGlitchtipSettings(); + if (!settings.glitchtipEnabled || !settings.glitchtipDsn) return; + await sendGlitchtipEvent({ + dsn: settings.glitchtipDsn, + input: { + environment: settings.glitchtipEnvironment || "production", + message, + level: "error", + type: e.name || "Error", + stack: redactOptional(e.stack), + sourceTag: "healthlog-api-handler", + }, + }); + } catch { + /* the reporter must never throw */ + } +} + +/** + * The in-stream producer threw. The stream is already open, so there is no + * 500 to return: answer with an error frame (unless nobody is listening) and + * keep a genuine defect visible. + */ +export function handleProducerFailure( + err: unknown, + args: { signal: AbortSignal; emitter: TurnEmitter; conversationId: string }, +): void { + // #781 — a client disconnect can also surface as an abort-shaped error + // on a path the model step's own catch does not own (e.g. a teardown + // rejection out of persistence). That is not a defect and nobody is + // listening: skip the GlitchTip forward and the error frame. + if (args.signal.aborted) { + annotate({ + action: { name: "insights.coach.cancelled" }, + meta: { conversationId: args.conversationId, lateAbort: true }, + }); + return; + } + // 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); + if (!providerError) { + // Not a tagged provider failure — a genuine server defect raised + // inside the open producer (e.g. a DB error in the post-stream + // `appendMessage` persistence). Record it on the wide event and + // forward it to GlitchTip so it stays visible instead of reading + // as a provider outage; the stream is open, so we still emit an + // error frame below rather than a 500. + getEvent()?.setError(err instanceof Error ? err : new Error(String(err))); + void reportCoachStreamDefect(err); + } + annotate({ + action: { name: "insights.coach.streamError" }, + meta: { + unwrapped: Boolean(providerError), + reported: !providerError, + firstStatus: providerError?.httpStatus ?? null, + }, + }); + if (!args.emitter.aborted()) { + const code = providerError + ? `coach.provider.${providerError.code}` + : "coach.provider.unavailable"; + args.emitter.emit({ type: "error", code, message: code }); + } +} diff --git a/src/lib/ai/coach/turn/model.ts b/src/lib/ai/coach/turn/model.ts new file mode 100644 index 000000000..a6d6ed35d --- /dev/null +++ b/src/lib/ai/coach/turn/model.ts @@ -0,0 +1,322 @@ +/** + * Run the model for one Coach turn: the tool-retrieval loop when every + * provider in the chain supports tools, otherwise the streaming no-tools + * completion over the full snapshot. A provider failure is classified into + * the structured `coach.*` code the stream answers with. + */ +import { annotate } from "@/lib/logging/context"; +import type { Locale } from "@/lib/i18n/config"; +import { localeLanguageNames as LANGUAGE_NAMES } from "@/lib/i18n/config"; +import { + AllProvidersFailedError, + runStreamingRawCompletionWithFallback, +} from "@/lib/ai/provider-runner"; +import type { ProviderChainType } from "@/lib/ai/provider-chain"; +import { singleUserTurn, type CompletionResult } from "@/lib/ai/types"; +import { PROMPT_VERSION } from "@/lib/ai/prompts/insight-generator"; +import { AI_BUDGETS } from "@/lib/ai/ai-budgets"; +import { appendMessage } from "@/lib/ai/coach/persistence"; +import { buildCoachToolRequest } from "@/lib/ai/coach/chat-request-builder"; +import { + COACH_TOOL_DEFS, + buildCoachDataInventory, + renderDataInventory, + renderFocusHint, + buildToolModeAddendum, + runCoachToolLoop, + type CoachToolTrace, +} from "@/lib/ai/coach/tools"; + +import { refundReservation, type TurnReservation } from "./budget"; +import type { TurnChain } from "./chain"; +import type { TurnContext } from "./context"; +import { classifyBubblingProviderError } from "./errors"; +import type { TurnEmitter } from "./types"; + +export type ModelOutcome = + | { + ok: true; + result: CompletionResult; + /** + * v1.38.19 — typed as the chain's provider union, not a bare string: + * the budget reconcile attributes the turn's tokens to the cost owner of + * the hop named here. + */ + workingProviderType: ProviderChainType; + 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. + */ + 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. + */ + noToolsSnapshotPayloads: unknown[]; + /** + * The DATA INVENTORY manifest (per-domain sample counts) that rode this + * turn's tool-mode prompt. The all-missed activation needs it as a + * WIDENER: the counts were in front of the model, so a plain count + * restatement ("you've logged 42 BP readings") must stay grounded even + * on a turn whose every tool missed. Never an ACTIVATOR on its own — + * see the v1.32.1 note below. + */ + inventoryPayloads: unknown[]; + totalTokens: 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). + */ + cachedTokens: number; + } + | { ok: false; code: string }; + +export async function runTurnModel(args: { + userId: string; + locale: Locale; + signal: AbortSignal; + conversationId: string; + ctx: TurnContext; + chain: TurnChain; + toolMode: boolean; + reservation: TurnReservation; + /** + * The turn's frame channel. Not written to yet: the model step is where + * live progress frames will originate. + */ + emitter: TurnEmitter; +}): Promise { + const { userId, locale, signal, conversationId, ctx, chain, toolMode } = args; + const { effectiveScope, workoutEvidence, turnContext, snapshot } = ctx; + 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 toolRequest = buildCoachToolRequest({ + systemPrompt: ctx.systemPrompt, + toolModeAddendum: buildToolModeAddendum(locale), + focusHint: renderFocusHint(effectiveScope?.sources), + workoutEvidence, + dataInventory: renderDataInventory(inventory), + guidedBlock: turnContext.guidedBlock, + transcript: turnContext.transcript, + languageName: LANGUAGE_NAMES[locale], + }); + const loop = await runCoachToolLoop({ + userId, + providers: chain, + system: toolRequest.system, + messages: toolRequest.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, + // v1.22 (#89) — per-user response timeout for each tool-round call. + timeoutMs: ctx.aiResponseTimeoutMs, + }); + // v1.32.1 — the numeric verifier ACTIVATES only when this turn actually + // delivered figures the model was told to ground against: a pinned + // workout-evidence block or a present tool result. The DATA INVENTORY + // manifest (sample counts per domain) is NOT an activator — it rides + // every tool-mode prompt even on a turn where the model answered + // without calling a tool, and on that turn the base prompt deliberately + // carries no pre-computed figures (the model must fetch them), so the + // verifier must stay dormant and leave the prompt-level grounding rule + // as the backstop, exactly as on `main`. Activating it off the + // counts-only inventory would flag a snapshot figure the model cited + // without a fresh tool call as ungrounded (a real regression caught by + // the integration suite). When the turn IS active, the inventory counts + // still WIDEN the authoritative set so a plain count restatement + // ("you've logged 42 BP readings") stays grounded. + const presentToolPayloads = [ + ...(workoutEvidence === null ? [] : [workoutEvidence]), + // A miss carries no `data` but may carry `available` — the bounded + // out-of-window aggregate rule 3 lets the model cite. Ground it. + ...(loop.toolResults ?? []).map((r) => r.data ?? r.available), + ]; + return { + ok: true, + result: loop.result, + // The loop reports the hop it landed on as a bare string; it is + // assigned from `workingProvider.providerType` one frame up. + workingProviderType: loop.workingProviderType as ProviderChainType, + toolTrace: loop.toolTrace, + toolResultPayloads: + presentToolPayloads.length > 0 + ? [...presentToolPayloads, inventory.entries] + : [], + noToolsSnapshotPayloads: [], + inventoryPayloads: [inventory.entries], + totalTokens: loop.totalTokens, + cachedTokens: loop.cachedTokens, + }; + } + // 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 still runs on the complete text. + let streamedDeltas = 0; + const fallback = await runStreamingRawCompletionWithFallback({ + surface: "coach", + userId, + providers: chain, + onDelta: () => { + streamedDeltas += 1; + }, + // v1.20.0 — the no-tools path still builds one assembled user turn (the + // transcript-flattening includes the current authoritative snapshot + // on every stateless no-tools request), so it ships as a single user + // message. The stable persona rides `system` and is cache-eligible. + params: singleUserTurn({ + system: ctx.systemPrompt, + user: ctx.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, + // v1.22 (#89) — per-idle-gap timeout for the streaming local call / + // whole-call timeout for the buffered cloud fallback. + timeoutMs: ctx.aiResponseTimeoutMs, + }), + }); + annotate({ + action: { name: "coach.stream.deltas" }, + meta: { deltas: streamedDeltas }, + }); + const result = fallback.result; + return { + ok: true, + result, + workingProviderType: fallback.workingProvider.providerType, + toolTrace: [], + toolResultPayloads: [], + // v1.21.2 (A8) — the no-tools path grades only when it delivered the + // full SNAPSHOT. Stateless provider requests always do; retaining the + // conditional keeps the builder contract explicit and testable. + noToolsSnapshotPayloads: turnContext.includeFullSnapshot + ? [ + snapshot.sections, + ...(workoutEvidence === null ? [] : [workoutEvidence]), + ] + : [], + inventoryPayloads: [], + totalTokens: result.tokensUsed ?? 0, + cachedTokens: result.cachedInputTokens ?? 0, + }; + } catch (err) { + // The provider chain failed outright — no tokens were billed, so refund + // the full reservation before surfacing the error frame. + await refundReservation(userId, args.reservation); + // #781 — the client walked away mid-generation. The request's abort + // signal is threaded into every provider call, so the teardown surfaces + // here as an abort-shaped failure with `request.signal` already flipped. + // Close the turn honestly instead of letting the user message dangle + // unanswered: persist an EMPTY assistant marker tagged + // `providerType: "cancelled"` (the same channel "refusal" and "nudge" + // already use for non-provider rows) so the thread shows the turn as + // interrupted and offers a retry on reload. The content is empty by + // construction — nothing guarded was produced, and anything the client + // ever SAW was persisted before the first token frame left, so no + // unguarded partial text is ever written. The reservation was refunded + // above; a retry pays exactly what a fresh turn pays. + if (signal.aborted) { + await appendMessage({ + conversationId, + role: "assistant", + content: "", + providerType: "cancelled", + promptVersion: PROMPT_VERSION, + }).catch(() => { + // Marker persistence is best-effort — a failure leaves the + // dangling user turn, which is exactly the pre-#781 state. + }); + annotate({ + action: { name: "insights.coach.cancelled" }, + meta: { conversationId }, + }); + return { ok: false, code: "coach.cancelled" }; + } + 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; + } +} diff --git a/src/lib/ai/coach/turn/persist.ts b/src/lib/ai/coach/turn/persist.ts new file mode 100644 index 000000000..516fcde44 --- /dev/null +++ b/src/lib/ai/coach/turn/persist.ts @@ -0,0 +1,76 @@ +/** + * Persist a Coach reply and record the turn on the wide event. + */ +import { annotate } from "@/lib/logging/context"; +import type { ProviderChainType } from "@/lib/ai/provider-chain"; +import { PROMPT_VERSION } from "@/lib/ai/prompts/insight-generator"; +import type { CoachProvenance } from "@/lib/ai/coach/types"; +import { appendMessage } from "@/lib/ai/coach/persistence"; + +import type { TurnContext } from "./context"; +import type { ModelOutcome } from "./model"; + +export async function persistAssistantReply(args: { + conversationId: string; + replyText: string; + provenance: CoachProvenance; + model: Extract; + ctx: TurnContext; + toolMode: boolean; +}): Promise<{ messageId: string }> { + const { conversationId, replyText, provenance, model, ctx, toolMode } = args; + const workingProviderType: ProviderChainType = model.workingProviderType; + const totalTokensSpent = model.totalTokens; + + // Persist the assistant message BEFORE we begin streaming; if the + // client disconnects we still have the canonical row. + const assistantMessage = await appendMessage({ + conversationId, + role: "assistant", + content: replyText, + metricSource: provenance, + 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; 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: model.result.model ?? null, + }); + + // v1.18.7 — the day's spend was already reconciled against the + // reservation immediately after the provider returned, 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, + historyTurns: ctx.turnContext.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: model.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 && ctx.turnContext.includeFullSnapshot, + promptChars: ctx.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: ctx.snapshot.provenance.metrics.length, + }, + }); + + return { messageId: assistantMessage.id }; +} diff --git a/src/lib/ai/coach/turn/pipeline.ts b/src/lib/ai/coach/turn/pipeline.ts new file mode 100644 index 000000000..7b109ac53 --- /dev/null +++ b/src/lib/ai/coach/turn/pipeline.ts @@ -0,0 +1,176 @@ +/** + * One Coach chat turn, after the route has authenticated, validated, + * rate-limited and screened the inbound message. + * + * Before the stream opens: resolve the conversation and record the user's + * turn, assemble the context, resolve the chain (consent-gated), re-check the + * capability at the egress site, and reserve the budget. Any of these can + * answer on its own (a refusal, a 404, an error frame). + * + * Inside the stream: run the model, settle the budget, guard the reply, + * surface the cards, build the provenance, persist, and only then emit the + * client-visible frames. + */ +import { createSseStream } from "@/lib/sse/create-stream"; + +import { settleReservation, reserveTurnBudget } from "./budget"; +import { surfaceCards } from "./cards"; +import { resolveTurnChain } from "./chain"; +import { assembleTurnContext } from "./context"; +import { persistUserTurn, resolveTurnConversation } from "./conversation"; +import { handleProducerFailure } from "./errors"; +import { runTurnModel } from "./model"; +import { persistAssistantReply } from "./persist"; +import { buildTurnProvenance } from "./provenance"; +import { guardReply } from "./reply-guards"; +import { + SSE_HEADERS, + createTurnEmitter, + emitReply, + startHeartbeat, +} from "./sse"; +import type { ReplyOutcome, TurnEmitter, TurnInput } from "./types"; + +export async function runCoachTurn(input: TurnInput): Promise { + const { userId, locale, message } = input; + + // ── Conversation resolution ────────────────────────────────── + const resolved = await resolveTurnConversation({ + userId, + conversationId: input.conversationId, + message, + locale, + }); + if ("refusal" in resolved) return resolved.refusal; + const conversation = resolved.conversation; + const workingConversationId = conversation.conversationId; + + await persistUserTurn(workingConversationId, message); + + const ctx = await assembleTurnContext({ + userId, + locale, + message, + scope: input.scope, + guidedQuestion: input.guidedQuestion, + workoutId: input.workoutId, + conversation, + }); + + // ── Provider chain ────────────────────────────────────────── + const chainResult = await resolveTurnChain(userId); + if (!chainResult.ok) return chainResult.response; + const { chain, toolMode } = chainResult; + + // The last check before the budget reservation and the provider call: the + // `coach` capability again, at the egress site, so the call site names its + // capability whatever happens to the gate at the top. Inside one request + // the capability inputs are memoised, so this answers from the same + // resolution as the top gate; a fresh read at the wire belongs to the + // provider chokepoints, not to this route. + const refusedAtEgress = await input.recheckCapability(); + if (refusedAtEgress) return refusedAtEgress; + + const budget = await reserveTurnBudget({ userId, chain, toolMode }); + if (!budget.ok) return budget.response; + const { reservation } = budget; + + // v1.22 (#89) — the provider call + every safety guard + persistence run + // INSIDE the SSE stream. 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. + async function produceReply(emitter: TurnEmitter): Promise { + const model = await runTurnModel({ + userId, + locale, + signal: input.signal, + conversationId: workingConversationId, + ctx, + chain, + toolMode, + reservation, + emitter, + }); + if (!model.ok) return model; + + await settleReservation(userId, reservation, { + totalTokens: model.totalTokens, + cachedTokens: model.cachedTokens, + servedBy: model.workingProviderType, + }); + + const guarded = await guardReply({ + userId, + locale, + conversation, + ctx, + toolMode, + model, + }); + if (!guarded.ok) return guarded; + const reply = guarded.reply; + + const cards = await surfaceCards({ + userId, + coachPrefs: ctx.coachPrefs, + reply, + }); + const provenance = buildTurnProvenance({ + snapshotProvenance: ctx.snapshot.provenance, + reply, + suggestion: cards.suggestion, + action: cards.action, + toolTrace: model.toolTrace, + }); + const { messageId } = await persistAssistantReply({ + conversationId: workingConversationId, + replyText: reply.replyText, + provenance, + model, + ctx, + toolMode, + }); + + return { + ok: true, + replyText: reply.replyText, + provenance, + suggestion: cards.suggestion, + action: cards.action, + messageId, + totalTokens: model.totalTokens, + model: model.result.model ?? null, + }; + } + + // ── 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 visible cadence reads like a live chat reply. + const stream = createSseStream(async (controller) => { + const emitter = createTurnEmitter(controller); + const stopHeartbeat = startHeartbeat(controller); + + let outcome: ReplyOutcome; + try { + outcome = await produceReply(emitter); + } catch (err) { + stopHeartbeat(); + handleProducerFailure(err, { + signal: input.signal, + emitter, + conversationId: workingConversationId, + }); + return; + } + stopHeartbeat(); + + await emitReply(emitter, outcome, workingConversationId); + }); + + return new Response(stream, { status: 200, headers: SSE_HEADERS }); +} diff --git a/src/lib/ai/coach/turn/provenance.ts b/src/lib/ai/coach/turn/provenance.ts new file mode 100644 index 000000000..6000fb403 --- /dev/null +++ b/src/lib/ai/coach/turn/provenance.ts @@ -0,0 +1,88 @@ +/** + * The provenance envelope a Coach reply is persisted and streamed with: + * what the snapshot covered, plus what this turn added (key values, cards, + * the tool trace, the grounded figures, the withheld-figure count). + */ +import { annotate } from "@/lib/logging/context"; +import { PROMPT_VERSION } from "@/lib/ai/prompts/insight-generator"; +import type { CoachProvenance, CoachSuggestion } from "@/lib/ai/coach/types"; +import type { CoachSuggestedAction } from "@/lib/ai/coach/suggest-action"; +import type { CoachToolTrace } from "@/lib/ai/coach/tools"; + +import type { GuardedReply } from "./reply-guards"; + +export function buildTurnProvenance(args: { + snapshotProvenance: CoachProvenance; + reply: GuardedReply; + suggestion: CoachSuggestion | null; + action: CoachSuggestedAction | null; + toolTrace: CoachToolTrace[]; +}): CoachProvenance { + const { snapshotProvenance, reply, toolTrace } = args; + const surfacedSuggestion = args.suggestion; + const surfacedAction = args.action; + const sentinel = reply.keyValuesSentinel; + const outboundBlocked = reply.outboundBlocked; + const { groundedFigures, unverifiedStripped } = reply; + + const enrichedProvenance: CoachProvenance = { + ...snapshotProvenance, + ...(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. + ...(!outboundBlocked && 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, + })), + } + : {}), + // v1.32.9 (Coach Guard II / G2) — persist THIS turn's tool figures so a + // later turn's Grounding Ledger can recall them (D3-safe: tool trace, not + // prose). Dropped on a blocked turn (its figures rode the discarded reply). + ...(!outboundBlocked && groundedFigures.length > 0 + ? { groundedFigures } + : {}), + // v1.32.14 — the count of figures the grounding guard withheld this turn, + // so the quiet "some figures couldn't be checked" notice renders under the + // bubble and survives a conversation reload. Count only, no values. Omitted + // when nothing was withheld. + ...(unverifiedStripped > 0 + ? { unverifiedFigures: unverifiedStripped } + : {}), + }; + 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, + }, + }); + } + return enrichedProvenance; +} diff --git a/src/lib/ai/coach/turn/reply-guards.ts b/src/lib/ai/coach/turn/reply-guards.ts new file mode 100644 index 000000000..00be74862 --- /dev/null +++ b/src/lib/ai/coach/turn/reply-guards.ts @@ -0,0 +1,338 @@ +/** + * Every guard a Coach reply passes before a byte of it reaches the client: + * the sentinel strips (KEYVALUES, SUGGEST-REMINDER, REMEMBER, SUGGEST-ACTION), + * the outbound safety screen, the grounding ledger, the empty-record + * replacement, and the learn-link scrub. + */ +import { annotate } from "@/lib/logging/context"; +import { auditLog } from "@/lib/auth/audit"; +import type { Locale } from "@/lib/i18n/config"; +import { getServerTranslator } from "@/lib/i18n/server-translator"; +import { PROMPT_VERSION } from "@/lib/ai/prompts/insight-generator"; +import { + screenCoachReply, + coachOutboundFallback, +} from "@/lib/ai/coach/outbound-guard"; +import { parseKeyValuesSentinel } from "@/lib/ai/coach/keyvalues"; +import { + findUnverifiedCoachNumbersInLedger, + stripUnverifiedNumbers, +} from "@/lib/ai/coach/coach-prose-grounding"; +import { + buildGroundingLedger, + figuresForPersistence, +} from "@/lib/ai/coach/grounding-ledger"; +import { scrubUnknownLearnLinks } from "@/lib/ai/coach/learn-link-guard"; +import { parseSuggestReminder } from "@/lib/ai/coach/suggest-reminder"; +import { + parseRememberSentinel, + captureReminderFromSentinel, +} from "@/lib/ai/coach/reminders"; +import { parseSuggestAction } from "@/lib/ai/coach/suggest-action"; + +import type { TurnContext } from "./context"; +import type { ModelOutcome } from "./model"; +import type { TurnConversation } from "./types"; + +export interface GuardedReply { + replyText: string; + keyValuesSentinel: ReturnType; + suggestParse: ReturnType; + actionParse: ReturnType; + /** True when the outbound screen replaced the reply with its fallback. */ + outboundBlocked: boolean; + groundedFigures: number[]; + unverifiedStripped: number; +} + +export async function guardReply(args: { + userId: string; + locale: Locale; + conversation: TurnConversation; + ctx: TurnContext; + toolMode: boolean; + model: Extract; +}): Promise<{ ok: true; reply: GuardedReply } | { ok: false; code: string }> { + const { userId, locale, conversation, ctx, toolMode, model } = args; + const workingConversationId = conversation.conversationId; + const { toolTrace, toolResultPayloads } = model; + + const rawReply = (model.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.keyvalues.parse_failed" }, + meta: { + kept: sentinel.keyValues.length, + reason: "empty_prose_after_strip", + promptVersion: PROMPT_VERSION, + }, + }); + 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. 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 turn. + // + // v1.30.25 — the capture writes `proposed`, not `active`. The block is + // model output, and the model reads a prompt carrying document-sourced + // text, so the write needs the same propose-then-confirm moat every other + // model-driven write already has. See `captureReminderFromSentinel`. + 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, locale, ctx.scheduleDoses); + 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 the editorial elision mark `[…]` and annotated (v1.32.14 — the withheld + // count also rides the provenance envelope to drive the per-message notice). + // 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. + // v1.32.9 (Coach Guard II / G2+G3) — grade the prose against the typed + // Grounding Ledger rather than a per-turn magnitude bag. Activation is + // unchanged from Guard I: the ledger is graded ONLY when THIS turn delivered + // fresh figures the model was told to ground against (a present tool result + // / pinned workout, or the full snapshot on the no-tools path). The + // cross-turn, memory, schedule, reference, and guided sources only WIDEN an + // active grading — they never ACTIVATE it, so a snapshot figure the model + // cited on a no-tool turn is still left alone (the v1.32.1 regression guard + // holds). Assistant prose is never a ledger source (D3). + // + // The turn that CALLED tools and got nothing back is graded too. It used to + // be the one turn the verifier sat out: a pure miss carries no `data` and no + // `available`, `runCoachToolLoop` drops that shape from `toolResults`, and + // the no-tools snapshot fallback below is populated only in the non-tool + // branch — so `activatingPayloads` came out empty and every figure in the + // reply shipped unchecked. That is backwards. A turn whose every tool + // reported `{ present: false }` is not an absence of evidence about the + // reply, it is evidence that the record holds nothing to cite, which is + // exactly when a fabricated figure is both likeliest and most harmful. + // + // `toolTrace` is the discriminant, because it records every tool that ran + // INCLUDING the pure misses. It separates the two cases the old condition + // conflated: tools ran and found nothing (grade — the model was told the + // record is empty and answered with numbers anyway), versus the model + // answered without calling a tool at all (stay dormant — the v1.32.1 + // regression, where the base prompt deliberately carries no pre-computed + // figures and flagging a legitimately recalled one was a real defect). + const missedEveryTool = + toolMode && toolTrace.length > 0 && toolResultPayloads.length === 0; + const activatingPayloads = + toolResultPayloads.length > 0 + ? toolResultPayloads + : missedEveryTool + ? model.inventoryPayloads + : model.noToolsSnapshotPayloads; + let groundedFigures: number[] = []; + // v1.32.14 — count of figures withheld from THIS reply (each rewritten to the + // elision mark). Hoisted out of the guard block so it can ride the provenance + // envelope and drive the quiet per-message notice. Stays 0 on a blocked turn + // (the guard is skipped, its fallback prose carries no figures). + let unverifiedStripped = 0; + // True when the guard stripped a figure on a turn whose every tool missed — + // i.e. the model put a number on a record that holds none. Drives the + // honest-replacement below. + let fabricatedOnEmptyRecord = false; + if (activatingPayloads.length > 0 || missedEveryTool) { + const ledger = buildGroundingLedger({ + toolPayloads: activatingPayloads, + priorToolFigures: conversation.priorToolFigures, + priorUserMessages: conversation.priorUserMessages, + // The self-context (goals / about-me) rides the system prompt on both + // paths — the ledger's `memory` source. + memoryTexts: ctx.aboutMe ? [ctx.aboutMe] : [], + scheduleDoses: ctx.scheduleDoses, + // Reference grounding rides the prompt only on the no-tools + // full-snapshot path; the guided block rides both paths. + referenceGrounding: + !toolMode && ctx.turnContext.includeFullSnapshot + ? ctx.snapshot.referenceGrounding + : null, + guidedBlock: ctx.turnContext.guidedBlock, + }); + // Persist THIS turn's tool figures (bare magnitudes) so a LATER turn can + // recall them via the ledger — D3: the tool trace, never the prose. + groundedFigures = figuresForPersistence(ledger); + if (!outbound.block) { + const unverified = findUnverifiedCoachNumbersInLedger( + replyText, + ledger, + locale, + // On an all-missed turn an EMPTY ledger is the finding, not a reason + // to skip: nothing was retrievable, so nothing can reconcile. + { gradeAgainstEmptyLedger: missedEveryTool }, + ); + if (unverified.length > 0) { + const { prose: corrected, stripped } = stripUnverifiedNumbers( + replyText, + unverified, + ); + replyText = corrected; + unverifiedStripped = stripped; + fabricatedOnEmptyRecord = missedEveryTool && stripped > 0; + 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, + }, + }); + } + } + } + + // A stripped reply on an empty record is not worth sending. Eliding the + // digits leaves the sentences that carried them — "your sleep averaged […] + // minutes, up from […]" still asserts a series, an average and a direction + // for a metric the tools just reported has no readings at all. The false + // claim survives the strip; only its precision goes. So replace the turn + // with the honest answer instead of shipping the elided mush. + // + // Scoped tightly: this fires only when every tool missed AND the guard + // actually stripped something. A reply that cited a legitimately grounded + // figure (a prior turn's tool result, an inventory count) reconciles + // against the ledger, strips nothing, and is left alone. + // + // REPLACE, not withhold — the same policy the outbound screen uses on this + // surface, for the same reason: the user is waiting on a synchronous answer + // and silence reads as a failure. + // + // Accepted trade-off: a reply mixing one grounded recall with one + // fabrication is replaced wholesale, losing the good half. That is the + // right way round — the alternative leaves an invented figure's sentence + // on screen. + if (fabricatedOnEmptyRecord) { + replyText = getServerTranslator(locale).t("coach.noRecordedData"); + // The reply now carries no figures at all, so the withheld-figure notice + // would be describing prose the reader can no longer see. The replacement + // copy states the same thing in plain words. + unverifiedStripped = 0; + annotate({ + action: { name: "coach.prose.empty_record_replaced" }, + meta: { promptVersion: PROMPT_VERSION }, + }); + await auditLog("insights.coach.empty_record_replaced", { + userId, + details: { conversationId: workingConversationId }, + }); + } + + // 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, + }, + }); + } + } + + return { + ok: true, + reply: { + replyText, + keyValuesSentinel: sentinel, + suggestParse, + actionParse, + outboundBlocked: outbound.block, + groundedFigures, + unverifiedStripped, + }, + }; +} diff --git a/src/lib/ai/coach/turn/sse.ts b/src/lib/ai/coach/turn/sse.ts new file mode 100644 index 000000000..9eb296148 --- /dev/null +++ b/src/lib/ai/coach/turn/sse.ts @@ -0,0 +1,218 @@ +/** + * The Coach chat wire: SSE framing, the keepalive heartbeat, the token + * cadence, and the three stream shapes a turn can answer with (a guarded + * reply, a refusal, a structured error). + * + * Imports nothing from the tool registry or the snapshot builder, so a + * pipeline that must never reach either can still share this framing. + */ +import { HttpError } from "@/lib/api-handler"; +import { prisma } from "@/lib/db"; +import { PROMPT_VERSION } from "@/lib/ai/prompts/insight-generator"; +import type { AiUnavailableReason } from "@/lib/ai/capabilities/types"; +import type { CoachStreamEvent } from "@/lib/ai/coach/types"; +import { appendMessage, createConversation } from "@/lib/ai/coach/persistence"; +import { createSseStream, type SseController } from "@/lib/sse/create-stream"; + +import type { ReplyOutcome, TurnEmitter } from "./types"; + +export const SSE_HEADERS: Record = { + "Content-Type": "text/event-stream; charset=utf-8", + "Cache-Control": "no-cache, no-transform", + Connection: "keep-alive", + "X-Accel-Buffering": "no", +}; + +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"); + +/** + * Keepalive: an SSE comment frame (`: ka`) the client parser ignores, + * flushed on an interval through the pre-first-token (prompt-processing) + * and generation phases. Returns the stop function. + */ +export function startHeartbeat(controller: SseController): () => void { + const heartbeat = setInterval(() => { + controller.enqueue(HEARTBEAT_FRAME); + }, HEARTBEAT_MS); + return () => clearInterval(heartbeat); +} + +/** The frame channel of one turn, over one stream controller. */ +export function createTurnEmitter(controller: SseController): TurnEmitter { + return { + emit: (frame) => controller.enqueue(encodeFrame(frame)), + aborted: () => controller.signal.aborted, + }; +} + +/** + * Split a full assistant reply into ~roughly-word-sized chunks so the + * UI gets a "streaming" feel even when the underlying provider client + * returned the body in one shot. + */ +function tokeniseForStreaming(content: string): string[] { + if (!content) return []; + // Split on whitespace boundaries while preserving the spaces — keeps + // word boundaries intact and avoids the UI having to glue tokens. + const matches = content.match(/\S+\s*/g); + return matches ?? [content]; +} + +/** + * v1.12.0 — yield control back to the event loop for one tick so the + * stream controller flushes the just-enqueued frame before the next one + * is produced. `setTimeout(0)` (rather than a bare `Promise.resolve()` + * microtask) hands the turn back to the platform's stream pump so each + * SSE frame lands in its own network chunk; a microtask would drain + * before the runtime gets a chance to flush. The delay is intentionally + * zero — we want incremental delivery, not an artificial typewriter + * pause. + */ +function flushTick(): Promise { + return new Promise((resolve) => setTimeout(resolve, 0)); +} + +/** + * Emit a finished turn. A failed outcome is one `error` frame. A guarded + * reply goes out as `token* → provenance → suggestion? → suggestedAction? → + * done`; the token frames carry the FULLY-GUARDED text, and every guard ran + * before the first one leaves. + */ +export async function emitReply( + emitter: TurnEmitter, + outcome: ReplyOutcome, + conversationId: string, +): Promise { + if (!outcome.ok) { + if (!emitter.aborted()) { + emitter.emit({ + 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 (emitter.aborted()) return; + emitter.emit({ type: "token", token: tok }); + await flushTick(); + } + if (emitter.aborted()) return; + emitter.emit({ type: "provenance", metricSource: outcome.provenance }); + // v1.18.1 (Workstream C) — additive `suggestion` frame. + if (outcome.suggestion) { + emitter.emit({ type: "suggestion", suggestion: outcome.suggestion }); + } + // v1.22 (F6) — additive `suggestedAction` frame. + if (outcome.action) { + emitter.emit({ + type: "suggestedAction", + suggestedAction: outcome.action, + }); + } + // v1.18.9 — additive `usage` envelope on the `done` frame. + emitter.emit({ + type: "done", + conversationId, + messageId: outcome.messageId, + usage: { + totalTokens: outcome.totalTokens || null, + model: outcome.model, + }, + }); +} + +/** + * Emit a refusal as a single `token` frame followed by `done`. No + * provider call, no persisted assistant message — the user message is + * still kept on disk so the rail shows the conversation history + * accurately. The user message landing on disk is a deliberate choice; + * the rail otherwise wouldn't show the user's attempt at all. + */ +export async function streamRefusal(args: { + userId: string; + conversationId: string | undefined; + message: string; + refusalText: string; +}): Promise { + let conversationId = args.conversationId; + if (!conversationId) { + const created = await createConversation({ + userId: args.userId, + title: args.message, + }); + conversationId = created.id; + } else { + const owned = await prisma.coachConversation.findFirst({ + where: { id: conversationId, userId: args.userId }, + select: { id: true }, + }); + if (!owned) { + throw new HttpError(404, "coach.conversation.notFound"); + } + } + + await appendMessage({ + conversationId, + role: "user", + content: args.message, + }); + const refusalMessage = await appendMessage({ + conversationId, + role: "assistant", + content: args.refusalText, + metricSource: { windows: [], metrics: ["general"] }, + providerType: "refusal", + promptVersion: PROMPT_VERSION, + }); + + const stream = createSseStream((controller) => { + controller.enqueue(encodeFrame({ type: "token", token: args.refusalText })); + controller.enqueue( + encodeFrame({ + type: "done", + conversationId, + messageId: refusalMessage.id, + }), + ); + }); + + return new Response(stream, { status: 200, headers: SSE_HEADERS }); +} + +export function streamProviderError(args: { + code: string; + /** The capability reason, when the refusal came from one. */ + reason?: AiUnavailableReason; +}): Response { + const stream = createSseStream((controller) => { + controller.enqueue( + encodeFrame({ + type: "error", + code: args.code, + message: args.code, + ...(args.reason ? { reason: args.reason } : {}), + }), + ); + }); + // Status 200 so the streaming client reads the SSE body and parses + // the structured `error` frame (HTTP-status branches drop the + // structured code on the floor). + return new Response(stream, { status: 200, headers: SSE_HEADERS }); +} diff --git a/src/lib/ai/coach/turn/types.ts b/src/lib/ai/coach/turn/types.ts new file mode 100644 index 000000000..c8a58b9dd --- /dev/null +++ b/src/lib/ai/coach/turn/types.ts @@ -0,0 +1,87 @@ +/** + * Shared shapes of one Coach chat turn (`POST /api/insights/chat`). + * + * The turn runs as a pipeline: resolve the conversation, assemble the + * context, resolve the provider chain, reserve the budget, then — inside the + * SSE stream — run the model, guard the reply, surface the cards, build the + * provenance, persist, and emit. Each step lives in its own module under + * `src/lib/ai/coach/turn/`; these are the values that pass between them. + * + * Deliberately free of the tool registry and the snapshot builder, so the + * wire modules (`sse.ts`, `errors.ts`) can import from here without pulling + * either into their import graph. + */ +import type { Locale } from "@/lib/i18n/config"; +import type { CoachTurn } from "@/lib/ai/coach/chat-request-builder"; +import type { CoachSuggestedAction } from "@/lib/ai/coach/suggest-action"; +import type { + CoachProvenance, + CoachScope, + CoachStreamEvent, + CoachSuggestion, +} from "@/lib/ai/coach/types"; + +/** The validated request, narrowed to the caller, ready for the pipeline. */ +export interface TurnInput { + userId: string; + locale: Locale; + /** The request's abort signal, threaded into every provider call. */ + signal: AbortSignal; + conversationId: string | undefined; + message: string; + scope: CoachScope | undefined; + guidedQuestion: string | undefined; + workoutId: string | undefined; + /** + * The `coach` capability gate, re-run at the egress site. Owned by the + * route (the capability inventory reads it there); returns the refusal + * response, or null when the Coach is still available. + */ + recheckCapability: () => Promise; +} + +/** The conversation a turn continues (or the one it just created). */ +export interface TurnConversation { + conversationId: string; + priorTurns: CoachTurn[]; + /** + * v1.32.9 (Coach Guard II / G2) — cross-turn Grounding Ledger sources. Prior + * USER messages (numbers the user themselves stated) and the persisted + * per-turn tool figures of prior turns (`groundedFigures` — server-computed + * magnitudes, NEVER the assistant's prose, per D3). Both feed reconciliation + * so a figure recalled a turn later is not stripped, without a laundering + * path through assistant narration. + */ + priorUserMessages: string[]; + priorToolFigures: number[]; + /** + * v1.11.1 — rolling summary of the elided older turns, read stale-while- + * revalidate; null for a fresh conversation or when none is on file. + */ + priorSummary: string | null; +} + +/** + * The client-visible frame channel of one turn. Created inside + * `createSseStream` and threaded to the model step, so a step can put a frame + * on the wire while the turn is still running. + */ +export interface TurnEmitter { + emit(frame: CoachStreamEvent): void; + /** True once the consumer cancelled the stream (client disconnect). */ + aborted(): boolean; +} + +/** What the in-stream producer hands back for client-visible emission. */ +export type ReplyOutcome = + | { + ok: true; + replyText: string; + provenance: CoachProvenance; + suggestion: CoachSuggestion | null; + action: CoachSuggestedAction | null; + messageId: string; + totalTokens: number; + model: string | null; + } + | { ok: false; code: string }; diff --git a/src/lib/documents/__tests__/fenced-chat-module-graph.test.ts b/src/lib/documents/__tests__/fenced-chat-module-graph.test.ts index fe2946dce..a9e4da651 100644 --- a/src/lib/documents/__tests__/fenced-chat-module-graph.test.ts +++ b/src/lib/documents/__tests__/fenced-chat-module-graph.test.ts @@ -69,7 +69,7 @@ const TOOL_ROUTE = resolve(SRC, "app/api/insights/chat/route.ts"); // Modules that constitute the tool loop / health snapshot — forbidden in the // fenced graph. const TOOL_MODULES = [ - resolve(SRC, "lib/ai/coach/tools.ts"), + resolve(SRC, "lib/ai/coach/tools/index.ts"), resolve(SRC, "lib/ai/coach/snapshot.ts"), ]; @@ -101,6 +101,39 @@ describe("fenced-chat module graph — the fence is structural", () => { } }); + it("the Coach turn's wire modules reach neither the tool registry nor the snapshot builder", () => { + // `turn/sse.ts` (framing, refusal and error streams) and `turn/errors.ts` + // (provider-error classification) are the parts of the Coach turn a fenced + // pipeline could share. They stay usable there only while they are + // tool-free. + const reachable = reachableFrom([ + resolve(SRC, "lib/ai/coach/turn/sse.ts"), + resolve(SRC, "lib/ai/coach/turn/errors.ts"), + ]); + for (const forbidden of TOOL_MODULES) { + expect( + reachable.has(forbidden), + `Coach turn wire modules must NOT import ${forbidden}`, + ).toBe(false); + } + // Sanity: the walk actually left the entry files. + expect(reachable.has(resolve(SRC, "lib/ai/coach/turn/types.ts"))).toBe( + true, + ); + }); + + it("sanity: the tool route DOES reach the tool registry through the turn pipeline", () => { + const reachable = reachableFrom([TOOL_ROUTE]); + // Without this the negative checks above would pass against a path that + // no longer exists — which is how `tools.ts` (now `tools/index.ts`) sat + // here unmatched after the tool registry became a directory. + for (const target of TOOL_MODULES) { + expect(reachable.has(target), `Tool route must reach ${target}`).toBe( + true, + ); + } + }); + it("sanity: the fenced pipeline DOES reach its document loader + prompt builder", () => { const reachable = reachableFrom(FENCED_ENTRIES); expect(reachable.has(resolve(SRC, "lib/documents/content-index.ts"))).toBe(