> For the complete documentation index, see [llms.txt](https://fleet.hackmap.win/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://fleet.hackmap.win/concepts/streams-and-events.md).

# Streams and the event bus

Two primitives for moving data around. Streams are for work product. The event bus is for signals. Mixing them up tends to cause grief.

For shared state that doesn't fit either — a pool of cookies, a global counter, a lock — see [primitives](/concepts/primitives.md).

## Streams

One per automation type. Backed by a Redis ZSET at `stream:<type>`. Each entry is a JSON envelope: id, slot id, timestamp, payload.

Producers declare their output type on the automation class:

```python
class ScrapeResult(BaseModel):
    url: str
    status: int

@register("my-scraper")
class MyScraper(ContinuousAutomation[MyConfig]):
    Output = ScrapeResult

    async def run_slot(self, ctx):
        await ctx.emit(ScrapeResult(url="https://...", status=200))
```

`ctx.emit(...)` accepts only instances of the declared `Output` type and raises `TypeError` otherwise.

Consumers reach in via a typed `Stream[P]` reference (typically declared once in a shared contracts package):

```python
from cf_contracts import SCRAPER_OUT  # Stream[ScrapeResult]("cf-scraper", ...)

async def run_slot(self, ctx):
    items = await ctx.stream(SCRAPER_OUT).pop(n=10)
    # items is list[ScrapeResult] — fully typed
```

Or with the REST API (admin token; useful for external clients that don't run as automations):

```bash
curl -X POST http://master/api/v1/automations/my-scraper/output/pop \
  -H "Authorization: Bearer $ADMIN_TOKEN" \
  -d '{"n": 10}'
```

The REST form returns envelopes:

```json
[{"id": "abc...", "slot_id": 0, "ts": 1778..., "payload": {...}}]
```

`pop` is destructive — popped entries are gone. If you need to inspect without consuming, use `peek`. If you need a count, use `length`.

## Bounding

Streams are bounded by two parameters: max length (default 50,000) and TTL (default 3600 seconds). On every push the framework prunes anything older than the TTL and trims to the length cap.

A producer can run forever even if no consumer ever pops. Memory stays bounded. If a consumer eventually attaches, they get the most recent entries up to the cap — older entries are gone.

There's no per-message ACK or retry. Pop succeeds atomically. The framework doesn't model "process this message at-least-once" — if your handler crashes after popping, the message is lost. Build idempotency into your pipeline if you need it.

## Inter-automation flow

The pattern that motivated this whole project: `turnstile-solver` is a continuous automation that emits tokens into `stream:turnstile-solver`. `account-creator` is a batch automation that calls `ctx.stream(TURNSTILE_OUT).pop()` whenever it needs to bypass a captcha, then emits the resulting account into its own stream. They run in separate worker processes, possibly on different machines. The only thing they share is the master's Redis and the shared contracts package that declares the `Stream[TurnstileToken]` reference.

Neither knows the other's runtime code. If you swap the captcha solver out for a different one with the same `Output` shape, account-creator doesn't need to change.

If a stream isn't the right shape — say, you want non-destructive reads, claims with TTL, or multiple consumers sharing a small inventory of items — that's a [pool](/concepts/primitives.md), not a stream.

## The event bus

For signals, not work product. Backed by Redis pubsub at `event:<topic>`.

```python
# publish
await ctx.events.publish("slot.started", {"slot_id": ctx.slot_id})

# subscribe (in another automation, or in the same one)
async for ev in ctx.events.subscribe("slot.started"):
    print(ev)
```

Three differences from streams:

No persistence. If you weren't subscribed when an event fired, you missed it.

No back-pressure. The publisher doesn't wait, doesn't know who's listening.

No throughput cap. Pubsub will deliver as fast as Redis can broadcast.

Use the event bus for things you don't mind losing: "worker came online", "slot crashed", "config was patched". Use streams for things you do mind losing: any unit of work produced.

A common mistake: trying to use the event bus to ship work because it feels faster. It does feel faster (no ZSET writes, no TTL pruning), but the first time a consumer is briefly disconnected, you'll lose data and have no way to know.

## When neither is enough

You want exactly-once semantics. The framework doesn't have those. Use streams plus an external dedup layer (a Redis SET of seen IDs, a database constraint on the consumer side).

You want bidirectional communication between automations. Use a request/response pattern over a third primitive (HTTP, a database, your own message broker) and treat the automation as one side of an external integration.

You want N consumers to share a small inventory of items (cookies, accounts, fingerprints) where each consumer borrows rather than consumes. That's a [pool](/concepts/primitives.md). Streams are wrong for it because `pop` is destructive.


---

# Agent Instructions
This documentation is published with GitBook. GitBook is the documentation platform designed so that both humans and AI agents can read, navigate, and reason over technical content effectively. Learn more at gitbook.com.

## Querying This Documentation
If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question.

Perform an HTTP GET request on the current page URL with the `ask` query parameter, and the optional `goal` query parameter:

```
GET https://fleet.hackmap.win/concepts/streams-and-events.md?ask=<question>&goal=<endgoal>
```

`ask` is the immediate question: it should be specific, self-contained, and written in natural language.
`goal` is optional and describes the broader end goal you are ultimately trying to accomplish on behalf of the user. GitBook uses it to tailor the answer towards what is most useful for that goal.

The response will contain a direct answer to the question and relevant excerpts and sources from the documentation.

Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.