From 869f1d9c11517487ed4f099939640127095460bf Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sat, 20 Jun 2026 16:58:23 +0200 Subject: [PATCH] fix(coach): wire the outbound screen and an SSE cancel handler Apply the outbound safety screen to the assembled Coach reply before persistence and streaming, with a per-turn audit row on a block. createSseStream gains a cancel handler + a shared AbortSignal: a client disconnect now flips the signal so the token-emit loop stops enqueuing into a closed connection and any provider fetch wired to the signal is torn down. The chat emit loop honours it between frames. --- src/app/api/insights/chat/route.ts | 43 +++++++++++++++++++++++++++--- src/lib/sse/create-stream.ts | 29 +++++++++++++++++++- 2 files changed, 68 insertions(+), 4 deletions(-) diff --git a/src/app/api/insights/chat/route.ts b/src/app/api/insights/chat/route.ts index ea8907ad7..f73812a66 100644 --- a/src/app/api/insights/chat/route.ts +++ b/src/app/api/insights/chat/route.ts @@ -66,6 +66,10 @@ import { reconcileSpend, } from "@/lib/ai/coach/budget"; 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 { getSelfContextTextForUser } from "@/lib/ai/coach/about-me"; import { buildCoachSnapshot } from "@/lib/ai/coach/snapshot"; @@ -564,9 +568,33 @@ Reply now as the assistant, in ${locale === "de" ? "German" : "English"}.`; // + dismissal memory + cooldown + dedup against a live COACH reminder). // A suppressed proposal leaves the prose unchanged and emits no card. const suggestParse = parseSuggestReminder(proseAfterStrip); - const replyText = suggestParse.prose.trim() || proseAfterStrip; + let replyText = suggestParse.prose.trim() || proseAfterStrip; + + // v1.18.10 (HIGH-2) — OUTBOUND safety screen on the assembled assistant + // reply, before persistence and streaming. The inbound `detectRefusal` + // guards the user's message; this guards the model's reply for a + // dose-prescription or a fabricated clinical risk score that slipped past + // the system-prompt GLP-1/grounding contracts. On a trip the turn is + // replaced with a calm, grounded fallback and any reminder suggestion / + // key-value provenance is dropped — the user never sees the unsafe text. + const outbound = screenCoachReply(replyText); + if (outbound.block && outbound.reason) { + replyText = coachOutboundFallback(outbound.reason, locale); + annotate({ + action: { name: "insights.coach.outbound_blocked" }, + meta: { reason: outbound.reason, promptVersion: PROMPT_VERSION }, + }); + await auditLog("insights.coach.outbound_blocked", { + userId, + details: { + conversationId: workingConversationId, + reason: outbound.reason, + }, + }); + } + let surfacedSuggestion: CoachSuggestion | null = null; - if (suggestParse.cadence) { + if (!outbound.block && suggestParse.cadence) { const decision = await gateSuggestion({ prisma, userId, @@ -614,7 +642,11 @@ Reply now as the assistant, in ${locale === "de" ? "German" : "English"}.`; } const enrichedProvenance: typeof snapshot.provenance = { ...snapshot.provenance, - ...(sentinel.keyValues.length > 0 ? { keyValues: sentinel.keyValues } : {}), + // 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 } : {}), }; if (sentinel.malformed) { @@ -692,9 +724,14 @@ Reply now as the assistant, in ${locale === "de" ? "German" : "English"}.`; // pace). const stream = createSseStream(async (controller) => { for (const tok of tokeniseForStreaming(replyText)) { + // v1.18.10 (A-2) — stop tokenising the moment the client disconnects; + // the cancel handler flips this signal, so we don't pace frames into a + // closed connection. + if (controller.signal.aborted) return; controller.enqueue(encodeFrame({ type: "token", token: tok })); await flushTick(); } + if (controller.signal.aborted) return; controller.enqueue( encodeFrame({ type: "provenance", diff --git a/src/lib/sse/create-stream.ts b/src/lib/sse/create-stream.ts index 78181a7c2..15ee3cb13 100644 --- a/src/lib/sse/create-stream.ts +++ b/src/lib/sse/create-stream.ts @@ -20,6 +20,14 @@ export interface SseController { enqueue: (chunk: T) => void; + /** + * v1.18.10 (A-2) — aborts when the consumer cancels the stream (client + * disconnect, navigation away, `Response.body.cancel()`). The `emit` + * callback should check `signal.aborted` between frames and stop producing, + * and pass `signal` to any in-flight provider fetch so a mid-stream + * disconnect tears the upstream call down instead of streaming into the void. + */ + signal: AbortSignal; } /** @@ -27,17 +35,36 @@ export interface SseController { * supplied `emit` function inside a try/finally that always closes the * controller. Use the returned stream with the standard SSE response * headers (`Content-Type: text/event-stream`, `Cache-Control: no-cache`). + * + * v1.18.10 (A-2) — the stream now wires a `cancel` handler: when the consumer + * cancels (client disconnect), the shared `AbortController` fires, so the + * `emit` callback's `signal` flips to aborted and any provider fetch wired to + * it is torn down. Without this the emit loop kept enqueuing frames (and an + * abort-less upstream fetch kept running) after the client had gone. */ export function createSseStream( emit: (controller: SseController) => void | Promise, ): ReadableStream { + const abort = new AbortController(); return new ReadableStream({ async start(controller) { + const sse: SseController = { + enqueue: (chunk) => { + // A late enqueue after the consumer cancelled would throw on the + // closed controller — swallow it so the abort path stays clean. + if (abort.signal.aborted) return; + controller.enqueue(chunk); + }, + signal: abort.signal, + }; try { - await emit(controller); + await emit(sse); } finally { controller.close(); } }, + cancel(reason) { + abort.abort(reason); + }, }); }