State Is Not an Event: Designing Real-Time Agent UIs

The previous article, Where Does Agent State Live?, placed the durable run record and restart-authoritative snapshots in PostgreSQL, a UI and ops hot projection in a Redis Hash, and the resumable graph snapshot in a checkpoint. Hash fields do not replace PostgreSQL rows.

Suppose a user opens an agent run screen. The browser must answer three questions:

• What is the system doing now, and how far has it progressed?
• Which user-meaningful changes led here?
• If the connection returns after a break, where should updates resume?

A snapshot, an event timeline, and a delivery cursor answer those questions.

Snapshot
  Current status: Applying rules, 65%
  Current work: Reviewing rule 3 of 4

Timeline
  ✓ Documents collected
  ✓ Inputs normalized
  → Applying rules
  ! Approval required

Cursor
  last_event_id = 1760000000123-0

app.example.com/runs/7d5

Document review runTravel-expense exception review

●Live connection

Current statusApplying rules

65%updated just now

Continuous changes such as 61% → 62% update this card without creating timeline entries.

Current work · Snapshot

Reviewing rule 3 of 4

Scope

APAC / Korea

As of

2026-10-05

Next action

Wait for reviewer approval

Activity · Event timeline

1. 09:41
   Documents collected12 documents · current versions checked
2. 09:42
   Inputs normalizedamounts · dates · organizational scope
3. 09:43
   Applying rules3 of 4 rules checked
4. Waiting
   Approval requiredA reviewer must confirm the exception

Reconnect state is part of the UI contractReconnecting → cursor replay → snapshot resync when history expired

Continuously overwriting Hash fields preserves only the latest values. Keeping only events creates the opposite problem: every screen load must replay the entire history to calculate current progress. Storing both is not accidental duplication. It is an intentional projection across current state, event history, and delivery stream.

A Redis Stream is a delivery stream, not a long-term audit ledger. It fits real-time fan-out, reconnect cursors, and a bounded replay window. Do not treat it as the sole store for events that legal, security, or compliance must read years later. Stream trim, TTL, and memory pressure are operational policy. History needed for audit, reproduction, or dispute response belongs in a durable tier—PostgreSQL tables, append-only log pipelines, or object storage. Trimming the stream must not erase run results and checkpoints that remain in PostgreSQL.

The browser reads a snapshot and a timeline together

Dimension	State	Event
Question	What is the status now?	What happened?
Write pattern	Overwrite the same field	Append a new entry
Read pattern	Read the latest snapshot once	Read a range after a cursor
Typical structure	Redis Hash	Redis Stream
Example	progress=65	rules_applied
Retention	May use a short TTL	Retain for the replay window
UI role	Restore the screen immediately	Timeline and incremental updates

The essential difference is time.

State
time ──────────────────────────────────────────────>
                   { progress: 65, step: apply_rules }
                   one point: now

Events
time ──────────────────────────────────────────────>
     documents_collected ─ inputs_normalized ─ rules_applied ─ ...
                   order of change

Progress 65 is not an event. Once it becomes 70, the old value is usually irrelevant. approval_required, by contrast, is a domain event with meaning for user action and audit. If event types proliferate into progress_changed_to_61, progress_changed_to_62, and so on, the timeline becomes noise.

A useful starting rule is:

• Put frequently changing numeric progress and the current step in state.
• Record user-legible phase transitions and important domain facts as events.
• Do not create an event per token. Treat token streaming as a separate, high-volume channel.

Browser regionReadsRenders

• Status headersnapshotstatus · progress · current_step
• Current worksnapshotactive item · approval state
• Timelinerecent eventsphase changes · approval · recovery · completion
• Live updatescursor + SSEevents after last_event_id

The snapshot restores the screen, events extend the timeline, and the cursor fixes the boundary between them.

Store the snapshot and timeline in different representations

Following the key naming from Article 6, one run gets two keys:

agent:run:{run_id}:state    Redis Hash
agent:run:{run_id}:events   Redis Stream

The run_id inside braces is a Redis Cluster hash tag. Placing both keys in the same slot allows a transaction or Lua script to operate on them together.

The state Hash is small and current.

