The Event-Driven / Publish-Subscribe Pattern, Implemented via the Reactive Reducer

Back to: Intro

A little background

Ten patterns in, and this one is the most surprising - and the one where I have to draw a line through the middle of the pattern and tell you which half is real.

The half that’s real is almost tautological: the reactive reducer is an event-driven system, internally. Every context write is an event. The microstep loop is the reaction. The guards are the subscriptions. The context store is the bus. The pattern this article is supposed to implement is the thing the reducer already is, and everything below about storms, dedup, races and feedback loops follows from that and holds up.

The half that isn’t real is the outside. Pub-sub has two legs - something publishes in, something subscribes out - and the FSM today has neither as a first-class mechanism. There is no stream tap, no webhook receiver, no sensor: an external event reaches a machine only because something in-process calls updateContext. And the emit/send actions, which are genuine DSL syntax the reducer genuinely executes, currently only write a log line. No bus, no webhook, no subscriber. That’s true on the interactive runtime and on the headless targets alike.

So read this article as: the internal event system is the load-bearing claim and it’s sound; the external legs are declared intent with the transport not yet built. I’d rather say that here than let you find it by wiring a dashboard to metric.breach and waiting.

In most frameworks, event-driven multi-agent systems bolt a message broker onto a bunch of agents: each agent subscribes to event types, reacts when a relevant event fires, and publishes its own events. The workflow emerges from the subscriptions. The failure modes are the familiar ones - event storms, duplicate processing, race conditions, accidental feedback loops - and they’re all properties of a loose event system where nothing controls the reactions.

The reactive reducer is a controlled event system. Every reaction is a declared edge; every subscription is a guard; the loop is bounded. The looseness is in the coupling (agents don’t know about each other); the control is in the topology (the machine decides what reacts, and when).

The pattern

Event-Driven / Publish-Subscribe. Agents subscribe to event types and react independently when relevant events are emitted. Instead of one coordinator explicitly directing every transition, work is triggered by changes in the environment or by events produced by other agents. The overall workflow emerges from subscriptions, event contracts, and reactions, making the pattern naturally asynchronous and loosely coupled.

Good for:

  • Long-running systems
  • Asynchronous workflows
  • Distributed agent systems
  • Monitoring and reactive automation
  • Enterprise event architectures
  • Systems where agents should act only when relevant changes occur

Common failure points:

  • Event storms
  • Duplicate processing
  • Race conditions
  • Unclear causality
  • Difficult global reasoning
  • Accidental feedback loops

The machine: the reducer is the bus

A monitoring system. A watcher agent reports a metric event; a set of reactive agents - an alerter, a scaler, a notifier - each react to it. The FSM is the whole system: the watcher drives it, the reactors are summoned by the machine as the edges fire, and the emit actions declare what would go out to a wider bus.

fsm:
  name: metric-monitor
  version: 1
  initialState: watch

  context:
    metric:    { type: string }                          # the observed event
    status:    { type: enum, values: [breach, settled] } # the routing trigger, consumed each cycle
    reactions: { type: map,  of: string }                # each reactor's action, keyed by author

  agents:
    watcher: { canTransition: [watch] }

  states:
    watch:
      description: >-
        Report the current state of the metric stream. Stage the observation -
        updateContext(field=metric, value=<name=value, threshold, direction>) - and then classify it:
        updateContext(field=status, value=breach) or value=settled.
      requires:
        - { name: metric, from: { agent: watcher } }
        - { name: status, from: { agent: watcher } }
      transitions:
        - toState: react
          guards: [ { backend: cel, expression: "context.status == 'breach'" } ]
          onTransition:
            - { emit: "metric.breach" }   # declared publish - see the note below on what this does today
        - toState: done
          guards: [ { backend: cel, expression: "context.status == 'settled'" } ]

    react:
      description: >-
        A metric breach: "{{ context.metric }}". Take your part of the response and record it:
        updateContext(field=reactions, value=<what you did>).
      requires:
        # All three summoned at once - subscribers to the same event, not a relay chain.
        - { name: reactions, from: { agent: alerter },  prompt: "Raise the alert. Record severity." }
        - { name: reactions, from: { agent: scaler },   prompt: "Scale the affected service. Record the action." }
        - { name: reactions, from: { agent: notifier }, prompt: "Page the on-call. Record who, with what." }
      exit:
        - { emit: "breach.handled" }
      transitions:
        - toState: watch
          onTransition:
            # Consume the event so the next cycle is a fresh one. The reactions map has to be
            # cleared too - an entry that is still there satisfies its requirement, and a cycle
            # that never asks anyone is a cycle that silently re-uses the last response.
            - { assign: { metric: "", status: "" } }
            - { remove: { reactions: "*" } }

    done:
      description: "Metric settled."
      terminal: true

  policy:
    maxIterations: 20            # bounds the watch/react cycle (headless)
    maxExchangesPerStage: 30
    maxExchangesPerAgentPerStage: 15
    checkpointEvery: 5

