Server-Sent Events

Package: @tagna/udiot/server API: createEventStream() Added: v0.16.0

createEventStream() provides a small Express-compatible Server-Sent Events helper for browser live updates. It is designed for read-only streams such as agent progress, build logs, notifications, and telemetry tails. Client input should arrive through normal HTTP requests.

Basic Usage

import express from 'express';
import { createEventStream } from '@tagna/udiot/server';

const app = express();
const runs = createEventStream({
  channel: 'runs',
  heartbeatMs: 15000,
});

app.get('/events/runs', runs.handler);

app.post('/runs/:id/start', async (req, res) => {
  await runs.publish({ runId: req.params.id, status: 'started' }, 'run');
  res.status(202).json({ ok: true });
});

Browser client:

const source = new EventSource('/events/runs');

source.addEventListener('run', (event) => {
  const payload = JSON.parse(event.data);
  console.log(payload.runId, payload.status);
});

Options

Payloads are framed safely: CR, LF and CRLF in data all start a new data: line, so text cannot inject event:, id: or retry: fields, and publish() rejects event types that contain line breaks.

Auth Gate

Supply a real application-owned verifyRunAccess(req) function that checks credentials and authorization for the requested tenant/channel. A header being present is not authentication. The basic public-stream example above has no authentication; do not use it for private run data. Browser EventSource cannot set an arbitrary Authorization header: design secure session/cookie-based access or an appropriate client integration, and do not put long-lived tokens in URLs.

const stream = createEventStream({
  channel: 'org-runs',
  // Application-owned: verify credentials AND access to this stream/channel.
  authorize: async (req) => verifyRunAccess(req),
});

When authorize returns false, the handler responds with 401 and does not subscribe the connection.

Redis Backend

Use Redis when more than one process must publish to the same channel.

import Redis from 'ioredis';
import { createEventStream } from '@tagna/udiot/server';

const stream = createEventStream({
  channel: 'runs',
  backend: 'redis',
  redis: {
    publisher: new Redis(process.env.REDIS_URL),
    subscriber: new Redis(process.env.REDIS_URL),
  },
});

The subscriber connection is dedicated to Redis subscribe mode. Do not reuse it for regular Redis commands.

Shutdown

Call close() during graceful shutdown to end open responses and unsubscribe from Redis channels.

process.on('SIGTERM', async () => {
  await stream.close();
  process.exit(0);
});

Relationship To Run Events

createEventStream() is generic. To stream agent progress to a browser, bridge RunEventStream into SSE:

import { getDefaultRunEventStream } from '@tagna/udiot/telemetry';

const sse = createEventStream({ channel: 'run-events' });
const runEvents = getDefaultRunEventStream();

for await (const event of runEvents.subscribe('run_123')) {
  await sse.publish(event, 'run-event');
}

The SSE helper handles browser transport; RunEventStream handles run-specific history, replay, and telemetry semantics.