🦄 Event-driven agentic loops

Stop mutating conversational state in-place—log every user input, tool result, and LLM chunk, then project that event stream into the UI, the prompt, and persistence as independent views.

Video

Episode Summary

Vaibhav and Anders peel back how SageKit’s chat agent handles real-time approvals, queued follow-ups, and user interrupts without race conditions. The core insight: treat the backend like a game server. Every interaction is an append-only event, and each consumer—LLM loop, UI, persistence—receives a projection that suits its contract. We walk through the architecture, wire up a Bun/Effect-TS prototype, and show how an event log makes queuing, cancellation, and tooling far easier to reason about (and test).

Why Event-Sourced Agents

• Linear agent loops crumble once you need interrupts, approvals, or queued inputs; events give you a single truth you can replay.
• Different surfaces want different stories: the UI should show pending approvals, while the LLM should never see queued user messages until they are active.
• Testing becomes deterministic—replay the same event log and assert the derived state without standing up the UI or the model.

Demo Architecture

• Event bus as the write path. All services publish or subscribe to the same EventBus, making it trivial to fork streams or add instrumentation without rewiring the world.

    return {
      publish: (event: Event) =>
        pipe(
          PubSub.publish(pubsub, event),
          Effect.tap(() =>
            Effect.sync(() => console.log('[EventBus]', event.type))
          )
        ),

      subscribe: <E extends Event>(filter: (event: Event) => event is E) =>
        Stream.fromPubSub(pubsub, { scoped: true }).pipe(
          Effect.map(stream => stream.pipe(Stream.filter(filter)))
        ),

• Reducers own domain logic. The message reducer queues user inputs while streaming and flushes them when the LLM finishes—no shared mutable state, just pure functions reacting to events.

    case 'user_message': {
      if (state.isStreaming || state.streamingMessageIndex !== null) {
        return {
          ...state,
          queuedUserMessages: [
            ...state.queuedUserMessages,
            { id: generateId(), content: event.content, timestamp: event.timestamp }
          ]
        }
      }
      return addMessage(state, {
        id: generateId(),
        role: 'user',
        type: 'text',
        content: event.content,
        timestamp: event.timestamp
      })
    }