HSET agent:run:{7d5}:state
  status          running
  current_step    apply_rules
  progress        65
  last_event_id   1760000000123-0
  updated_at      2026-10-04T05:10:00Z

The event Stream is an append-only timeline.

1760000000010-0  documents_collected
1760000000060-0  inputs_normalized
1760000000123-0  rules_started

A Redis Stream entry contains an ID and field-value pairs. With *, XADD asks Redis to generate a monotonically increasing ID. It commonly looks like <millisecondsTime>-<sequenceNumber>, but applications should treat it as an opaque cursor, not derive business time from it. (XADD)

Fix the event schema first

Redis Streams do not enforce a schema, so the application needs a stable envelope.

from dataclasses import dataclass, field
from datetime import UTC, datetime
from typing import Any

@dataclass(frozen=True)
class RunEvent:
    type: str
    run_id: str
    attempt_id: str
    fencing_token: int
    payload: dict[str, Any] = field(default_factory=dict)
    schema_version: int = 1
    occurred_at: str = field(
        default_factory=lambda: datetime.now(UTC).isoformat()
    )

Use stable, machine-readable names for event types, not UI sentences.

Good
documents_collected
rules_applied
approval_required
run_succeeded

Avoid
All documents have been loaded!
The review is almost done

The client derives display copy and localization from the event type and payload. Copy can then change without breaking the historical event schema.

Keep occurred_at distinct from the Stream ID:

• The Stream ID provides ordering and a cursor inside Redis.
• occurred_at is the domain timestamp written by the producer.
• Do not assume that clocks across producers agree perfectly.
• Within one stream, prefer Stream ID order for the UI.

Build bounded event history with XADD

Plain async redis-py code can begin like this:

import json
from redis.asyncio import Redis

MAX_EVENTS_PER_RUN = 1_000

def run_events_key(run_id: str) -> str:
    return f"agent:run:{{{run_id}}}:events"

async def append_run_event(
    redis: Redis,
    event: RunEvent,
) -> str:
    event_id = await redis.xadd(
        run_events_key(event.run_id),
        fields={
            "type": event.type,
            "attempt_id": event.attempt_id,
            "fencing_token": event.fencing_token,
            "schema_version": event.schema_version,
            "occurred_at": event.occurred_at,
            "payload_json": json.dumps(event.payload),
        },
        maxlen=MAX_EVENTS_PER_RUN,
        approximate=True,
    )
    return event_id

This standalone XADD helper is a minimal illustration for an event that does not change the snapshot. Do not call it directly from competing worker attempts without atomically validating the current fencing token. Real transitions that change both snapshot and timeline use the fenced publish_transition below.

This also requests approximate trimming, equivalent to MAXLEN ~ 1000. Exact trimming keeps the threshold precisely but can require more work. Approximate trimming removes entries efficiently in units of Redis’s internal structure, so the actual length may temporarily exceed the threshold. (XTRIM)

If retention must be time-based rather than count-based, consider a MINID strategy. Either way, permanent retention of every event should not be the default. A Redis Stream is an operational log for the UI replay window. Put long-term audit records in PostgreSQL or object storage.

One transition changes the snapshot and timeline atomically

Two separate commands leave room for an intermediate failure.

await redis.hset(state_key, mapping={"progress": 65})
await redis.xadd(events_key, {"type": "rules_started"})

If the worker dies after the first command, state says 65% but has no corresponding event. If the event is written first and the state update fails, the timeline says rule application started while the snapshot remains on the previous step.

Not every state update needs an event. When one domain transition changes both snapshot and timeline, however, they should share an atomic operation.

A Lua script is convenient because the generated Stream ID must also become last_event_id in state. The script can receive the XADD result and immediately pass it to HSET.