Two details in there are worth pointing at before the walk-through, because both are places an earlier draft of this article was wrong.

emit sits in onTransition and exit, not in an entry block on a transition - a transition takes onTransition; entry and exit are state-level hooks. And status is an enum rather than a settled: bool, because a boolean’s presence is what a requirement checks, and false is present. settled = false would have satisfied the requirement and fired the done edge. One enum carrying both outcomes has no such gap, and it’s the field the loop-back consumes.

Event-Driven / Publish-Subscribe work flow:

status == breach (emit metric.breach)

status == settled

all three reactors responded, event consumed

watch

react

done

Three things to hold onto - two claims and one caveat:

  • The reducer is the internal event bus, and that part is entirely real. The watch state parks waiting on the watcher. When the watcher classifies the observation, the reducer re-runs, the breach guard passes, and the edge fires. The reactors are summoned by the machine - they don’t poll and they don’t subscribe in code. Their subscription is the requirement that names them; their reaction is the context write that satisfies it. The workflow emerges from guards and requirements, exactly as the pattern describes, and the emergence is controlled by a declared topology.
  • The reactors are subscribers to one event, not a relay chain. All three requirements name the same reactions field with three different sources, so all three are summoned concurrently and each entry is keyed by whoever wrote it. Nothing routes the alerter’s output to the scaler; they’re independent reactions to the same breach, which is what the pattern actually asks for. An earlier version of this article chained them state-to-state, which reads as pub-sub but is really a pipeline wearing its clothes.
  • emit does not currently reach anything. It’s real DSL syntax, it parses, the reducer executes it - and what it does is write a log line. There is no bridge from an emit action to the application bus, on the interactive runtime or on the headless targets. So metric.breach is not something a dashboard can subscribe to today; it’s a declaration of the publish contract with the transport unbuilt. Wiring it is the cheapest of the gaps here - the action exists and a bus exists, so it’s a middleware, not a design - but until that lands, treat every emit in this article as documentation of intent rather than an integration point.

And the mirror of that caveat, on the way in: the watcher is an agent, not a stream. Nothing taps a metrics pipeline. watch parks and asks an LLM what the metric is doing, which is fine for illustrating the topology and is not monitoring. A real ingress leg means an external adapter turning stream events into context writes on a specific instance - the same middleware shape as the egress leg, mirrored, plus the instance-routing question. Neither leg exists yet, and this is the one pattern in the series where that absence is the pattern itself rather than a detail at the edge of it.

Walking one breach-and-settle

The watcher reports: metric = "p99_latency=420ms, threshold=300ms, direction=up", status = breach. The reducer re-runs, the breach guard passes, the edge fires, and emit: "metric.breach" runs - logging, today, what it will one day publish.

The machine enters react and summons all three reactors at once. The alerter records a SEV2 under reactions['agent:alerter']; the scaler records “scaled web tier 4→8” under reactions['agent:scaler']; the notifier records the page under reactions['agent:notifier']. None of them waited on the others, and none of them read the others’ output - they reacted to the same breach independently, which is what makes this pub-sub rather than a pipeline. When all three keys exist, the state’s requirements are satisfied; exit fires breach.handled, the loop-back edge clears the event and the reactions map, and the machine re-parks at watch for the next cycle. The metric returns to normal; the watcher stages status = settled, and the done edge fires.

notifierscaleralerterwatcherMachine (reducer)notifierscaleralerterwatcherMachine (reducer)emit metric.breach(logged, notpublished)all three keys present → exit emits breach.handledevent + reactions cleared → watchmetric + status=breachA2A summon (react)A2A summon (react)A2A summon (react)reactions['agent:alerter']reactions['agent:scaler']reactions['agent:notifier']metric + status=settleddone (terminal)

What the reducer does to the failure points

