Watch the Event Stream#

Invoking gets you one result. Watching gets you the ongoing facts: the events the work emits while it happens. This is the observe half of the agent loop, and it is what lets you recover from a partial failure instead of trusting a single return value — see Submitted Is Not Completed.

Subscriptions are hot#

You see events emitted after you subscribe, plus whatever is still in the ring buffer. Not the whole history.

There is no replay-from-the-beginning on the bus, and that is a design decision rather than a missing feature. Durability is a separate layer: a RedEX log or a retaining adapter, covered in Durable Logs. A consumer that needs to see everything since a known point needs one of those, not a longer timeout.

Two consumption models, split by binding#

This is the sharpest asymmetry on the whole spine, and it is not cosmetic:

• Rust, TypeScript and Python give you a stream. You subscribe and iterate; the runtime delivers.
• Go and C give you a cursor. You call Poll(limit, cursor), get a batch and a next cursor, and call again.

Neither is a wrapper over the other at the API level, and code does not port between them by changing syntax. A Go consumer is a loop you drive; a Python consumer is a loop that drives you. Plan the shape of your consumer around the binding you are on rather than around the one the example was written in.

The bus is location-transparent#

The same consumption code works whether the publisher is in-process or several hops away on the mesh. Between mesh nodes a subscriber joins a named channel by the publisher's node id and the publisher fans out to its roster; the reading side does not change.

What does change is what "still in the buffer" means. In-process, the ring buffer is the whole story. Across the mesh, an event has to arrive before it can be buffered, so a consumer that starts late misses more than it would locally.

Watching capabilities is a different thing#

This page is about the event stream — the data flowing through a node. Watching the set of available tools change is on Discover, and it uses a different surface with a different cadence model. The two are easy to confuse by name and share nothing in implementation.

The concepts are in Channels and Events and Causality.

Watch it — Go#

Go is cursor-based. There is no subscribe iterator; you poll and page forward.

cursor := ""
for {
    resp, err := bus.Poll(100, cursor)
    if err != nil {
        log.Fatal(err)
    }
    for _, ev := range resp.Events {
        var reading struct {
            SensorID string  `json:"sensor_id"`
            Celsius  float64 `json:"celsius"`
        }
        if err := json.Unmarshal(ev, &reading); err == nil && reading.Celsius > 80 {
            fmt.Printf("HOT: %s at %.1fC\n", reading.SensorID, reading.Celsius)
        }
    }
    if resp.NextID == "" {
        break        // caught up
    }
    cursor = resp.NextID
}

Poll(limit, cursor) returns a *PollResponse with Events []json.RawMessage and a NextID. An empty string cursor starts from the earliest buffered event; an empty NextID means you have caught up to the tail.

Events arrive as json.RawMessage, so you unmarshal each one yourself. There is no typed subscribe to do it for you.

For a live loop#

for {
    resp, err := bus.Poll(100, cursor)
    if err != nil {
        log.Fatal(err)
    }
    // ... handle resp.Events ...
    if resp.NextID != "" {
        cursor = resp.NextID
    }
    time.Sleep(200 * time.Millisecond)   // your cadence, your choice
}

Keep the cursor when NextID comes back empty. Overwriting it with the empty string restarts from the earliest buffered event and replays everything you have already handled. That is the one bug this loop shape reliably produces.

The polling cadence is yours to pick. Nothing in the binding chooses it for you, and there is no push path to fall back to — this is the binding asymmetry described above, not a gap waiting to be filled.

Verify it worked#

stats, err := bus.Stats()
if err != nil {
    log.Fatal(err)
}
fmt.Printf("consumed against %d ingested\n", stats.EventsIngested)
if stats.EventsIngested == 0 {
    log.Fatal("nothing was ever accepted to watch")
}

If Poll returns nothing, check the transport before the loop: memory counts events and discards them. See Quickstart.

Next: Move artifacts.