The Event Pipeline

The fundamentals course introduced the Event Bus as the glue between CAO's components. Now let's look at what actually flows through it — and the five specialized services that make message delivery feel instant.

Think of it as a newsroom. Reporters (FIFOs) feed raw stories into a wire service (EventBus), editors (StatusMonitor) decide what is breaking news, and the dispatch desk (InboxService) routes the right story to the right person at the right time.

F

FifoManager (Reporters)

Reads raw bytes from named FIFOs at ~/.aws/cli-agent-orchestrator/fifos/{id}.fifo. Coalesces rapid-fire chunks within a 50ms window before publishing.

B

EventBus (Wire Service)

Async pub/sub backbone. Wildcard topic matching (terminal.*.output), bounded queues (1024 default), drop-and-log on overflow.

S

StatusMonitor (Editors)

Maintains an 8KB rolling buffer per terminal. Runs provider-specific pattern matching to detect IDLE, PROCESSING, COMPLETED, WAITING_USER_ANSWER, or ERROR. Uses debounced edge detection with a sticky latch.

I

InboxService (Dispatch Desk)

Routes the right message at the right time. When a terminal reaches IDLE or COMPLETED, delivers one pending message from its inbox.

L

LogWriter (Archivist)

Subscribes to terminal.*.output alongside everyone else. Batches writes to per-terminal log files so you always have a full record.

Watch a Chunk Travel

Follow a single piece of terminal output as it flows through every stage of the pipeline. Notice how the EventBus appears twice -- once routing raw output, once routing the derived status change. This two-phase publish is the key to decoupling detection from delivery.

P

pipe-pane

F

FifoManager

B

EventBus

S

StatusMonitor

I

InboxService

L

LogWriter

Click "Next Step" to begin

Step 0 / 7

💡

Key Insight: Two-Phase Publish

The EventBus carries two kinds of topics: terminal.{id}.output for raw data, and terminal.{id}.status for derived state changes. StatusMonitor subscribes to the first and publishes the second. InboxService only subscribes to the second -- it never sees raw output.

The Sticky Latch

StatusMonitor uses debounced edge detection -- it does not re-evaluate status on every single chunk. Instead it waits for output to settle, then checks. And once it determines a terminal is IDLE, it applies a "sticky latch":

🔒

Why a Latch?

Once an agent reaches IDLE, the StatusMonitor latches that status. It will not drop to PROCESSING just because the TUI redraws a progress bar or refreshes the screen. Any new input sent to the terminal unlocks the latch and allows a transition back to PROCESSING — that includes handoff, assign, inbox delivery, and cao session send.

⚠️

Without the Latch

A TUI redraw would momentarily look like new output, causing StatusMonitor to flip the terminal to PROCESSING. InboxService would then hold messages hostage until the next IDLE detection until the next genuine idle detection.

Here is the core loop, simplified. The real code adds debouncing, buffer rotation, and the latch logic, but the publish/subscribe pattern remains the same:

Python

# services/status_monitor.py (simplified)
queue = bus.subscribe("terminal.*.output")  # wildcard!
async for topic, data in queue:
    terminal_id = topic.split(".")[1]
    buffer[terminal_id] += data
    status = provider.get_status(buffer[terminal_id])
    if status != last_status[terminal_id]:
        bus.publish(f"terminal.{terminal_id}.status", {"status": status})

Plain English

Subscribe to ALL terminal output using a wildcard pattern.

When new data arrives, extract the terminal ID from the topic.

Append the data to that terminal's rolling buffer.

Ask the provider-specific detector what state the terminal is in.

If the state changed from last time, announce the new status on the bus.