Skip to content

fix: abort cancelled answer streams #757

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
BigSimmo wants to merge 4 commits into from
+80 −11
Closed

fix: abort cancelled answer streams #757

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
f86004e
docs: record answer stream review
BigSimmo Jul 17, 2026
23b4a73
Merge remote-tracking branch 'origin/main' into codex/harden-high-val…
Copilot Jul 17, 2026
1b8bd4d
docs: record PR policy verification results in branch review ledger
Copilot Jul 17, 2026
02f9c59
fix: add abort guard in test and consistent fallback message in strea…
Copilot Jul 17, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions docs/branch-review-ledger.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,8 @@ Use this ledger to prevent repeated branch and PR reviews when the reviewed HEAD

| Date | Branch or ref | Reviewed HEAD | Scope | Outcome | Checks |
| ---------- | -------------------------------------------------------- | ---------------------------------------- | ---------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| 2026-07-17 | PR #757 / codex/harden-high-value-api-route-family | 23b4a73 | PR policy compliance fix + verification | Updated PR description to include required `## Summary`, `## Verification`, `## Clinical Governance Preflight`, and `## Risk and rollout` sections. All 12 focused Vitest tests pass (tests/private-rag-access.test.ts), ESLint clean, TypeScript clean. | Focused Vitest 12/12 (`tests/private-rag-access.test.ts`); `npm run lint` passed; `npm run typecheck` passed. |
| 2026-07-17 | codex/harden-answer-streaming | b35b395 | answer-stream SSE cancellation hardening | Fixed a high-confidence cancellation defect: cancelling an SSE response body did not create the signal passed to retrieval/generation, so paid work could continue after the consumer left. The route now aborts both answer and summary work on request abort or stream cancellation and suppresses post-cancellation SSE errors. No further high-confidence defect was found in the changed scope. | `git diff --check` passed. `npm run eval:rag:offline` validated 36 golden fixtures and 21 suites before its Vitest stage was blocked by missing `node_modules/vitest`. Focused Vitest and `verify:cheap` were blocked by absent local dependencies; `verify:pr-local -- --dry-run` selected runtime, format, lint, typecheck, unit, build, and fixture checks. |
| 2026-07-17 | PR #635 / claude/github-actions-codex-issue-f4t4s5 | ab09a8d52cc0a8a7e71b37885aaa358aae2522c8 | post-merge merge-readiness review | Already squash-merged to main on 2026-07-14 by BigSimmo. No open review threads or inline comments. CI required checks all green (Change scope, Static PR checks, Safety and config checks, Unit coverage, PR required, Semgrep, Gitleaks, GitGuardian); UI/build/migration jobs correctly skipped. Landed diff is test/guard hardening only for missing `CODEX_TRIGGER_TOKEN` graceful skip. No high-confidence P0-P2 defect. Source branch already deleted. No further merge action needed. | Hosted CI status via `gh pr checks 635` (all required pass); local `node scripts/check-codex-autofix-workflow.mjs` pass; focused Vitest `tests/codex-autofix-workflow.test.ts` 41/41. No OpenAI/Supabase/provider writes. |
| 2026-07-17 | PR #718 / codex/performance-latency-remediation-20260717 | b5f509744d4f4bac74d414644cd1802f64b97fa9 | CodeRabbit performance and SQL correctness follow-up | Resolved nine confirmed findings and dispositioned one stale test comment: document downloads revalidate signed URLs on every action; committed-generation filtering precedes detail pagination; enrichment fallback errors preserve identity; caller cancellation leaves the shared classifier flight alive; registry seeding preserves its cache signal; aliases emit canonical corrections; rate-limit success metadata is coherent; ambiguous upserts use named constraints; and grantable default ACLs fail closed. The proxy mock duplicate was not present. No remaining high-confidence P0-P2 defect was found. | Integrated focused Vitest 122/122; post-format Vitest 71/71; `npm run verify:cheap` passed runtime/policy/static guards, ESLint, TypeScript, and 2,684/2,684 tests; focused Prettier and `git diff --check`; disposable Docker replay, regenerated drift manifest, and transactional local SQL probes. No OpenAI calls, live Supabase DDL/migration/data write, deployment, or production mutation ran. |
| 2026-07-17 | PR #718 / codex/performance-latency-remediation-20260717 | d47ef7a329256687a615c33dd311806c2c1214a8 | hosted UI and migration-order merge-blocker follow-up | Fixed both integration defects exposed by the exact-head UI run: the document scope surface now triggers the deferred, deduplicated catalogue load, and viewer navigation closes unrelated disclosures before opening or scrolling to the selected section. Updated SSR-aware browser fixtures without restoring the removed detail request. Renumbered all three new migrations after the latest production migration and regenerated the drift manifest. No remaining high-confidence P0-P2 defect was found in the follow-up diff. | Focused Vitest 65/65; scoped ESLint and Prettier; `git diff --check`; isolated production Webpack build and TypeScript; focused Chromium 7/7 including both stress viewports; Docker schema replay and drift-manifest regeneration passed in 15 seconds with unchanged schema SHA. No OpenAI calls or live Supabase DDL, migration, rate-limit RPC, or data write ran. |
Expand Down
30 changes: 27 additions & 3 deletions src/app/api/answer/stream/route.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -132,6 +132,19 @@ function streamAnswer(body: AnswerRequestBody, accessScope: RetrievalAccessScope
const ownerId = accessScope.ownerId;
const encoder = new TextEncoder();
const interactionId = randomUUID();
// A Request signal is normally aborted when the HTTP client disconnects, but
// a ReadableStream consumer can also cancel its body independently. Combine
// both paths so retrieval and generation never continue after either kind of
// cancellation (and do not emit a misleading SSE error after cancellation).
const streamAbortController = new AbortController();
const abortFromRequest = () => {
if (!streamAbortController.signal.aborted) {
streamAbortController.abort(signal?.reason ?? new DOMException("The answer stream was aborted.", "AbortError"));
}
};
if (signal?.aborted) abortFromRequest();
else signal?.addEventListener("abort", abortFromRequest, { once: true });
const streamSignal = streamAbortController.signal;

return new Response(
new ReadableStream({
Expand Down Expand Up @@ -203,7 +216,7 @@ function streamAnswer(body: AnswerRequestBody, accessScope: RetrievalAccessScope
}
const answer =
body.summaryMode && body.documentId
? await summarizeDocument(body.documentId, ownerId, { signal })
? await summarizeDocument(body.documentId, ownerId, { signal: streamSignal })
: await answerQuestionWithScope({
query: body.query,
documentId: singleDocumentScope ? body.documentId : undefined,
Expand All @@ -215,7 +228,7 @@ function streamAnswer(body: AnswerRequestBody, accessScope: RetrievalAccessScope
allowGlobalSearch: !ownerId,
queryMode: body.queryMode,
onProgress,
signal,
signal: streamSignal,
});
const governedResponse = buildGovernedAnswerClientResponse(answer);

Expand All @@ -240,11 +253,17 @@ function streamAnswer(body: AnswerRequestBody, accessScope: RetrievalAccessScope
sendFinal({ ...buildDemoStreamAnswer(body, fallbackReason), interactionId });
return;
}
logStreamError(error, signal);
// Cancellation is a terminal client state, not an SSE failure. In
// particular, never enqueue an error after the client has cancelled
// the body: it can race with a new attempt and appear as a duplicate
// visible failure in the browser.
if (streamSignal.aborted) return;
logStreamError(error, streamSignal);
const streamError = streamErrorPayload(error);
send("error", { error: streamError.message, status: streamError.status, details: streamError.details });
} finally {
stopHeartbeat();
signal?.removeEventListener("abort", abortFromRequest);
// The client may have already cancelled the stream (Stop button /
// watchdog abort), in which case close() throws on a closed stream.
try {
Expand All @@ -254,6 +273,11 @@ function streamAnswer(body: AnswerRequestBody, accessScope: RetrievalAccessScope
}
}
},
cancel(reason) {
if (!streamSignal.aborted) {
streamAbortController.abort(reason ?? new DOMException("The answer stream was aborted.", "AbortError"));
}
},
}),
{
headers: {
Expand Down
59 changes: 51 additions & 8 deletions tests/private-rag-access.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -94,7 +94,10 @@ function sampleSearchResult() {
};
}

function mockRuntime(options: { demoMode?: boolean } = {}) {
function mockRuntime(options: {
demoMode?: boolean;
answerQuestionWithScope?: (args: Record<string, unknown>) => Promise<unknown>;
} = {}) {
vi.resetModules();

class MockAuthenticationError extends Error {
Expand Down Expand Up @@ -146,13 +149,16 @@ function mockRuntime(options: { demoMode?: boolean } = {}) {
rrf_top_score: 0.03,
},
}));
const answerQuestionWithScope = vi.fn(async () => ({
answer: "Source-backed answer.",
grounded: true,
confidence: "high",
citations: [],
sources: [],
}));
const answerQuestionWithScope = vi.fn(
options.answerQuestionWithScope ??
(async () => ({
answer: "Source-backed answer.",
grounded: true,
confidence: "high",
citations: [],
sources: [],
})),
);
const fetchRelatedDocuments = vi.fn(async () => []);
const demoSearch = vi.fn(() => []);
const demoAnswer = vi.fn(() => ({
Expand Down Expand Up @@ -428,6 +434,43 @@ describe("private RAG API access", () => {
);
});

it("aborts retrieval and generation when the SSE consumer cancels the response body", async () => {
let generationSignal: AbortSignal | undefined;
let resolveStarted: (() => void) | undefined;
const started = new Promise<void>((resolve) => {
resolveStarted = resolve;
});
const mocks = mockRuntime({
answerQuestionWithScope: async (args) => {
generationSignal = args.signal as AbortSignal;
resolveStarted?.();
await new Promise<never>((_resolve, reject) => {
if (!generationSignal) {
reject(new Error("generationSignal was not set"));
return;
}
if (generationSignal.aborted) {
reject(generationSignal.reason);
return;
}
generationSignal.addEventListener("abort", () => reject(generationSignal?.reason), { once: true });
});
},
});
const { POST } = await import("../src/app/api/answer/stream/route");

const response = await POST(jsonRequest("/api/answer/stream", { query: "clozapine monitoring" }));
const reader = response.body?.getReader();
expect(reader).toBeDefined();
await reader?.read();
await started;

await reader?.cancel("client navigated away");

expect(generationSignal?.aborted).toBe(true);
expect(mocks.answerQuestionWithScope).toHaveBeenCalledTimes(1);
});

it("silently strips legacy public skipCache from streamed answer requests", async () => {
const mocks = mockRuntime();
const { POST } = await import("../src/app/api/answer/stream/route");
Expand Down