Skip to content

Share one server-sent events helper across the four streams #434

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

Merged
abhishekgahlot2 merged 1 commit into from
Nov 10, 2026
+705 −556
Merged

Share one server-sent events helper across the four streams #434

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
1 change: 1 addition & 0 deletions backend/AGENTS.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,7 @@ below are relative to `backend/`.
| Untrusted text in prompts | `src/bots/prompt-safe.ts` `promptSafeJson` |
| Secrets in text | `src/secrets/redact.ts` |
| Messaging channels | `src/connectors/types.ts` (`Transport`, `Renderer`), `src/connectors/email/` as the worked example |
| An event stream to the browser | `src/util/sse.ts` `sseResponse`: headers, opening comment, heartbeat and teardown on disconnect; the route only writes frames |

The guard tests: `test/ratchet.test.ts` (an 800-line file cap with shrink-only
baselines, the dependency law, the product-name ratchet) and
Expand Down
146 changes: 52 additions & 94 deletions backend/src/chat/routes.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@ import { isMemoryScope, type MemoryScope } from "../memory/scope";
import { providerCredentialMissingMessage } from "../engines/provider-credential-gate";
import { credentialWaitSignal, resolveChatProviderCredential } from "../provider-gateway/credentials";
import { awaitWithSignal } from "../util/abortable-operation";
import { sseResponse } from "../util/sse";
import {
buildResourceAccessSnapshot,
formatResourceAccessContext,
Expand Down Expand Up @@ -66,9 +67,9 @@ chatRoutes.get("/models", async (c) => {
// Emits `event: context` (citations) once, then a burst of `event: delta` text
// tokens, then `event: done`. A failure surfaces as `event: error`. NO sandbox.
//
// Built as a raw ReadableStream (not hono streamSSE) so we own every header -
// `no-transform` + `X-Accel-Buffering: no` stop proxies buffering the stream
// (the same SSE-hygiene the runs `/events` route relies on).
// Served through `sseResponse` (util/sse.ts), the same stream helper the runs
// `/events` route relies on for headers, heartbeat and teardown when the client
// aborts, cancels the body or a write to it fails.
chatRoutes.post("/", async (c) => {
if (!chatLlmEnabled()) return c.json({ error: "chat is turned off" }, 503);
let body: { messages?: unknown; model?: unknown; memoryScope?: unknown };
Expand Down Expand Up @@ -127,101 +128,58 @@ chatRoutes.post("/", async (c) => {
const query = [...messages].reverse().find((m) => m.role === "user")?.content ?? "";
const threadId = `chat:${orgId}`;

const encoder = new TextEncoder();
const signal = c.req.raw.signal;

const stream = new ReadableStream<Uint8Array>({
start(controller) {
let closed = false;
const send = (frame: string): void => {
if (closed) return;
try {
controller.enqueue(encoder.encode(frame));
} catch {
/* controller already closed (client gone) */
return sseResponse({ signal }, (stream) => {
void (async () => {
try {
// Read-only retrieval first (best-effort, never throws) so the UI can
// show honest Sources before the answer streams.
const [context, resourceSnapshot] = await Promise.all([
retrieveChatContext({ orgId, userId, query, memoryScope, threadId }),
userId
? buildResourceAccessSnapshot(
{
orgId,
userId,
runId: threadId,
resources: [],
repos: [],
},
undefined,
{ inlineLimit: 500, exactInventoryTool: null },
)
: Promise.resolve(null),
]);
if (stream.closed) return;
stream.sendEvent("context", { citations: context.citations });

const system = [
CHAT_SYSTEM_PROMPT,
resourceSnapshot ? formatResourceAccessContext(resourceSnapshot) : "",
context.block,
].filter(Boolean).join("\n\n");
const llmMessages: ChatMessage[] = [{ role: "system", content: system }, ...messages];
let answer = "";
for await (const delta of chatTurnStream({ model, orgId, userId }, llmMessages, resolved, signal)) {
if (stream.closed) return;
answer += delta;
stream.sendEvent("delta", { delta });
}
};
const sendEvent = (event: string, data: unknown): void =>
send(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`);

// Prime headers/first bytes, then heartbeat idle streams.
send(": open\n\n");
const heartbeat = setInterval(() => send(": ping\n\n"), 25_000);
heartbeat.unref?.();

const cleanup = (): void => {
if (closed) return;
closed = true;
clearInterval(heartbeat);
signal.removeEventListener("abort", cleanup);
try {
controller.close();
} catch {
/* already closed */
if (!stream.closed) {
stream.sendEvent("done", {});
// Governed capture parity (item 7): a COMPLETED exchange (never an
// aborted stream) enqueues through the same outbox + salience gate
// as runs, marked with the chat origin. Best-effort by contract —
// captureChatExchange never throws into the stream.
void captureChatExchange({ orgId, userId, memoryScope, prompt: query, summary: answer, model });
}
};
if (signal.aborted) return cleanup();
signal.addEventListener("abort", cleanup);

void (async () => {
try {
// Read-only retrieval first (best-effort, never throws) so the UI can
// show honest Sources before the answer streams.
const [context, resourceSnapshot] = await Promise.all([
retrieveChatContext({ orgId, userId, query, memoryScope, threadId }),
userId
? buildResourceAccessSnapshot(
{
orgId,
userId,
runId: threadId,
resources: [],
repos: [],
},
undefined,
{ inlineLimit: 500, exactInventoryTool: null },
)
: Promise.resolve(null),
]);
if (closed) return;
sendEvent("context", { citations: context.citations });

const system = [
CHAT_SYSTEM_PROMPT,
resourceSnapshot ? formatResourceAccessContext(resourceSnapshot) : "",
context.block,
].filter(Boolean).join("\n\n");
const llmMessages: ChatMessage[] = [{ role: "system", content: system }, ...messages];
let answer = "";
for await (const delta of chatTurnStream({ model, orgId, userId }, llmMessages, resolved, signal)) {
if (closed) return;
answer += delta;
sendEvent("delta", { delta });
}
if (!closed) {
sendEvent("done", {});
// Governed capture parity (item 7): a COMPLETED exchange (never an
// aborted stream) enqueues through the same outbox + salience gate
// as runs, marked with the chat origin. Best-effort by contract —
// captureChatExchange never throws into the stream.
void captureChatExchange({ orgId, userId, memoryScope, prompt: query, summary: answer, model });
}
} catch (error) {
if (!closed) sendEvent("error", { error: chatFailure(error).reason });
} finally {
cleanup();
}
})();
},
});

return new Response(stream, {
headers: {
"Content-Type": "text/event-stream; charset=utf-8",
"Cache-Control": "no-cache, no-transform",
Connection: "keep-alive",
"X-Accel-Buffering": "no",
},
} catch (error) {
if (!stream.closed) stream.sendEvent("error", { error: chatFailure(error).reason });
} finally {
stream.close();
}
})();
});
});

Expand Down
58 changes: 9 additions & 49 deletions backend/src/runs/changes-route.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
import type { Hono } from "hono";
import type { AppEnv } from "../http";
import { sseResponse } from "../util/sse";
import { clientOrgChangeForUser, subscribeOrg } from "./org-signals";

/** GET /api/runs/changes - the tenant-scoped invalidation stream. */
Expand All @@ -11,54 +12,13 @@ export function registerRunChangesRoute(routes: Hono<AppEnv>): void {
routes.get("/changes", (c) => {
const orgId = c.get("orgId");
const userId = c.get("userId");
const encoder = new TextEncoder();
const signal = c.req.raw.signal;

const body = new ReadableStream<Uint8Array>({
start(controller) {
let closed = false;
const send = (frame: string): void => {
if (closed) return;
try {
controller.enqueue(encoder.encode(frame));
} catch {
cleanup();
}
};
const unsubscribe = subscribeOrg(orgId, (change) => {
const clientChange = clientOrgChangeForUser(change, userId);
if (!clientChange) return;
send(`event: change\ndata: ${JSON.stringify(clientChange)}\n\n`);
});
const heartbeat = setInterval(() => send(": ping\n\n"), 25_000);
heartbeat.unref?.();

function cleanup(): void {
if (closed) return;
closed = true;
clearInterval(heartbeat);
unsubscribe();
signal.removeEventListener("abort", cleanup);
try {
controller.close();
} catch {
// The browser may already have closed the stream.
}
}

send(": open\nretry: 1500\n\n");
if (signal.aborted) cleanup();
else signal.addEventListener("abort", cleanup);
},
});

return new Response(body, {
headers: {
"Content-Type": "text/event-stream; charset=utf-8",
"Cache-Control": "no-cache, no-transform",
Connection: "keep-alive",
"X-Accel-Buffering": "no",
},
});
// `retry` has the browser reconnect quickly after a drop; the subscription
// is the only thing to undo when it leaves.
return sseResponse({ signal: c.req.raw.signal, retryMs: 1500 }, (stream) =>
subscribeOrg(orgId, (change) => {
const clientChange = clientOrgChangeForUser(change, userId);
if (clientChange) stream.sendEvent("change", clientChange);
}),
);
});
}
Loading
Loading