PUBLISH_TRANSITION = """
local current_fence = tonumber(redis.call('HGET', KEYS[1], 'fencing_token'))
local proposed_fence = tonumber(ARGV[4])

if current_fence and proposed_fence < current_fence then
    return redis.error_reply('stale fencing token')
end

local event_id = redis.call(
    'XADD', KEYS[2], 'MAXLEN', '~', ARGV[1], '*',
    'type', ARGV[2],
    'attempt_id', ARGV[3],
    'fencing_token', ARGV[4],
    'schema_version', ARGV[5],
    'occurred_at', ARGV[6],
    'payload_json', ARGV[7]
)

redis.call(
    'HSET', KEYS[1],
    'attempt_id', ARGV[3],
    'fencing_token', ARGV[4],
    'status', ARGV[8],
    'current_step', ARGV[9],
    'progress', ARGV[10],
    'last_event_id', event_id,
    'updated_at', ARGV[6]
)

redis.call('EXPIRE', KEYS[1], ARGV[11])
redis.call('EXPIRE', KEYS[2], ARGV[11])
return event_id
"""

async def publish_transition(
    redis: Redis,
    *,
    event: RunEvent,
    status: str,
    current_step: str,
    progress: int,
    ttl_seconds: int,
) -> str:
    result = await redis.eval(
        PUBLISH_TRANSITION,
        2,
        f"agent:run:{{{event.run_id}}}:state",
        run_events_key(event.run_id),
        MAX_EVENTS_PER_RUN,
        event.type,
        event.attempt_id,
        event.fencing_token,
        event.schema_version,
        event.occurred_at,
        json.dumps(event.payload),
        status,
        current_step,
        progress,
        ttl_seconds,
    )
    return result

The script rejects a stale worker token before changing the event and snapshot together. If updates within one attempt can also race, add the version comparison from the previous article. This atomicity provides local Redis consistency, not a distributed transaction with the PostgreSQL run record. Commit a terminal result to PostgreSQL first, then reconcile the Redis projection and terminal event. A failed Redis update must remain repairable from durable state.

last_event_id is the snapshot boundary

A naive real-time connection has this race:

1. Client reads the snapshot with HGETALL
2. Worker publishes event E
3. Client waits only for new events with XREAD '$'

Result: E occurred after the snapshot but before subscription, so it is missed

$ means “start at the end of the stream when XREAD is called and wait for later entries.” It skips events created between the snapshot read and that call.

If state and events update atomically and state includes last_event_id, connect in this order:

1. HGETALL state
   → snapshot + last_event_id = 1760000000123-0

2. If needed, use XRANGE to read recent history through last_event_id

3. XREAD from 1760000000123-0
   → replay events after that ID, then keep waiting

An event published after the snapshot has an ID greater than its cursor, so XREAD sees it. Events before the snapshot are already reflected in state; use XRANGE when the timeline also needs them.

Bootstrap returns the whole screen-restoration contract

When the UI first opens, return current state and a small number of recent events.

from dataclasses import dataclass

@dataclass(frozen=True)
class RunBootstrap:
    durable_status: str
    state: dict[str, str] | None
    events: list[tuple[str, dict[str, str]]]
    cursor: str | None
    replay_complete: bool

async def load_run_bootstrap(
    redis: Redis,
    *,
    run_id: str,
    durable_status: str,
    history_limit: int = 50,
) -> RunBootstrap:
    state_key = f"agent:run:{{{run_id}}}:state"
    events_key = run_events_key(run_id)

    state = await redis.hgetall(state_key)
    if not state:
        return RunBootstrap(
            durable_status=durable_status,
            state=None,
            events=[],
            cursor=None,
            replay_complete=False,
        )

    cursor = state.get("last_event_id", "0-0")

    # Read the latest history in reverse, then restore chronological UI order.
    recent_desc = await redis.xrevrange(
        events_key,
        max=cursor,
        min="-",
        count=history_limit,
    )
    recent = list(reversed(recent_desc))

    return RunBootstrap(
        durable_status=durable_status,
        state=state,
        events=recent,
        cursor=cursor,
        replay_complete=True,
    )

RunBootstrap is an API-to-browser contract, not a raw Redis response. Authorization and durable_status lookup happen first in PostgreSQL. With replay_complete=false, the client must not present the timeline as complete; it uses a durable terminal result or a degraded active view. Event payloads should contain only the minimum UI fields, not prompts, tool arguments, PII, or credentials.

Read events after the cursor with XREAD

After bootstrap, read new entries from the last cursor.

