Activity Stream

The activity stream is Polyant’s in-process pub/sub bus for operational events. Every interesting moment in the engine - an inbound message, a tool call, a memory write, a scheduled task firing, a webhook match - is emitted as a structured event onto a single shared bus. Subscribers (today: the SSE controller that feeds the Activity admin panel) attach to that bus and receive events synchronously.

It is deliberately small and unfancy: one process, one EventEmitter, a bounded ring buffer for late subscribers, no persistence. If you need durable history, look at Audit Logs or Conversations.

Design goals

• Zero coupling to consumers. Producers don’t know who is listening; they call an emitter function and move on.
• Synchronous emit. No queueing, no async hops between producer and bus. A producer that has work to do is never blocked waiting on a subscriber.
• Late-subscriber friendly. Reconnecting clients see a small replay of recent events so the UI doesn’t appear empty.
• Bounded memory. The ring buffer caps at 100 events; a crashed subscriber can’t leak memory.

Topology

                          PRODUCERS
  ┌──────────────────────────────────────────────────────────┐
  │                                                          │
  │  pipeline.ts ──► emitInbound                             │
  │  channel-manager.ts ──► emitOutbound                     │
  │  extractor.ts ──► emitMemory                             │
  │  scheduler.service.ts ──► emitCron                       │
  │  server/webhooks/webhook.controller.ts ──► emitWebhook   │
  │  agent-invoke.helpers.ts ──► emitAgentHandoff(start/end) │
  │  pipeline.ts ──► emitConversation                        │
  │  governance.service.ts ──► emitGovernance                │
  │                                                          │
  │  supervisor tool wrapper ──► tool start/end events       │
  │  ai-gateway (chat/chatStream) ──► bus-emitter (LLM tap)  │
  │      reasoning + reply text events                       │
  │                                                          │
  └──────────────────────────────────────────────────────────┘
                              │
                              ▼
                  ┌──────────────────────────┐
                  │     activity-bus.ts      │
                  │  ──────────────────────  │
                  │  • EventEmitter          │
                  │  • Ring buffer (100)     │
                  │  • subscribe(handler)    │
                  └──────────────────────────┘
                              │
            ┌─────────────────┴─────────────────┐
            ▼                                   ▼
  ┌─────────────────────┐             ┌────────────────────┐
  │ activity-stream     │             │ (future subscribers)│
  │ .controller.ts      │             │   metrics, audit,   │
  │ GET .../live        │             │   external sinks    │
  │ GET .../conversation│             └────────────────────┘
  └─────────────────────┘
            │
            ▼
   Admin panel /activity, conversation Live view

Three emit paths

1. Per-category emitters

Each producer calls a small typed helper from packages/engine/src/activity-stream/emitters/. There is one file per category and the name maps 1:1 to the event kind seen in the UI.

All eight helpers funnel into the single bus via activityBus.publish(event).

2. Tool calls, from the supervisor’s tool wrapper

Every tool the supervisor equips is wrapped by its governance wrapper (wrapToolWithGovernance in packages/engine/src/agents/supervisor/), and that wrapper publishes the tool events: a :start event before the executor runs, so the panel can show the tool in flight, and an :end event with the result preview, the status and the duration once it returns, fails or is blocked. The wrapper is the only place that owns the real start and end of each invocation, streaming or not, so a long multi-step turn shows each tool at the moment it ran.

3. The LLM tap

The supervisor’s reasoning is the other surface worth watching, and the supervisor doesn’t know about the activity bus - it just calls aiGateway.chat() or aiGateway.chatStream(). Rather than thread emit-calls through agent code, the gateway is wrapped by packages/engine/src/activity-stream/bus-emitter.ts:

• Streaming path. chatStream() returns an AsyncIterable whose fullStream is intercepted. Each step’s reasoning is aggregated and published as a thinking event when the step finishes, and the text of a final step as the agent’s reply.
• Non-streaming path. chat() returns the final response.steps[] array; the emitter replays the same reasoning and reply events after the call completes, so Room, webhook and scheduled turns still appear in the feed, just not progressively.

The tap deliberately ignores the SDK’s tool-call chunks: the wrapper above already publishes them with real timings, and replaying them here would double every tool row.

The bus itself

packages/engine/src/activity-stream/activity-bus.ts is ~80 lines. The contract is:

interface ActivityBus {
  publish(event: ActivityEvent): void;            // synchronous fan-out
  subscribe(handler: (e: ActivityEvent) => void): () => void; // returns unsubscribe
  replay(): ActivityEvent[];                      // last ≤100 events, in order
}

The implementation is a singleton (export const activityBus = new ActivityBus()) built on a node EventEmitter plus an in-memory array trimmed to BUFFER_SIZE = 100. Publishing is fire-and-forget: it never throws, never awaits, and never knows whether anyone is listening.

The buffer is global, not per-subscriber. A new subscriber receives the replay synchronously inside subscribe(), then a steady stream of live events from that point on.

One conversation

GET /api/activity-stream/conversation?conversationId=<id>&instanceId=<agent slug> is the same stream narrowed to one conversation, on any channel. It feeds the Live switch of the conversation detail page.

• Access is checked like reading the conversation itself. The route needs conversation:read, the caller must reach the named agent’s workspace, and the conversation must belong to that agent in its organization. A missing conversation and a foreign one get the same 404, so the route does not reveal which conversations exist. Both parameters are required (400 without them).
• It carries one more signal. Besides the bus events whose conversation id matches, it sends a named persisted event each time a message or a hook execution for the conversation is stored, so a client can re-read the transcript through the normal read path instead of reconstructing it from events.
• It shares the live feed’s limits. Both routes count against the same two connection caps, the total and the per-user one, both installation policies; a connection over either cap is refused with 503 and Retry-After: 60.

Why not a queue?

The bus is intentionally not durable, not transactional, not back-pressuring. It is an observability surface, not a delivery channel. Durable work (memory extraction, scheduled task runs, webhook backlog) all has its own persistent queue table elsewhere. If the activity stream drops an event because a subscriber is slow, nothing in the system breaks - the operator just doesn’t see one row in the live view.

See also

• Activity (admin panel) - the UI fed by this bus.
• REST API reference - SSE endpoint - the GET /api/activity-stream/live and GET /api/activity-stream/conversation contracts.

Code references

• packages/engine/src/activity-stream/activity-bus.ts - the bus.
• packages/engine/src/activity-stream/bus-emitter.ts - the LLM tap.
• packages/engine/src/activity-stream/emitters/*.ts - eight per-category emitters.
• packages/engine/src/agents/supervisor/index.ts - the tool wrapper that publishes tool start and end events.
• packages/engine/src/activity-stream/activity-stream.controller.ts - SSE endpoints (GET /api/activity-stream/live, GET /api/activity-stream/conversation).
• packages/engine/src/activity-stream/event-formatters.ts - domain-event → SSE payload mapping.