Short answer: there’s no Redis transport for task/run state updates in Prefect 3.x. Workers (really, the engine running inside your flow run) submit state updates over HTTP to the Prefect API, which persists them to the database. Redis is only available as an optional “Redis block” for user storage, not for core orchestration. Where this happens in code (client side): - State proposal from the engine: - propose_state (engine entry point): https://github.com/PrefectHQ/prefect/blob/main/src/prefect/utilities/engine.py - Client methods that submit states: - PrefectClient.set_task_run_state / set_flow_run_state: https://github.com/PrefectHQ/prefect/blob/main/src/prefect/client/orchestration/__init__.py - HTTP transport layer (httpx-based): - Prefect HTTP client and retry logic: https://github.com/PrefectHQ/prefect/blob/main/src/prefect/client/base.py Typical call path - A task starts/finishes → engine calls

propose_state(...)

- The client POSTs to the API:

# Conceptual shape in PrefectClient
  await self.request(
    "POST",
    f"/task_runs/{task_run_id}/set_state",
    json={
      "state": state_create.model_dump(mode="json", serialize_as_any=True),
      "force": force,
    },
  )

- Server applies orchestration, writes state to the DB, returns an OrchestrationResult. Server endpoints (where the API receives the updates): - Task run state: https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/api/task_runs.py - Flow run state: https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/api/flow_runs.py Related schema - State payload (what’s serialized to JSON): https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/schemas/actions.py (see

StateCreate

) Redis in Prefect 3.x - Redis is not used to transport state updates. The only Redis code in core is the optional block: - Redis block: https://github.com/PrefectHQ/prefect/blob/main/src/prefect/blocks/redis.py Notes - In Prefect 3.x, workers execute flow runs and the flow engine inside that run submits both flow and task run states via the client. There’s no queue/broker-based (e.g., Redis) path for task state updates. - If you’re actually on Prefect 2.x or looking for logging/heartbeats, let me know and I can point you to those code paths too.

Great callout — you’re right. In Prefect 3.x, events are emitted by an EventsWorker over a WebSocket connection (separate from the HTTP state updates). Here are the exact code paths and how it all fits together. Where EventsWorker lives - Events worker implementation: prefect/events/worker.py - Subclasses a queue-based service; runs an async loop that drains a queue and sends events via a WebSocket client. How it’s started and how events get enqueued - High-level emit helper (called from engine code): prefect/events/utilities.py - Typically funnels through something like

emit_event(...)

, which acquires the singleton EventsWorker instance and enqueues the event. - Queue service base: prefect/_internal/concurrency/services.py - Provides the queue + background loop pattern that EventsWorker uses (enqueue in the foreground, handle/send in the background). WebSocket client and send path - Events clients (WebSocket sender): prefect/events/clients.py - Builds the WebSocket client and defines the

emit(...)

call that pushes events to the server. - WebSocket utilities used by the client: prefect/_internal/websockets.py - Handles connection details, auth headers, ping/pong, and reconnect/backoff. Typical sending flow inside EventsWorker (conceptual shape):

# enqueue (synchronous)
EventsWorker.instance().send(event_dict)

# background loop (async)
while True:
    event = await queue.get()
    await events_client.emit(event)  # sends JSON over WS to the API

Where engine/worker emits events for task/flow states - Task engine emits task-run state change events: prefect/task_engine.py - Look for utilities like

emit_task_run_state_change_event(...)

. - Flow/task state proposal utilities (call emit on state changes): prefect/utilities/engine.py - State transitions here often trigger the corresponding event emission helpers. Server-side receiver - Event ingestion endpoint: prefect/server/api/events.py - Receives events (WebSocket) and forwards them into the server’s event pipeline. - Event processing pipeline: prefect/server/events/pipeline.py Relation to task/run state updates - State updates (accept/reject/wait/abort) still go over HTTP via the orchestration client: - Client: prefect/client/orchestration/__init__.py - Engine proposal: prefect/utilities/engine.py - Events are emitted in parallel via WebSocket by the EventsWorker for telemetry like state-change events. No Redis is involved in this transport. If you want, I can point you to the specific methods that build the event payload (e.g.,

event

,

resource

,

related

,

payload

) for task-run state changes.

clients just send messages to the

events/in

ws endpoint, which offloads to redis and then some background services consume the redis streams