async def read_next_events(
    redis: Redis,
    *,
    run_id: str,
    last_event_id: str,
    block_ms: int = 15_000,
    count: int = 100,
) -> list[tuple[str, dict[str, str]]]:
    response = await redis.xread(
        streams={run_events_key(run_id): last_event_id},
        count=count,
        block=block_ms,
    )

    if not response:
        return []

    _, entries = response[0]
    return entries

The ID passed to XREAD is an exclusive cursor: passing 176...-0 returns entries with greater IDs. BLOCK waits for a bounded period when no new data exists and may return an empty response after timeout. That gives the server an opportunity to emit an SSE heartbeat or detect a disconnected client. (XREAD)

COUNT matters too. After a long disconnect, process the backlog in bounded batches. Advance the next cursor only to the last event actually delivered.

Pass the Redis Stream ID as the SSE id

SSE uses a text format with blank lines between messages. Reusing the Redis Stream ID as the SSE id joins the storage cursor to the browser reconnect cursor.

id: 1760000000123-0
event: rules_started
data: {"run_id":"7d5...","rules_total":4}

A small encoder looks like this:

import json

def encode_sse(event_id: str, fields: dict[str, str]) -> str:
    payload = {
        "type": fields["type"],
        "occurred_at": fields["occurred_at"],
        "payload": json.loads(fields.get("payload_json", "{}")),
        "schema_version": int(fields.get("schema_version", 1)),
    }

    return (
        f"id: {event_id}\n"
        f"event: {fields['type']}\n"
        f"data: {json.dumps(payload)}\n\n"
    )

The HTML Standard defines an SSE event id and the last event ID a user agent uses when reconnecting. (Server-Sent Events)

This article does not build the complete SSE endpoint. Async generators, StreamingResponse, heartbeat comments, proxy buffering, and disconnect cancellation belong to Article 8. The point here is that SSE is a transport for cursor-aware Redis Stream events.

Reconnect means replay, and replay permits duplicates

Networks do not disconnect neatly at message boundaries.

Server: writes event 42 to the socket
Network: browser receives it
Browser: connection drops just after handling it
Server: cannot know whether the browser handled it
Browser: may reconnect with last ID 41

Event 42 may be delivered again. UI handlers therefore need to be idempotent by event ID.

const seenEventIds = new Set();

function applyRunEvent(message) {
  if (seenEventIds.has(message.lastEventId)) return;
  seenEventIds.add(message.lastEventId);

  const event = JSON.parse(message.data);
  timeline.upsert(message.lastEventId, event);
}

The browser need not keep every ID forever. Timeline items can use event IDs as keys, or the client can compare ordering against its last applied cursor. The critical rule is not to assume SSE provides exactly-once delivery.

State updates should also be idempotent. An absolute progress value such as 65 remains correct when replayed. A delta such as progress += 5 can incorrectly turn into 70.

If the cursor predates retention, return to the snapshot

After trimming, an old client’s cursor may no longer exist. Redis can still read retained entries with greater IDs, but it cannot reconstruct deleted events.

Silently returning partial history makes an incomplete timeline look complete. The server should compare the cursor with the oldest retained ID and require a resync when it is older.

def parse_stream_id(value: str) -> tuple[int, int]:
    milliseconds, sequence = value.split("-", 1)
    return int(milliseconds), int(sequence)

async def cursor_is_too_old(
    redis: Redis,
    *,
    run_id: str,
    cursor: str,
) -> bool:
    first = await redis.xrange(
        run_events_key(run_id),
        min="-",
        max="+",
        count=1,
    )
    if not first:
        return False

    oldest_id, _ = first[0]
    return parse_stream_id(cursor) < parse_stream_id(oldest_id)

The protocol can then:

• Send an SSE resync_required control event and close the connection.
• Ask the client to call GET /runs/{id} again.
• Send a new snapshot event and move the cursor to the latest boundary.

The simplest starting point is to refetch the snapshot and reconnect from its latest cursor. If older timeline segments are mandatory, increase Redis retention or add a durable event archive.

Define what happens when the snapshot is gone

As in Article 6, the Redis Hash may have a TTL. If both keys for an old completed run have expired, the durable PostgreSQL record is the fallback.

