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.
This commit is contained in:
Marc-André Bombeck committed 2026-06-20 16:58:23 +02:00
1 parent 64de009d66
commit 869f1d9c11
2 files changed
+68 -4

No files matched your search

+40 -3
View File
@@ -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",
+28 -1
View File
@@ -20,6 +20,14 @@
export interface SseController<T = Uint8Array> {
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<T = Uint8Array> {
* 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<void>,
): ReadableStream<Uint8Array> {
const abort = new AbortController();
return new ReadableStream<Uint8Array>({
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);
},
});
}