fix: streaming stalls and truncated answers caused by output-rails buffer misconfiguration

notification-server/src/connectionManager.js

@@ -1,5 +1,52 @@
 const activeConnections = new Map();
 
+/**
+ * Register an AbortController for an in-flight upstream request belonging to a
+ * connection, so it can be cancelled when the browser disconnects.
+ * @param {string} connectionId
+ * @param {AbortController} controller
+ */
+function registerAbortController(connectionId, controller) {
+  const connData = activeConnections.get(connectionId);
+  if (!connData) return;
+  if (!connData.abortControllers) {
+    connData.abortControllers = new Set();
+  }
+  connData.abortControllers.add(controller);
+}
+
+/**
+ * Remove a previously registered AbortController (upstream request finished).
+ * @param {string} connectionId
+ * @param {AbortController} controller
+ */
+function unregisterAbortController(connectionId, controller) {
+  const connData = activeConnections.get(connectionId);
+  connData?.abortControllers?.delete(controller);
+}
+
+/**
+ * Abort every in-flight upstream request for a connection. Called when the
+ * browser goes away, so we stop paying for generation nobody will read.
+ * @param {string} connectionId
+ */
+function abortConnectionRequests(connectionId) {
+  const connData = activeConnections.get(connectionId);
+  if (!connData?.abortControllers) return;
+
+  for (const controller of connData.abortControllers) {
+    try {
+      controller.abort();
+    } catch (error) {
+      console.error(`Failed to abort upstream request for ${connectionId}:`, error);
+    }
+  }
+  connData.abortControllers.clear();
+}
+
 module.exports = {
   activeConnections,
+  registerAbortController,
+  unregisterAbortController,
+  abortConnectionRequests,
 };

notification-server/src/server.js

@@ -82,4 +82,12 @@ const server = app.listen(serverConfig.port, () => {
   console.log(`LLM orchestration streaming at: /channels/:channelId/orchestrate/stream`);
 });
 
+// The SSE GET and the trigger POST are both long-lived by design, so Node's
+// default 300s requestTimeout would cut them off mid-answer. Disable the
+// per-request cap and let the upstream idle watchdog in streamingService.js
+// decide when a stream has genuinely stalled.
+server.requestTimeout = 0;
+server.headersTimeout = 65_000;
+server.keepAliveTimeout = 61_000;
+
 module.exports = server;

notification-server/src/sseUtil.js

@@ -1,18 +1,26 @@
 const { v4: uuidv4 } = require('uuid');
 const streamQueue = require("./streamQueue");
 const { createLLMOrchestrationStreamRequest } = require("./streamingService");
-const { activeConnections } = require("./connectionManager");
+const { activeConnections, abortConnectionRequests } = require("./connectionManager");
+
+// Comment frames are written this often so that every intermediate proxy sees
+// traffic and does not close the connection on its idle timer. EventSource
+// ignores comment frames, so this is invisible to the browser.
+const HEARTBEAT_INTERVAL_MS = Number(
+  process.env.SSE_HEARTBEAT_INTERVAL_MS || 15_000
+);
 
 function buildSSEResponse({ res, req, buildCallbackFunction, channelId }) {
   addSSEHeader(req, res);
-  keepStreamAlive(res);
+  const heartbeat = keepStreamAlive(res);
   const connectionId = generateConnectionID();
   const sender = buildSender(res);
 
   activeConnections.set(connectionId, {
     res,
     sender,
     channelId,
+    abortControllers: new Set(),
   });
 
   if (channelId) {
@@ -25,6 +33,9 @@ function buildSSEResponse({ res, req, buildCallbackFunction, channelId }) {
 
   req.on("close", () => {
     console.log(`Client disconnected from SSE for channel ${channelId}`);
+    clearInterval(heartbeat);
+    // Cancel any in-flight upstream generation - nobody is left to read it.
+    abortConnectionRequests(connectionId);
     activeConnections.delete(connectionId);
     cleanUp?.();
   });
@@ -37,6 +48,8 @@ function addSSEHeader(req, res) {
     'Content-Type': 'text/event-stream',
     'Cache-Control': 'no-cache',
     'Connection': 'keep-alive',
+    // Stops nginx-style proxies buffering the stream into a single response.
+    'X-Accel-Buffering': 'no',
     'Access-Control-Allow-Origin': origin,
     'Access-Control-Allow-Credentials': true,
     'Access-Control-Expose-Headers': 'Origin, X-Requested-With, Content-Type, Cache-Control, Connection, Accept'
@@ -49,8 +62,33 @@ function extractOrigin(reqOrigin) {
   return whitelisted ? reqOrigin : '*';
 }
 
+/**
+ * Keep the SSE connection warm with periodic comment frames.
+ *
+ * Previously a single `res.write('')`, which did nothing beyond the initial
+ * flush - any proxy between the browser and this server would still time the
+ * connection out during a long generation pause.
+ *
+ * @returns {NodeJS.Timeout} interval handle; the caller must clear it on close.
+ */
 function keepStreamAlive(res) {
   res.write('');
+  const heartbeat = setInterval(() => {
+    try {
+      // A `:` line is an SSE comment: ignored by EventSource, but it is traffic.
+      res.write(': ping\n\n');
+      if (typeof res.flush === "function") {
+        res.flush();
+      }
+    } catch (error) {
+      console.error("SSE heartbeat write failed:", error);
+      clearInterval(heartbeat);
+    }
+  }, HEARTBEAT_INTERVAL_MS);
+
+  // Do not hold the event loop open purely for a heartbeat.
+  heartbeat.unref?.();
+  return heartbeat;
 }
 
 function generateConnectionID() {