KDCube
← Our Journal
2026-07-13 · Journal

Connect Your Agentic Loop To The Event Bus

event bus ordered delivery reactive turn run-to-completion followup / steer one event, one turn

Bring your own agent — your graph, your framework, your control loop. KDCube hands it user messages through one door, off the event bus: ordered, one turn at a time per conversation, exactly once, nothing lost. A run-to-completion loop gets that for free; a loop that can absorb input mid-flight opts into more.

You have a working agentic loop. On KDCube it becomes an app, and users talk to it. So the first question is not "how do I run my graph" — it is "how does a user's message reach my loop, and what happens when a second message arrives while the first is still running?" The answer is the same for every agent on KDCube, whether it is the built-in ReAct engine or a ported LangGraph graph: the event bus delivers messages, and one door turns each delivered message into one turn.

ONE MESSAGE, ONE DOOR, ONE TURN User message prompt / followup / steer one message in Event-bus lane atomic appended to the ordered log + one wakeup run() shared door @on_reactive_event every agent shares it execute_core your loop runs the turn one turn append fires calls chat renders live streams tokens + steps back through the communicator
A user message becomes a lane record plus one wakeup; the shared door run() turns it into one turn of your loop, streamed back so chat renders live.

One Door For Every Turn

Nothing runs your agent except a wakeup. A user message (a prompt, a followup, or a steer) is appended to the conversation's event-bus lane — an ordered log — and, atomically, one wakeup is enqueued. When the processor claims that wakeup it opens the one door every agent shares:

external event (prompt / followup / steer)
   -> appended to the conversation event-bus lane (ordered log)
   -> one wakeup enqueued                                  [atomic]
        -> run()   (@on_reactive_event — the shared door on the app base)
             -> execute_core(state, thread_id, params)     your loop runs the turn

run() is the reactive-event door. Your execute_core reads the triggering message out of state and runs your loop to completion, streaming through the communicator so the reusable chat component renders it live. That is the whole integration surface: implement execute_core, and the event bus does the rest.

Ordered, One At A Time

Turns of a single conversation are serialized by a per-conversation lock. A second message that arrives while a turn is running does not start a second execute_core — it waits its turn, in order:

Event 1 -> wakeup -> conversation lock acquired -> run()/execute_core  (turn 1 running)
Event 2 arrives now -> wakeup enqueued -> cannot acquire the lock -> requeues (waits)
turn 1 ends -> lock released -> Event 2's wakeup claimed -> run()/execute_core  (turn 2)
  • Same conversation: one turn at a time, in arrival order. The lock holds across processor workers, so two workers cannot run two turns of one conversation at once.
  • Different conversations: run in parallel, on their own locks.

You do not implement any of this. What a turn is responsible for is releasing the piece of the event bus it was handed — so the message waiting behind it can run. That piece is the lane reservation.

The Lane Reservation

When the processor dispatches a wakeup it reserves the event-bus lane consumer for that conversation before the turn runs. The reservation is how the platform knows a turn is responsible for this lane. Whoever holds it must release it when the turn ends. Leave it held, and the next turn's wakeup — claimed inside the reservation's freshness window — is refused and silently dropped: the next turn never runs.

THE LANE RESERVATION LIFECYCLE Wakeup dispatched scheduled reserves the lane consumer before the turn runs Turn runs run() / execute_core serialized, one at a time Turn end none releases the reservation back to none Next wakeup claimed the queued message runs next turn, on time releases Reservation left held scheduled the turn did not release stays scheduled Next wakeup refused scheduled_consumer_fresh dropped · next turn never runs leaves it held Release the reservation at turn end and the next wakeup is claimed; leave it held and the next wakeup drops as scheduled_consumer_fresh.
The wakeup reserves the lane consumer before the turn and the turn end releases it back to none; a turn that leaves it scheduled makes the next wakeup drop as scheduled_consumer_fresh.

Releasing the reservation is the one place the two kinds of agent differ.

Two Ways To Consume

TWO WAYS TO CONSUME THE LANE ReAct — folds mid-turn Run-to-completion — one event, one turn Opens the lane handler reads the event bus while it runs a live consumer Folds the followup a mid-turn followup joins the running turn at a decision boundary Close gate releases the reservation and re-wakes anything still unconsumed Takes one event consumes the triggering message only the external-event batch on state Never watches the lane runs start to finish a mid-turn followup is not folded Shared door finalizes releases the reservation for it and re-wakes the queued followup — the next turn mid-turn at close after turn ReAct owns its lane lifecycle. One event, one turn — the next message promoted at turn end.
ReAct folds a mid-turn followup inside its own turn; a run-to-completion loop takes one event and lets the shared door promote the next.

ReAct — a live consumer that folds mid-turn. A ReAct turn opens the lane handler and reads the event bus while it runs. A followup that lands mid-turn is folded into the running turn at a decision boundary. At the close gate the turn releases the reservation and re-wakes anything still unconsumed. ReAct owns its lane lifecycle inside its own workflow.

Run-to-completion — one event, one turn. A ported graph or a bespoke loop runs start to finish without watching the lane. It consumes exactly the triggering message; a followup that lands mid-turn is not folded. Because it never opens the handler, the shared door releases the reservation for it after the turn — and re-wakes any message that queued during the turn, so that message becomes the next turn. The net contract for your loop: one event, one turn, strictly ordered, with the next message promoted the instant the current turn ends.

What You Write

For a run-to-completion loop, the integration is one method and one declaration:

class MyAppEntrypoint(BaseEntrypointWithEconomics):
    async def execute_core(self, *, state, thread_id, params):
        # the triggering event(s) arrive as the external-event batch on state
        question = external_events_text(state.get("external_events") or [])
        # ... run YOUR loop / graph to completion ...
        # stream tokens + steps through the current communicator (comm_ctx)
# per agent, in the app descriptor
conversation:
  accepts_followup: false     # this loop does not fold a new message mid-turn
  accepts_steer:    false     # this loop does not cancel + finalize mid-turn

That is the entire wiring. The door serializes turns, releases the lane reservation for you, and promotes queued messages. You never touch the event bus.

Followup Without Folding

accepts_followup / accepts_steer change what the composer offers, not what is delivered. For a run-to-completion loop (both false), a message sent mid-turn is queued and promoted to the next turn — the composer shows "Queue for next turn," and the door's finalize re-wakes it after the current turn ends. Nothing is lost and nothing is folded. An agent that genuinely can absorb input mid-flight declares true and owns the lane handler the way ReAct does — declare what you actually implement, not what you wish were true.

The Bug This Closes

Before the door finalized the lane for run-to-completion turns, such a turn left its reservation held. The next turn's wakeup, arriving inside the freshness window, was dropped as scheduled_consumer_fresh: the turn "completed" in the UI, but the next message never reached execute_core — nothing in the processor log — and only recovered after the window went stale. It read as an intermittent "second or third turn hangs." The finalize releases the reservation at turn end, as a state-conditional invariant owned by the shared door (release it if still held; a ReAct turn already released, so this is inert for it — decided by lane state, never by agent type). One event, one turn, next turn on time.

Read More