Redis state exists
  → snapshot + events + live SSE

Redis state missing, PostgreSQL run exists
  → return the durable terminal result
  → do not open a live stream

Redis state missing, PostgreSQL run is active
  → check worker heartbeat / queue ownership
  → rebuild the projection or return a degraded response

A Redis miss must not immediately become 404 run not found. PostgreSQL determines run existence and authorization.

Choose fan-out alternatives by UI consumption semantics

Why Pub/Sub alone makes reconnect difficult

Redis Pub/Sub is simple for live fan-out, but messages published while a subscriber is disconnected cannot be read later. Redis documents Pub/Sub as at-most-once delivery: a message lost to a network disconnect is not redelivered. (Redis Pub/Sub)

Pub/Sub
publish ───────────────> connected subscribers
           messages during disconnect disappear

Stream
XADD ──> retained entries
XREAD(last_id) ──> replay the disconnect gap

Streams are therefore a natural fit for agent UIs that require reconnect and replay. A hybrid may use Pub/Sub as a wake-up signal while reading payloads from a Stream, but it need not be the starting design. XREAD BLOCK can wake waiting readers and deliver new entries by itself.

Consumer groups are not the default for SSE fan-out

Redis Stream consumer groups are useful when messages should be distributed among consumers and the system must track acknowledgements and pending ownership. They fit worker queues.

Two browsers viewing the same run, however, must both receive every event.

SSE viewer A ─┐
              ├─ both must receive rules_started
SSE viewer B ─┘

Put both viewers in one consumer group and the event may go to only one of them. Start each SSE viewer as an independent XREAD reader with its own cursor. A worker queue and UI event feed can both use Redis Streams without sharing a consumption pattern.

One Stream per run?

The examples use one Stream key per run:

agent:run:{run_id}:events

That makes one run easy to read after authorization and aligns TTL and trimming with the run lifecycle. The cost is one key per run.

A shared Stream is also possible:

agent:events
  run_id=7d5 type=rules_started
  run_id=91a type=documents_collected

But XREAD cannot filter arbitrary fields. An SSE connection for one run would read unrelated events and filter them in application code. Tenant or partitioned streams are possible, but complicate ordering, hot partitions, and retention.

At an initial scale, per-run Streams give a simpler mental model and security boundary. Measure key cardinality and connection counts before changing the partitioning scheme.

Worker code that publishes state and events together

Putting the pieces together, a worker can publish a domain transition when a node finishes.

async def execute_rule_review_step(
    *,
    redis: Redis,
    run_id: str,
    attempt_id: str,
    fencing_token: int,
    state: dict,
) -> dict:
    await publish_transition(
        redis,
        event=RunEvent(
            type="rules_started",
            run_id=run_id,
            attempt_id=attempt_id,
            fencing_token=fencing_token,
            payload={"rules_total": 4},
        ),
        status="running",
        current_step="apply_rules",
        progress=65,
        ttl_seconds=24 * 60 * 60,
    )

    decision = await apply_policy_rules(state)
    next_state = {**state, "rule_decision": decision}

    await publish_transition(
        redis,
        event=RunEvent(
            type="rules_applied",
            run_id=run_id,
            attempt_id=attempt_id,
            fencing_token=fencing_token,
            payload={
                "rules_checked": decision.rules_checked,
                "requires_approval": decision.requires_approval,
            },
        ),
        status="running",
        current_step="approval",
        progress=80,
        ttl_seconds=24 * 60 * 60,
    )

    return next_state

The event does not copy the full rule decision. Keep only the small metadata needed to render the timeline; keep the result itself in a checkpoint or PostgreSQL record. Small Stream entries make replay and memory use easier to control.

Separate progress events from domain events

Not every progress change belongs on the timeline.

Change	State update	Event append
Progress 65 → 66	Yes	Usually no
Current step changes	Yes	Yes
One tool retry	Optional	If it matters to the user
Approval required	Yes	Yes
One token generated	Separate channel	No
Run succeeds	Yes	Yes

