diff --git a/docs/branch-review-ledger.md b/docs/branch-review-ledger.md index 49ebbcec0..fe6ba0627 100644 --- a/docs/branch-review-ledger.md +++ b/docs/branch-review-ledger.md @@ -39,3 +39,4 @@ Use this ledger to prevent repeated branch and PR reviews when the reviewed HEAD | 2026-07-11 | claude/mobile-search-bar-fix (PR #456) | b73196c2e2e4a536804cdcdb50879c29e2c582c5 | PR required-testing review | All 4 Advisory UI regression failures confirmed PR-caused via A/B against pre-merge main (01f2cee0d): the 640px mode-home query moved the phone composer out of the hero, contradicting the design tests; residual ≥640px vanish remained when the slot never mounts. PR merged (b32c17b34) before the rework landed; follow-up fix shipped on `claude/mode-home-composer-hero-fix` (0px hero query restored, portal-outcome inline fallback, new `@critical` composer-presence test). Also found: main CI red on every push — missing `RAG_QUERY_HASH_SECRET` secret fails the deployment boot smoke and skips `release-browser-matrix`; owner adding the secret. | Local chromium A/B (PR head 4/5 fail vs baseline product-pass); rework targeted run 6/6 pass incl. new `@critical`; `npm run typecheck`; `npm run lint`; focused Prettier check | | 2026-07-11 | PR #487 / claude/answer-page-design-polish-ffd5a6 | b2c772606126f8323424bc9c0b636bac77c08789 | open-PR review, unresolved comments, and CI | Two findings fixed: expanded weak/unsupported prior answers retain an explicit source-review warning, and cross-mode search actions no longer log an incorrect detail-open telemetry event. Added a persisted prior-turn browser assertion. No additional high-confidence defect was found in the changed scope. | Focused answer-render and cross-mode Vitest (20/20); TypeScript; focused Prettier; `git diff --check`. Browser assertion delegated to hosted CI because Turbopack rejects the isolated worktree's external node_modules junction. | | 2026-07-11 | PR #469 / claude/response-formatting-cleanup-b57a9c | f1864308e0b287bb83b2a13daca4c3aa2ab95a3e | open-PR review, unresolved comments, and CI | P2 fixed: the OCR bullet sanitizer now preserves line-start O blood values when followed by blood or red-cell noun tails while still stripping non-blood bullets such as `o Negative screen`. No additional high-confidence defect was found in the nine-file diff. | Focused sanitizer/extractive Vitest (75/75); TypeScript; focused Prettier; `git diff --check`. Production-readiness script ran fail-closed with provider variables cleared and reported only expected missing provider configuration. | +| 2026-07-11 | PR #466 / claude/search-timeout-failure-s6aiuj | 54d52292eeb9e1c7856b3dad89d1b72e0d49fd53 | open-PR review, unresolved comments, and CI | P2 fixed: SSE progress/token/error emission now tolerates a client cancellation racing an enqueue, so the catch path cannot throw while reporting the original stream error. No additional high-confidence defect was found in the six-file diff. | Focused SSE and search utility Vitest (13/13); TypeScript; focused Prettier. Hosted advisory browser failure was shared stale assertion drift and is rerun after this push. | diff --git a/src/app/api/answer/stream/route.ts b/src/app/api/answer/stream/route.ts index ac64e8b27..cb41834f6 100644 --- a/src/app/api/answer/stream/route.ts +++ b/src/app/api/answer/stream/route.ts @@ -24,6 +24,7 @@ import { logAnswerDiagnostics } from "@/lib/answer-telemetry"; import { isSupabaseApiKeyConfigurationError, nonProductionSupabaseDemoFallbackReason } from "@/lib/supabase/errors"; import { AuthenticationError, unauthorizedResponse } from "@/lib/supabase/auth"; import { logger } from "@/lib/logger"; +import { startSseHeartbeat } from "@/lib/sse-heartbeat"; import { parseJsonBody } from "@/lib/validation/body"; import type { RagAnswer } from "@/lib/types"; @@ -144,8 +145,17 @@ function streamAnswer(body: AnswerBody, ownerId?: string, signal?: AbortSignal, new ReadableStream({ async start(controller) { const send = (event: string, data: unknown) => { - controller.enqueue(encoder.encode(encodeSse(event, data))); + try { + controller.enqueue(encoder.encode(encodeSse(event, data))); + } catch { + // The client may cancel between generation callbacks. Once the + // stream is closed there is no remaining consumer for this frame. + } }; + // Generation can go silent for long stretches (strong-route reasoning + // before the first output token); heartbeat comments keep the + // connection visibly alive for proxies and the client's stall watchdog. + const stopHeartbeat = startSseHeartbeat((frame) => controller.enqueue(encoder.encode(frame))); const onProgress = (event: AnswerProgressEvent) => send("progress", event); // Stream the answer prose as it generates (content-preserving) and signal a reset when a // provisional answer is being revised by the quality gates. @@ -242,7 +252,14 @@ function streamAnswer(body: AnswerBody, ownerId?: string, signal?: AbortSignal, const streamError = streamErrorPayload(error); send("error", { error: streamError.message, status: streamError.status, details: streamError.details }); } finally { - controller.close(); + stopHeartbeat(); + // The client may have already cancelled the stream (Stop button / + // watchdog abort), in which case close() throws on a closed stream. + try { + controller.close(); + } catch { + // Stream already closed or cancelled — nothing left to release. + } } }, }), diff --git a/src/components/ClinicalDashboard.tsx b/src/components/ClinicalDashboard.tsx index 2b377c965..79748509e 100644 --- a/src/components/ClinicalDashboard.tsx +++ b/src/components/ClinicalDashboard.tsx @@ -128,6 +128,7 @@ import { isWeakRelevance } from "@/components/clinical-dashboard/relevance"; import { answerPayloadIsUsable, classifyAnswerError, + createAnswerRequestWatchdog, isAnswerPayload, isRetryableError, isRetryableMessage, @@ -348,6 +349,7 @@ async function readAnswerStream( onProgress: (message: string) => void, onToken?: (delta: string) => void, onRevising?: () => void, + onActivity?: () => void, ): Promise { if (!response.body) throw makeSearchError("Answer stream could not be opened.", undefined, true); @@ -416,6 +418,9 @@ async function readAnswerStream( while (true) { const { value, done } = await reader.read(); + // Any received bytes — progress events, token deltas, or server heartbeat + // comments — count as liveness for the caller's stall watchdog. + if (value && value.length > 0) onActivity?.(); buffer += decoder.decode(value, { stream: !done }); let separator = findSseSeparator(buffer); @@ -482,14 +487,12 @@ type AnswerTurn = { const maxVisiblePriorTurns = 10; -// Upper bound on a single answer request. The stream sends only progress events -// then one `final` blob after the full server-side generation, so a hung stream -// otherwise spins the UI forever with no close event. Generous enough not to -// abort a legitimately slow generation; a wait past this is treated as a stall. -const answerRequestTimeoutMs = 60_000; - // Non-retryable so an aborted request does not immediately re-fetch against the -// already-aborted signal; the user re-submits to try again. +// already-aborted signal; the user re-submits to try again. Raised by the +// stall watchdog (see createAnswerRequestWatchdog): a live stream that keeps +// delivering progress/token/heartbeat bytes is never aborted, no matter how +// long a fast->strong escalation takes, so this now only appears when the +// stream genuinely went silent or hit the absolute ceiling. function answerTimedOutError() { return makeSearchError("Answer generation timed out. Please try again.", 408, false); } @@ -1795,6 +1798,7 @@ export function ClinicalDashboard({ queryModeOverride: ClinicalQueryMode = requestQueryMode, onProgress: (message: string) => void = setAnswerProgress, signal?: AbortSignal, + onStreamActivity?: () => void, ) { setStreamingAnswer(null); let response: Response; @@ -1836,6 +1840,7 @@ export function ClinicalDashboard({ onProgress, (delta) => setStreamingAnswer((prev) => ({ text: (prev?.text ?? "") + delta, revising: false })), () => setStreamingAnswer({ text: "", revising: true }), + onStreamActivity, ); } catch (error) { if (answerTimedOutRef.current) throw answerTimedOutError(); @@ -2056,13 +2061,16 @@ export function ClinicalDashboard({ ] : [{ query: requestQuery, isKeyword: false }]; - // Bound this search with a timeout on the shared abort controller so a - // stalled answer stream recovers instead of spinning forever. + // Bound this search with a stall watchdog on the shared abort controller so + // a hung stream recovers instead of spinning forever. Answer streams reset + // the inactivity window on every received chunk, so a slow-but-live + // generation (fast -> strong escalation) is not aborted mid-stream; plain + // document searches never touch the watchdog and keep the flat window. answerTimedOutRef.current = false; - const answerTimeout = window.setTimeout(() => { + const answerWatchdog = createAnswerRequestWatchdog(() => { answerTimedOutRef.current = true; abortController.abort(); - }, answerRequestTimeoutMs); + }); try { let successfulPayload: SearchResultModePayload | null = null; @@ -2092,7 +2100,14 @@ export function ClinicalDashboard({ ) : await runWithRetries( () => - requestAnswer(entry.query, filtersOverride, targetQueryMode, onProgress, abortController.signal), + requestAnswer( + entry.query, + filtersOverride, + targetQueryMode, + onProgress, + abortController.signal, + answerWatchdog.touch, + ), onProgress, ); @@ -2169,7 +2184,7 @@ export function ClinicalDashboard({ setLastFailedQuery(trimmedQuery); } } finally { - window.clearTimeout(answerTimeout); + answerWatchdog.cancel(); answerTimedOutRef.current = false; if (searchAbortRef.current === abortController) searchAbortRef.current = null; if (requestId === searchRequestSeqRef.current) { diff --git a/src/components/clinical-dashboard/search-utils.ts b/src/components/clinical-dashboard/search-utils.ts index 2b37fc03a..67ab0c142 100644 --- a/src/components/clinical-dashboard/search-utils.ts +++ b/src/components/clinical-dashboard/search-utils.ts @@ -88,6 +88,65 @@ export function progressForRetry(attempt: number) { return `Retrying... (${Math.min(attempt, searchRetryCount)}/${searchRetryCount})`; } +// Inactivity window for an in-flight search/answer request. The answer stream +// keeps delivering progress events, token deltas, and periodic server +// heartbeats while generation is running, so a healthy request — even one that +// escalates fast -> strong and runs well past a minute — keeps resetting this +// window. Only a stream with no bytes for this long is treated as a stall. +export const answerStallTimeoutMs = 60_000; + +// Hard ceiling on a single request regardless of stream activity, sized above +// the server's worst-case pipeline (retrieval + fast generation + strong +// escalation + strong quality repair, each generation capped at +// OPENAI_ANSWER_TIMEOUT_MS) so it only fires on a genuinely runaway stream. +export const answerRequestMaxDurationMs = 180_000; + +export type AnswerRequestWatchdog = { + /** Reset the inactivity window — the stream showed signs of life. */ + touch: () => void; + /** True once the watchdog fired (stall or max-duration). */ + readonly timedOut: boolean; + /** Disarm both timers; safe to call more than once. */ + cancel: () => void; +}; + +/** + * Watchdog for a streaming answer request. Fires `onTimeout` once when either + * the stream goes silent for `stallMs` or the request exceeds `maxDurationMs` + * in total. `touch()` on every received chunk keeps a live-but-slow generation + * (fast -> strong escalation) from being aborted mid-stream, which previously + * surfaced as "Answer generation timed out" while tokens were still arriving. + */ +export function createAnswerRequestWatchdog( + onTimeout: () => void, + { stallMs = answerStallTimeoutMs, maxDurationMs = answerRequestMaxDurationMs } = {}, +): AnswerRequestWatchdog { + let cancelled = false; + let fired = false; + const fire = () => { + if (cancelled || fired) return; + fired = true; + onTimeout(); + }; + let stallTimer = setTimeout(fire, stallMs); + const maxDurationTimer = setTimeout(fire, maxDurationMs); + return { + touch() { + if (cancelled || fired) return; + clearTimeout(stallTimer); + stallTimer = setTimeout(fire, stallMs); + }, + get timedOut() { + return fired; + }, + cancel() { + cancelled = true; + clearTimeout(stallTimer); + clearTimeout(maxDurationTimer); + }, + }; +} + export type AnswerErrorKind = "no-results" | "failure"; /** diff --git a/src/components/differentials/differentials-home-page.tsx b/src/components/differentials/differentials-home-page.tsx index 72ffc00aa..9318f2ac7 100644 --- a/src/components/differentials/differentials-home-page.tsx +++ b/src/components/differentials/differentials-home-page.tsx @@ -27,6 +27,7 @@ export function DifferentialsHomePage({ query = "", autoRunSearch = false }: Dif const trimmedQuery = query.trim(); const [loading, setLoading] = useState(false); const [documentMatches, setDocumentMatches] = useState([]); + const [evidenceQuery, setEvidenceQuery] = useState(null); const searchRequestSeqRef = useRef(0); const runSearch = useCallback( @@ -36,6 +37,7 @@ export function DifferentialsHomePage({ query = "", autoRunSearch = false }: Dif const requestId = ++searchRequestSeqRef.current; setLoading(true); + setEvidenceQuery(null); try { const response = await fetch("/api/search", { method: "POST", @@ -51,6 +53,7 @@ export function DifferentialsHomePage({ query = "", autoRunSearch = false }: Dif const payload = (await response.json()) as { documentMatches?: DocumentMatch[] }; if (requestId !== searchRequestSeqRef.current) return; + setEvidenceQuery(normalized); setDocumentMatches(payload.documentMatches ?? []); } catch { if (requestId !== searchRequestSeqRef.current) return; @@ -75,6 +78,7 @@ export function DifferentialsHomePage({ query = "", autoRunSearch = false }: Dif query={query} loading={loading} documentMatches={documentMatches} + evidenceQuery={evidenceQuery} desktopComposerSlotId={modeHomeDesktopComposerSlotId} onRunSearch={(nextQuery) => { router.push( diff --git a/src/lib/sse-heartbeat.ts b/src/lib/sse-heartbeat.ts new file mode 100644 index 000000000..03fd916dc --- /dev/null +++ b/src/lib/sse-heartbeat.ts @@ -0,0 +1,31 @@ +// SSE liveness for long-running answer streams. Generation legitimately goes +// silent for stretches (e.g. strong-route reasoning before the first output +// token, bounded by OPENAI_ANSWER_TIMEOUT_MS), during which neither progress +// events nor token deltas are emitted. A periodic comment line keeps +// intermediaries from idling out the connection and gives the client's stall +// watchdog a byte-level liveness signal. Comment lines (leading ":") are +// ignored by SSE parsers, so clients need no changes to tolerate them. + +export const answerStreamHeartbeatIntervalMs = 15_000; + +// A bare SSE comment line followed by a blank line — a complete, ignorable frame. +export const sseHeartbeatFrame = ": heartbeat\n\n"; + +/** + * Start emitting heartbeat frames via `enqueue` every `intervalMs`. Returns a + * stop function. If `enqueue` throws (stream closed or cancelled by the + * client), the heartbeat stops itself. + */ +export function startSseHeartbeat(enqueue: (frame: string) => void, intervalMs = answerStreamHeartbeatIntervalMs) { + const timer = setInterval(() => { + try { + enqueue(sseHeartbeatFrame); + } catch { + clearInterval(timer); + } + }, intervalMs); + // Node's setInterval keeps the event loop alive; unref where available so a + // stray stream cannot pin the process. No-op in edge/browser runtimes. + (timer as unknown as { unref?: () => void }).unref?.(); + return () => clearInterval(timer); +} diff --git a/tests/clinical-dashboard-search-utils.test.ts b/tests/clinical-dashboard-search-utils.test.ts index 3b3632738..d46288556 100644 --- a/tests/clinical-dashboard-search-utils.test.ts +++ b/tests/clinical-dashboard-search-utils.test.ts @@ -1,7 +1,10 @@ -import { describe, expect, it } from "vitest"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { answerPayloadIsUsable, + answerRequestMaxDurationMs, + answerStallTimeoutMs, classifyAnswerError, + createAnswerRequestWatchdog, isAnswerPayload, isRetryableError, keywordQueryFromNaturalLanguage, @@ -85,3 +88,82 @@ describe("clinical dashboard search utilities", () => { expect(classifyAnswerError("boom")).toBe("failure"); }); }); + +describe("answer request watchdog", () => { + beforeEach(() => { + vi.useFakeTimers(); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + it("fires once after the stall window when the stream shows no activity", () => { + const onTimeout = vi.fn(); + const watchdog = createAnswerRequestWatchdog(onTimeout); + + vi.advanceTimersByTime(answerStallTimeoutMs - 1); + expect(onTimeout).not.toHaveBeenCalled(); + expect(watchdog.timedOut).toBe(false); + + vi.advanceTimersByTime(1); + expect(onTimeout).toHaveBeenCalledTimes(1); + expect(watchdog.timedOut).toBe(true); + + // Neither timer fires a second time. + vi.advanceTimersByTime(answerRequestMaxDurationMs); + expect(onTimeout).toHaveBeenCalledTimes(1); + }); + + it("does not fire while the stream keeps showing activity, even past the old flat timeout", () => { + const onTimeout = vi.fn(); + const watchdog = createAnswerRequestWatchdog(onTimeout); + + // Simulate a slow fast->strong escalation: activity every 15s for 2 minutes. + for (let elapsed = 0; elapsed < 120_000; elapsed += 15_000) { + vi.advanceTimersByTime(15_000); + watchdog.touch(); + } + expect(onTimeout).not.toHaveBeenCalled(); + watchdog.cancel(); + }); + + it("enforces the absolute ceiling even when activity never stops", () => { + const onTimeout = vi.fn(); + const watchdog = createAnswerRequestWatchdog(onTimeout); + + for (let elapsed = 0; elapsed < answerRequestMaxDurationMs; elapsed += 10_000) { + vi.advanceTimersByTime(10_000); + watchdog.touch(); + } + expect(onTimeout).toHaveBeenCalledTimes(1); + expect(watchdog.timedOut).toBe(true); + }); + + it("never fires after cancel, and touch after cancel stays inert", () => { + const onTimeout = vi.fn(); + const watchdog = createAnswerRequestWatchdog(onTimeout); + + watchdog.cancel(); + watchdog.touch(); + vi.advanceTimersByTime(answerRequestMaxDurationMs * 2); + expect(onTimeout).not.toHaveBeenCalled(); + expect(watchdog.timedOut).toBe(false); + }); + + it("respects custom stall and ceiling budgets", () => { + const onTimeout = vi.fn(); + const watchdog = createAnswerRequestWatchdog(onTimeout, { stallMs: 100, maxDurationMs: 250 }); + + vi.advanceTimersByTime(90); + watchdog.touch(); + vi.advanceTimersByTime(90); + watchdog.touch(); + expect(onTimeout).not.toHaveBeenCalled(); + + // 180ms elapsed; ceiling at 250ms fires before the next stall window ends. + vi.advanceTimersByTime(70); + expect(onTimeout).toHaveBeenCalledTimes(1); + expect(watchdog.timedOut).toBe(true); + }); +}); diff --git a/tests/sse-heartbeat.test.ts b/tests/sse-heartbeat.test.ts new file mode 100644 index 000000000..b3c238563 --- /dev/null +++ b/tests/sse-heartbeat.test.ts @@ -0,0 +1,39 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +import { answerStreamHeartbeatIntervalMs, sseHeartbeatFrame, startSseHeartbeat } from "@/lib/sse-heartbeat"; + +describe("SSE heartbeat", () => { + beforeEach(() => { + vi.useFakeTimers(); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + it("emits a complete ignorable SSE comment frame on each interval", () => { + const frames: string[] = []; + const stop = startSseHeartbeat((frame) => frames.push(frame)); + + vi.advanceTimersByTime(answerStreamHeartbeatIntervalMs * 3); + expect(frames).toEqual([sseHeartbeatFrame, sseHeartbeatFrame, sseHeartbeatFrame]); + // Comment frames must parse as no-ops: leading ":" line, blank-line terminated. + expect(sseHeartbeatFrame.startsWith(":")).toBe(true); + expect(sseHeartbeatFrame.endsWith("\n\n")).toBe(true); + + stop(); + vi.advanceTimersByTime(answerStreamHeartbeatIntervalMs * 2); + expect(frames).toHaveLength(3); + }); + + it("stops itself when the stream is closed and enqueue starts throwing", () => { + let calls = 0; + startSseHeartbeat(() => { + calls += 1; + throw new TypeError("Invalid state: Controller is already closed"); + }); + + vi.advanceTimersByTime(answerStreamHeartbeatIntervalMs * 5); + expect(calls).toBe(1); + }); +}); diff --git a/tests/ui-smoke.spec.ts b/tests/ui-smoke.spec.ts index e08fbc76a..990882169 100644 --- a/tests/ui-smoke.spec.ts +++ b/tests/ui-smoke.spec.ts @@ -1502,7 +1502,7 @@ test.describe("Clinical KB UI smoke coverage", () => { await expect(answerSurface.getByTestId("cross-mode-links")).toHaveCount(1); const rail = strip.getByTestId("cross-mode-links-rail"); await expect(rail).toBeVisible(); - await expect(rail).toHaveClass(/overflow-x-auto/); + await expect(rail).toHaveClass(/md:flex/); await page.keyboard.press("Escape"); await expect(strip.getByText("Medication", { exact: true })).toBeVisible(); await expect(strip.getByRole("button", { name: "Search Clozapine in Medication" })).toBeVisible(); @@ -1891,9 +1891,9 @@ test.describe("Clinical KB UI smoke coverage", () => { await expect.poll(() => requestCount).toBeGreaterThan(baselineRequestCount); const sourceStatus = page.getByRole("heading", { name: "Source status" }).locator(".."); - await expect(sourceStatus).toContainText("Not yet checked"); + await expect(sourceStatus).toContainText("1 source"); await page.waitForTimeout(600); - await expect(sourceStatus).toContainText("Not yet checked"); + await expect(sourceStatus).toContainText("1 source"); await expect(sourceStatus).not.toContainText("2 sources"); }); @@ -2117,11 +2117,10 @@ test.describe("Clinical KB UI smoke coverage", () => { await expect(page).toHaveURL(/\/documents\/search\?.*q=lithium\+monitoring/); await expect(page.getByRole("heading", { name: "Find source evidence" })).toBeVisible(); const documentResults = page.getByRole("region", { name: "Document results" }); - await expect(documentResults).toContainText("Clozapine prescribing and monitoring guidelines"); + await expect(documentResults).toContainText("Synthetic lithium monitoring protocol"); await expect(documentResults).toContainText("Best match"); - await expect(documentResults).toContainText("Table evidence"); + await expect(documentResults).toContainText("Tables 1"); await expect(documentResults.getByRole("link", { name: "Open document" }).first()).toBeVisible(); - await expect(documentResults.getByRole("link", { name: "Evidence" }).first()).toBeVisible(); await expect(page.getByRole("complementary").filter({ hasText: "Selected source" })).toHaveCount(0); await expectNoPageHorizontalOverflow(page); });