append data prefix, env var gate

Author: huangjeff5

genkit-tools/telemetry-server/src/broadcast-manager.ts

@@ -71,7 +71,7 @@ export class BroadcastManager {
     }
 
     const data = JSON.stringify(event);
-    const messageToSend = `${data}\n\n`;
+    const messageToSend = `data: ${data}\n\n`;
 
     // Note: response.write() doesn't throw on dead sockets in Express - it
     // returns false and the 'close' event handles cleanup. No try/catch needed.

genkit-tools/telemetry-server/src/index.ts

@@ -81,7 +81,7 @@ export async function startTelemetryServer(params: {
       const currentTrace = await params.traceStore.load(traceId);
       if (currentTrace) {
         const snapshot = JSON.stringify(currentTrace);
-        response.write(`${snapshot}\n\n`);
+        response.write(`data: ${snapshot}\n\n`);
       }
 
       // Register this connection for broadcasts

js/core/src/tracing/node-telemetry-provider.ts

@@ -89,8 +89,8 @@ async function cleanUpTracing(): Promise<void> {
  */
 function createTelemetryServerProcessor(): SpanProcessor {
   const exporter = new TraceServerExporter();
-  // Use RealtimeSpanProcessor in dev for real-time span updates
-  return isDevEnv()
+  // Use RealtimeSpanProcessor when explicitly enabled via env var (set by CLI)
+  return isDevEnv() && process.env.GENKIT_ENABLE_REALTIME_TELEMETRY === 'true'
     ? new RealtimeSpanProcessor(exporter)
     : new BatchSpanProcessor(exporter);
 }