Progress is a continuous projection; an event is a meaningful transition. Without this distinction, LLM tokens, internal debug logs, and user-facing events end up in one stream. Data with different traffic and retention profiles should not share a channel: doing so makes slow clients and memory pressure harder to manage.

Token streaming in particular is different from progress streaming. Tokens are high-volume partial output; rules_applied is a low-volume control signal. The follow-up streaming article treats that distinction separately.

Separate what appears in the browser from what remains internal:

Channel	Include	Screen placement
Timeline event	Phase completion, approval, user intervention, recovery/failure, terminal completion	Activity timeline
State only	Continuous progress, heartbeat, internal current-attempt state	Status card or hidden
Separate token channel	Partial generated output	Result panel
Debug / audit	Detailed traces, raw tool arguments, long-term records	Operations and audit systems

An event type is not display copy. The client maps rules_applied and its small payload to localized text such as “Rule review completed.” Storing an English sentence in the Stream turns copy edits and multilingual UI into event-schema changes.

You do not need full event sourcing

Separating state and events does not require rebuilding all application state exclusively from event replay.

This design is smaller than full event sourcing:

PostgreSQL     durable business truth
Redis Hash    disposable current projection
Redis Stream  bounded operational event history
Checkpoint    resumable execution snapshot

Deleting the Redis Stream still leaves the PostgreSQL result and checkpoint, and the Hash can be rebuilt. Events help the UI and observability, but they are not the sole source of every business invariant.

Consider full event sourcing when audit, temporal queries, or deterministic rebuilds impose stronger requirements. A real-time progress UI alone is not a reason to convert the entire system to event-sourced architecture.

The browser returns through replay or resync

The browser finally opens a run screen like this:

Browser                  API/SSE server                 Redis
   │                           │                          │
   │ GET /runs/{id}/bootstrap  │                          │
   ├──────────────────────────>│ HGETALL state            │
   │                           ├─────────────────────────>│
   │                           │ XREVRANGE recent events   │
   │                           ├─────────────────────────>│
   │ snapshot + history+cursor │                          │
   │<──────────────────────────┤                          │
   │                           │                          │
   │ SSE connect(last cursor)  │                          │
   ├──────────────────────────>│ XREAD cursor BLOCK       │
   │                           ├─────────────────────────>│
   │ id + event + data         │<─────────────────────────┤
   │<──────────────────────────┤                          │
   │                           │                          │
   │ disconnect / reconnect    │ XREAD last-event-id      │
   ├──────────────────────────>├─────────────────────────>│

The pattern has three important properties:

1. The snapshot renders immediately without replaying the full history.
2. The cursor prevents events after the snapshot from being missed.
3. A reconnect beyond Stream retention recovers through snapshot resync.

1. 1Bootstrapsnapshot + recent events + cursor
2. 2Live SSEupsert events after cursor
3. 3Disconnectretain last applied ID

Is the cursor within the replay window?

YesReplay later events; tolerate duplicates

Noresync_required → call bootstrap again

A Redis miss is not a 404. The durable PostgreSQL record determines a terminal result or a degraded active response.

Implementation checklist

• Are the current snapshot and append-only events stored under different keys?
• Does state hold a consistent last_event_id?
• Are snapshot and event updates atomic when one transition changes both?
• Does the event envelope include type, schema version, timestamp, and attempt identity?
• Is the fencing token validated before writing the snapshot and event?
• Is the Redis Stream ID treated as an opaque cursor?
• Does reconnect replay entries after the last event ID?
• Can the client safely process duplicate events?
• Is there a resync protocol when the cursor predates the trim window?
• Does the Stream have a maximum length or time-based retention?
• Does a Redis miss fall back to durable PostgreSQL state?
• Are credentials and sensitive data removed from event payloads?
• Are UI viewers kept out of an inappropriate consumer group?
• Are progress, timeline, token, and debug/audit channels separated by screen purpose and retention?

We now know where state and events live and which cursor connects them. Next, side-effecting tools—such as transfers—must converge on one user intent under queue redelivery and timeouts.

The next article, Can an Agent Tool Execute Exactly Once?, covers approval, fingerprints, idempotency keys, and outcome_unknown. Then Let the Agent Speak While It Works streams approval, execution, and reconciliation over SSE.