Event storms. An event storm is a reaction that triggers more reactions faster than the system can settle. The reducer settles - it runs to a stable configuration after every event, and the microstep cap (128) bounds the reactions per event. A storm of external events (the bus floods the watcher) is bounded by policy.maxIterations on the watch/react cycle and by the watcher’s own rate (it stages one metric event per turn). The system can’t storm internally - the loop is bounded - and external storms are a watcher-throughput problem you can see in the parked record, not a silent cascade.

Duplicate processing. The consume-the-event discipline is the dedup. The react → watch edge clears metric and status and empties the reactions map, so the next breach is a fresh event and the machine doesn’t re-react to the old one. Both halves of that matter: clearing the scalars re-arms the routing, and clearing the map re-arms the asks - a map entry that survives the loop-back satisfies its requirement, so the next cycle would advance without summoning anyone and quietly re-use the previous response as if it were new. That’s the characteristic failure of accumulating state in a loop, and it’s why nothing that gates a cycle should survive the cycle.

Because a write re-runs the reducer to a stable configuration before the next write is applied, there’s also no window where the same event is processed twice. The no-progress guard is the backstop within a single advance: a re-entry with unchanged context stops.

Race conditions. The reducer serializes them. Every event is a discrete context write, and the machine processes writes one at a time - a write merges, the reducer re-runs to stable, the next write applies on top. The reactive agents run concurrently, but their writes are ordered by the machine. There’s no lost-update, no read-modify-write race, because the store has a single writer at a time. “Race conditions” is a concurrency problem; here it’s a queue, the same answer as the blackboard article.

Unclear causality. Causality is structural. Every reaction is a fired edge with a named trigger - the context write that made its guard pass - and every emit is a declared action on a specific edge or hook. To answer “why did the scaler scale?” you read the edge that fired (watch → react), the context that triggered it (status = breach), and the entry the scaler wrote under its own key. The causal chain is the edge sequence plus the writer-keyed map - journaled, replayable, not reconstructed from a bus trace. Attribution here is not a convention the reactors follow; the runtime keys each entry by its actual writer.

Difficult global reasoning. The global state is the store, and it’s typed and inspectable. At any moment, the machine is at one state, the store holds the current event and the reactions so far, and the parked record shows what’s pending. You don’t have to correlate a distributed trace to understand the global state - you read the store. The “global reasoning” the pattern finds hard is exactly what the reactive reducer makes easy: the global state is a single, typed, journaled record.

Accidental feedback loops. A feedback loop is a reaction that triggers its own trigger. The reducer makes loops declared, not accidental. If a reaction is supposed to feed back into the watcher (the react → watch edge), it’s an edge you drew, with a consume-the-event onTransition and a maxIterations bound. An undeclared feedback loop can’t happen, because the reducer only fires declared edges - a reaction that isn’t an edge simply doesn’t exist. The loop is either the one you designed or it doesn’t happen. It’s also worth noticing that the absent external legs are, right now, an accidental safety property: an emit that reached a bus that fed a watcher would be exactly the cross-process cycle nothing in the machine can see.

Where this leaves the pattern, honestly. Internally, the reducer is a controlled event system and every claim above holds. Externally, it is not yet an event system at all: no ingress adapter, and emit logging rather than publishing. Composing many reducers into a mesh - each with its own store, reacting to each other’s events across process boundaries - is the right shape and the headless targets are the right substrate for it, but the connective tissue is unbuilt in both directions. Of the thirteen patterns in this series, this is the one where the reducer’s fit is a genuine claim about the inside and an open item about the outside, and I’d rather it be marked that way than counted as a clean fit on the strength of syntax that parses.

The whole thing, one idea

MomentWhoWhat happened in context
watchwatchermetric + status = breach set → emit metric.breach (logged)
reactalerter, scaler, notifier (concurrent)reactions keyed by all three authors
react → watch-exit emits breach.handled; event + reactions cleared
watchwatchermetric + status = settled
done-terminal

No message broker bolted onto a pile of agents. No loose event mesh. The reducer was the bus; the requirements were the subscriptions; the writer-keyed map was the attribution; the consume-the-event discipline was the dedup; and the loop was bounded by the microstep cap and maxIterations. The workflow emerged from the guards, and the machine was the only thing that moved it.

All of which is a claim about the inside of the box. The emits in that table went to a log, and the watcher was an agent rather than a stream. The internal event system is the part this pattern already has; the transport at both edges is the part it doesn’t.

Up next: 12. Quorum / Voting