SSE and WebSockets

The plugin can add task, queue, worker, global, and custom stream routes under /queues/events:

from litestar_queues import QueueConfig
from litestar_queues.events import EventDeliveryConfig, EventStreamConfig, QueueEventsConfig

queue_config = QueueConfig(
    events=QueueEventsConfig(
        channels=channels_backend,
        delivery=EventDeliveryConfig(),
        stream=EventStreamConfig(
            transports={"sse", "websocket"},
            scopes={"task", "queue"},
            channel_authorizer=authorize_channel,
        ),
    ),
)

Use SSE for server-to-browser updates with automatic browser reconnection. Use WebSockets when the connection also needs bidirectional application messages. Both deliver the same JSON QueueEvent envelope.

When QueueEventsConfig.channels is set, generated stream routes use that same Channels source even if the application registers another ChannelsPlugin. Without an explicit source, the routes use the registered plugin. Event publishing and route registration remain separate settings so a web process can serve events published by another process.

Envelope and delivery semantics

Top-level JSON fields use camel case, including taskId, workerId, scopeKey, progressCurrent, occurredAt, and eventKey. Null-valued keys are preserved; user-owned payload keys are not renamed. Consumers should deduplicate by eventKey when present and by id otherwise.

Live streams deliver events; they do not store tasks. Keepalives keep an idle connection open, but they do not prove that a task is healthy. A slow client may miss best-effort events, depending on the Channels backend and its backlog policy. Query Event history when a client must replay events.

Authorization

Application-level guards apply to the generated routes. Use guards to add stream-specific protection and channel_authorizer to check each scope and key. The scopes setting controls which route families are registered; it does not authorize access. Never accept a task ID or queue name from a client without checking that the tenant and user may access it.

For an intentionally public stream, set unauthenticated_access="allow" to acknowledge that choice and suppress the startup warning. Set include_in_schema=True separately when the generated routes should appear in the OpenAPI schema.

Canonical runnable sources

The memory examples are the complete examples. Each backend copy changes only the queue-storage setup and keeps the same frontend behavior.

WebSocket application:

examples/htmx_realtime_websocket/app.py

import asyncio
from pathlib import Path

from litestar import Litestar, get, post
from litestar.channels import ChannelsPlugin
from litestar.channels.backends.memory import MemoryChannelsBackend
from litestar.di import NamedDependency
from litestar.plugins.htmx import HTMXPlugin, HTMXTemplate
from litestar.plugins.jinja import JinjaTemplateEngine
from litestar.response import Template
from litestar.template.config import TemplateConfig
from litestar_vite import PathConfig, ViteConfig, VitePlugin

from litestar_queues import QueueConfig, QueuePlugin, QueueService, WorkerConfig, task
from litestar_queues.events import (
    EventBufferConfig,
    EventDeliveryConfig,
    EventStreamConfig,
    QueueEventsConfig,
    TaskExecutionContext,
    publish_task_log,
)

__all__ = ("index", "restart_demo", "run_crawl", "status_json")


SHARED_ROOT = Path(__file__).parent.parent / "shared"
VITE_ENTRY = "resources/main-ws.ts"
BACKEND_NAME = "memory"
DEMO_STEPS = 6
DEMO_STEP_DELAY = 1
DEMO_KEY = "demo:current"

CRAWL_LINES = (
    "The task started.",
    "The task is visiting the next page.",
    "The task is sleeping before the next page.",
    "The task is visiting another page.",
    "The task is almost done.",
    "The task finished the crawl.",
)
DISCOVERED_PAGE_STEP = DEMO_STEPS // 2
DISCOVERED_PAGE_URL = "https://example.invalid/articles/queues-101"


# -- docs-task-start --
@task("examples.htmx_realtime_websocket.crawl", queue="demo", retries=0, timeout=90)
async def run_crawl(*, _task_context: TaskExecutionContext) -> dict[str, str]:
    ctx = _task_context
    await publish_task_log("The task started", payload={"task_id": ctx.task_id}, immediate=True)

    for step in range(1, DEMO_STEPS + 1):
        line = CRAWL_LINES[(step - 1) % len(CRAWL_LINES)]
        message = f"{step}/{DEMO_STEPS} - {line}"
        await ctx.progress(current=step, total=DEMO_STEPS, message=message, payload={"line": message})
        if step == DISCOVERED_PAGE_STEP:
            # One meaningful domain event at a real transition, with a payload
            # distinct from the generic per-step progress text above.
            await ctx.event(
                "crawl.page_discovered",
                message=f"Discovered {DISCOVERED_PAGE_URL}",
                payload={"url": DISCOVERED_PAGE_URL},
            )
        await asyncio.sleep(DEMO_STEP_DELAY)

    return {"task_id": ctx.task_id, "status": "complete"}


# -- docs-task-end --


# -- docs-routes-start --
@get("/")
async def index() -> Template:
    return Template("index.html", context={"backend_name": BACKEND_NAME, "vite_entry": VITE_ENTRY})


@get("/demo/status")
async def status_json() -> dict[str, str]:
    return {"backend": BACKEND_NAME}


@post("/demo/restart")
async def restart_demo(queue_service: NamedDependency[QueueService]) -> Template:
    existing = await queue_service.get_queue_backend().get_task_by_key(DEMO_KEY)
    result = await queue_service.enqueue(run_crawl, key=DEMO_KEY, description="Run the HTMX realtime queue event demo")
    task_id = str(result.id)
    reused = existing is not None and not existing.is_terminal and existing.id == result.id
    return HTMXTemplate(
        template_name="partials/stream_mount_ws.html",
        context={"task_id": task_id, "task_ws_url": f"/queues/events/tasks/{result.id}"},
        push_url=False,
        re_target="#stream-mount",
        trigger_event="queue-demo:started",
        params={"taskId": task_id, "reused": reused},
        after="swap",
    )


# -- docs-routes-end --


# -- docs-app-config-start --
channels = ChannelsPlugin(
    backend=MemoryChannelsBackend(history=200),
    arbitrary_channels_allowed=True,
    subscriber_max_backlog=1000,
    subscriber_backlog_strategy="dropleft",
)

queue_config = QueueConfig(
    # This demo runs everything in one process: a process-local queue and a
    # process-local Channels backend, so the worker must live here too. The
    # Redis and Valkey variants share both across processes instead.
    queue_backend="memory",
    worker=WorkerConfig(placement="asgi", graceful_shutdown_timeout=5),
    events=QueueEventsConfig(
        channels=channels,
        delivery=EventDeliveryConfig(buffer=EventBufferConfig(batch_size=8, flush_interval=0.2)),
        stream=EventStreamConfig(
            scopes={"task"},
            replay_limit=25,
            heartbeat_interval=15,
            unauthenticated_access="allow",
            transports={"websocket"},
        ),
    ),
)

vite_config = ViteConfig(enabled=True, mode="htmx", paths=PathConfig(root=SHARED_ROOT, resource_dir="resources"))

app = Litestar(
    route_handlers=[index, status_json, restart_demo],
    template_config=TemplateConfig(directory=SHARED_ROOT / "templates", engine=JinjaTemplateEngine),
    signature_namespace={"NamedDependency": NamedDependency},
    plugins=[HTMXPlugin(), channels, QueuePlugin(queue_config), VitePlugin(config=vite_config)],
)
# -- docs-app-config-end --

WebSocket frontend. All ten examples share one Vite project under examples/shared, with an entry point per transport:

examples/shared/resources/main-ws.ts

// docs: htmx-extension-start
import htmx from "htmx.org"
import { registerHtmxExtension } from "litestar-vite-plugin/helpers"
import "./styles.css"

// htmx 2's ESM build never attaches itself to window, and the extensions read
// the global. Publish it first, register the litestar JSON extension, then load
// the WebSocket extension AFTER the global exists (a dynamic import runs in
// order, unlike a hoisted static import). Vite loads this module async, so it
// can run after DOMContentLoaded has already fired while readyState is
// "interactive" — htmx's own boot listener misses that window, so process the
// DOM explicitly.
;(window as unknown as { htmx: typeof htmx }).htmx = htmx
registerHtmxExtension()
void import("htmx-ext-ws").then(() => htmx.process(document.body))
// docs: htmx-extension-end

const MAX_LINES = 40
// After the final event, wait for the last lines to drift out of view, then
// return to the idle hint unless another task starts first.
const IDLE_DELAY_MS = 45_000
const IDLE_TEXT = document.querySelector<HTMLElement>("#crawl-lines p")?.textContent ?? "Press restart to send a background task."
let idleTimer: ReturnType<typeof setTimeout> | null = null

type QueueEvent = {
  type: string
  message?: string | null
  payload?: Record<string, unknown>
}

function parseQueueEvent(raw: string): QueueEvent | null {
  try {
    return JSON.parse(raw) as QueueEvent
  } catch {
    return null
  }
}

function eventLine(event: QueueEvent): string | null {
  if (event.message) return event.message
  if (typeof event.payload?.line === "string") return event.payload.line
  return null
}

function setReadout(text: string, completed = false): void {
  const readout = document.querySelector<HTMLElement>("#job-readout")
  if (!readout) return
  readout.textContent = text
  readout.classList.toggle("completed", completed)
}

function appendCrawlLine(event: QueueEvent): void {
  const target = document.querySelector<HTMLElement>("#crawl-lines")
  if (!target) return
  const text = eventLine(event)
  if (!text || target.lastElementChild?.textContent === text) return
  const line = document.createElement("p")
  line.textContent = text
  line.dataset.eventType = event.type
  target.append(line)
  while (target.children.length > MAX_LINES) target.firstElementChild?.remove()
}

function markBackendMessage(event: QueueEvent): void {
  const status = document.querySelector<HTMLElement>("#delivery-status")
  const label = document.querySelector<HTMLElement>("#delivery-label")
  if (!status || !label) return
  label.textContent = `Backend message received · ${event.type}`
  status.classList.remove("received")
  void status.offsetHeight
  status.classList.add("received")
}

// The one adapter the extensions cannot replace: queue frames are JSON, so they
// cannot be swapped as HTML. Parse each frame, ignore ping heartbeats, and
// append the event line. The terminal event only flips the readout — the last
// progress line is already the on-screen finale.
function handleQueueEvent(raw: string): void {
  const event = parseQueueEvent(raw)
  if (!event || event.type === "ping") return
  markBackendMessage(event)
  if (event.type === "task.completed") {
    setReadout("Task complete", true)
    if (idleTimer) clearTimeout(idleTimer)
    idleTimer = setTimeout(goIdle, IDLE_DELAY_MS)
    return
  }
  appendCrawlLine(event)
}

function goIdle(): void {
  const plane = document.querySelector<HTMLElement>("#crawl-lines")
  if (!plane) return
  const hint = document.createElement("p")
  hint.textContent = IDLE_TEXT
  plane.replaceChildren(hint)
  plane.classList.add("idle")
  setReadout("Awaiting launch")
}

function resetCrawl(taskId?: string, reused = false): void {
  if (idleTimer) clearTimeout(idleTimer)
  idleTimer = null
  const plane = document.querySelector<HTMLElement>("#crawl-lines")
  if (plane) {
    plane.classList.remove("idle")
    plane.replaceChildren()
    // The crawl keyframe is one-shot and time-based; restart it so every
    // task's lines enter from the bottom instead of mid-flight.
    plane.style.animation = "none"
    void plane.offsetHeight
    plane.style.animation = ""
  }
  setReadout(
    taskId
      ? reused
        ? `Task ${taskId} is already running`
        : `Task ${taskId} queued`
      : "Task queued",
  )
}

// docs: websocket-client-start
// htmx-ext-ws fires htmx:wsBeforeMessage (cancelable) for every socket message
// before it would swap. detail.message is the raw JSON string; preventDefault
// keeps htmx from injecting JSON into the DOM.
document.body.addEventListener("htmx:wsBeforeMessage", (event) => {
  event.preventDefault()
  const detail = (event as CustomEvent<{ message: string }>).detail
  handleQueueEvent(detail.message)
})
// docs: websocket-client-end

// docs: stream-adapter-start
// HTMXTemplate(trigger_event="queue-demo:started") fires after the restart swap
// replaces #stream-mount. Swapping the element reconnects the stream (the old
// WebSocket closes with the removed element); reset the crawl to match.
document.body.addEventListener("queue-demo:started", (event) => {
  const detail = (event as CustomEvent<{ taskId?: string; reused?: boolean }>).detail
  resetCrawl(detail?.taskId, detail?.reused === true)
})
// docs: stream-adapter-end

examples/shared/templates/partials/stream_mount_ws.html

<!--
  Returned by POST /demo/restart. This element IS the WebSocket connection: the
  htmx WS extension opens a socket for ws-connect when the element is swapped in,
  and closes it automatically when the next restart replaces the element.
  Connection lifecycle == swap lifecycle. Every queue message fires
  htmx:wsBeforeMessage; the JS adapter renders it as a crawl line and cancels the
  default HTML swap so raw JSON is never injected into the DOM.
-->
<div
  id="stream-mount"
  class="stream-mount"
  aria-hidden="true"
  hx-ext="ws"
  ws-connect="{{ task_ws_url }}"
  data-task-id="{{ task_id }}"
></div>

SSE application:

examples/htmx_realtime_sse/app.py

import asyncio
from pathlib import Path

from litestar import Litestar, get, post
from litestar.channels import ChannelsPlugin
from litestar.channels.backends.memory import MemoryChannelsBackend
from litestar.di import NamedDependency
from litestar.plugins.htmx import HTMXPlugin, HTMXTemplate
from litestar.plugins.jinja import JinjaTemplateEngine
from litestar.response import Template
from litestar.template.config import TemplateConfig
from litestar_vite import PathConfig, ViteConfig, VitePlugin

from litestar_queues import QueueConfig, QueuePlugin, QueueService, WorkerConfig, task
from litestar_queues.events import (
    EventBufferConfig,
    EventDeliveryConfig,
    EventStreamConfig,
    QueueEventsConfig,
    TaskExecutionContext,
    publish_task_log,
)

__all__ = ("index", "restart_demo", "run_crawl", "status_json")


SHARED_ROOT = Path(__file__).parent.parent / "shared"
VITE_ENTRY = "resources/main-sse.ts"
BACKEND_NAME = "memory"
DEMO_STEPS = 6
DEMO_STEP_DELAY = 1
DEMO_KEY = "demo:current"

CRAWL_LINES = (
    "The task started.",
    "The task is visiting the next page.",
    "The task is sleeping before the next page.",
    "The task is visiting another page.",
    "The task is almost done.",
    "The task finished the crawl.",
)
DISCOVERED_PAGE_STEP = DEMO_STEPS // 2
DISCOVERED_PAGE_URL = "https://example.invalid/articles/queues-101"


# -- docs-task-start --
@task("examples.htmx_realtime_sse.crawl", queue="demo", retries=0, timeout=90)
async def run_crawl(*, _task_context: TaskExecutionContext) -> dict[str, str]:
    ctx = _task_context
    await publish_task_log("The task started", payload={"task_id": ctx.task_id}, immediate=True)

    for step in range(1, DEMO_STEPS + 1):
        line = CRAWL_LINES[(step - 1) % len(CRAWL_LINES)]
        message = f"{step}/{DEMO_STEPS} - {line}"
        await ctx.progress(current=step, total=DEMO_STEPS, message=message, payload={"line": message})
        if step == DISCOVERED_PAGE_STEP:
            # One meaningful domain event at a real transition, with a payload
            # distinct from the generic per-step progress text above.
            await ctx.event(
                "crawl.page_discovered",
                message=f"Discovered {DISCOVERED_PAGE_URL}",
                payload={"url": DISCOVERED_PAGE_URL},
            )
        await asyncio.sleep(DEMO_STEP_DELAY)

    return {"task_id": ctx.task_id, "status": "complete"}


# -- docs-task-end --


# -- docs-routes-start --
@get("/")
async def index() -> Template:
    return Template("index.html", context={"backend_name": BACKEND_NAME, "vite_entry": VITE_ENTRY})


@get("/demo/status")
async def status_json() -> dict[str, str]:
    return {"backend": BACKEND_NAME}


@post("/demo/restart")
async def restart_demo(queue_service: NamedDependency[QueueService]) -> Template:
    existing = await queue_service.get_queue_backend().get_task_by_key(DEMO_KEY)
    result = await queue_service.enqueue(run_crawl, key=DEMO_KEY, description="Run the HTMX realtime queue event demo")
    task_id = str(result.id)
    reused = existing is not None and not existing.is_terminal and existing.id == result.id
    return HTMXTemplate(
        template_name="partials/stream_mount_sse.html",
        context={"task_id": task_id, "task_sse_url": f"/queues/events/sse/tasks/{result.id}"},
        push_url=False,
        re_target="#stream-mount",
        trigger_event="queue-demo:started",
        params={"taskId": task_id, "reused": reused},
        after="swap",
    )


# -- docs-routes-end --


# -- docs-app-config-start --
channels = ChannelsPlugin(
    backend=MemoryChannelsBackend(history=200),
    arbitrary_channels_allowed=True,
    subscriber_max_backlog=1000,
    subscriber_backlog_strategy="dropleft",
)

queue_config = QueueConfig(
    # This demo runs everything in one process: a process-local queue and a
    # process-local Channels backend, so the worker must live here too. The
    # Redis and Valkey variants share both across processes instead.
    queue_backend="memory",
    worker=WorkerConfig(placement="asgi", graceful_shutdown_timeout=5),
    events=QueueEventsConfig(
        channels=channels,
        delivery=EventDeliveryConfig(buffer=EventBufferConfig(batch_size=8, flush_interval=0.2)),
        stream=EventStreamConfig(
            scopes={"task"}, replay_limit=25, heartbeat_interval=15, unauthenticated_access="allow", transports={"sse"}
        ),
    ),
)

vite_config = ViteConfig(enabled=True, mode="htmx", paths=PathConfig(root=SHARED_ROOT, resource_dir="resources"))

app = Litestar(
    route_handlers=[index, status_json, restart_demo],
    template_config=TemplateConfig(directory=SHARED_ROOT / "templates", engine=JinjaTemplateEngine),
    signature_namespace={"NamedDependency": NamedDependency},
    plugins=[HTMXPlugin(), channels, QueuePlugin(queue_config), VitePlugin(config=vite_config)],
)
# -- docs-app-config-end --

SSE frontend:

examples/shared/resources/main-sse.ts

// docs: htmx-extension-start
import htmx from "htmx.org"
import { registerHtmxExtension } from "litestar-vite-plugin/helpers"
import "./styles.css"

// htmx 2's ESM build never attaches itself to window, and the extensions read
// the global. Publish it first, register the litestar JSON extension, then load
// the SSE extension AFTER the global exists (a dynamic import runs in order,
// unlike a hoisted static import). Vite loads this module async, so it can run
// after DOMContentLoaded has already fired while readyState is "interactive" —
// htmx's own boot listener misses that window, so process the DOM explicitly.
;(window as unknown as { htmx: typeof htmx }).htmx = htmx
registerHtmxExtension()
void import("htmx-ext-sse").then(() => htmx.process(document.body))
// docs: htmx-extension-end

const MAX_LINES = 40
// After the final event, wait for the last lines to drift out of view, then
// return to the idle hint unless another task starts first.
const IDLE_DELAY_MS = 45_000
const IDLE_TEXT = document.querySelector<HTMLElement>("#crawl-lines p")?.textContent ?? "Press restart to send a background task."
let idleTimer: ReturnType<typeof setTimeout> | null = null

type QueueEvent = {
  type: string
  message?: string | null
  payload?: Record<string, unknown>
}

function parseQueueEvent(raw: string): QueueEvent | null {
  try {
    return JSON.parse(raw) as QueueEvent
  } catch {
    return null
  }
}

function eventLine(event: QueueEvent): string | null {
  if (event.message) return event.message
  if (typeof event.payload?.line === "string") return event.payload.line
  return null
}

function setReadout(text: string, completed = false): void {
  const readout = document.querySelector<HTMLElement>("#job-readout")
  if (!readout) return
  readout.textContent = text
  readout.classList.toggle("completed", completed)
}

function appendCrawlLine(event: QueueEvent): void {
  const target = document.querySelector<HTMLElement>("#crawl-lines")
  if (!target) return
  const text = eventLine(event)
  if (!text || target.lastElementChild?.textContent === text) return
  const line = document.createElement("p")
  line.textContent = text
  line.dataset.eventType = event.type
  target.append(line)
  while (target.children.length > MAX_LINES) target.firstElementChild?.remove()
}

function markBackendMessage(event: QueueEvent): void {
  const status = document.querySelector<HTMLElement>("#delivery-status")
  const label = document.querySelector<HTMLElement>("#delivery-label")
  if (!status || !label) return
  label.textContent = `Backend message received · ${event.type}`
  status.classList.remove("received")
  void status.offsetHeight
  status.classList.add("received")
}

// The one adapter the extensions cannot replace: queue frames are JSON, so they
// cannot be swapped as HTML. Parse each frame, ignore ping heartbeats, and
// append the event line. The terminal event only flips the readout — the last
// progress line is already the on-screen finale.
function handleQueueEvent(raw: string): void {
  const event = parseQueueEvent(raw)
  if (!event || event.type === "ping") return
  markBackendMessage(event)
  if (event.type === "task.completed") {
    setReadout("Task complete", true)
    if (idleTimer) clearTimeout(idleTimer)
    idleTimer = setTimeout(goIdle, IDLE_DELAY_MS)
    return
  }
  appendCrawlLine(event)
}

function goIdle(): void {
  const plane = document.querySelector<HTMLElement>("#crawl-lines")
  if (!plane) return
  const hint = document.createElement("p")
  hint.textContent = IDLE_TEXT
  plane.replaceChildren(hint)
  plane.classList.add("idle")
  setReadout("Awaiting launch")
}

function resetCrawl(taskId?: string, reused = false): void {
  if (idleTimer) clearTimeout(idleTimer)
  idleTimer = null
  const plane = document.querySelector<HTMLElement>("#crawl-lines")
  if (plane) {
    plane.classList.remove("idle")
    plane.replaceChildren()
    // The crawl keyframe is one-shot and time-based; restart it so every
    // task's lines enter from the bottom instead of mid-flight.
    plane.style.animation = "none"
    void plane.offsetHeight
    plane.style.animation = ""
  }
  setReadout(
    taskId
      ? reused
        ? `Task ${taskId} is already running`
        : `Task ${taskId} queued`
      : "Task queued",
  )
}

// docs: sse-client-start
// htmx-ext-sse fires htmx:sseBeforeMessage (cancelable) for each named frame in
// sse-swap before it would swap. detail is the raw MessageEvent; preventDefault
// keeps htmx from injecting JSON into the DOM.
document.body.addEventListener("htmx:sseBeforeMessage", (event) => {
  event.preventDefault()
  const message = (event as CustomEvent<MessageEvent<string>>).detail
  handleQueueEvent(message.data)
})
// docs: sse-client-end

// docs: stream-adapter-start
// HTMXTemplate(trigger_event="queue-demo:started") fires after the restart swap
// replaces #stream-mount. Swapping the element reconnects the stream (the old
// EventSource closes with the removed element); reset the crawl to match.
document.body.addEventListener("queue-demo:started", (event) => {
  const detail = (event as CustomEvent<{ taskId?: string; reused?: boolean }>).detail
  resetCrawl(detail?.taskId, detail?.reused === true)
})
// docs: stream-adapter-end

examples/shared/templates/partials/stream_mount_sse.html

<!--
  Returned by POST /demo/restart. This element IS the SSE connection: the htmx
  SSE extension opens an EventSource for sse-connect when the element is swapped
  in, and closes it automatically when the next restart replaces the element.
  Connection lifecycle == swap lifecycle. sse-swap lists every queue event type
  so htmx:sseBeforeMessage fires for each; the JS adapter renders them as crawl
  lines and cancels the default HTML swap (hx-swap="none" belt-and-suspenders).
-->
<div
  id="stream-mount"
  class="stream-mount"
  aria-hidden="true"
  hx-ext="sse"
  sse-connect="{{ task_sse_url }}"
  sse-swap="task.started,task.progress,task.log,crawl.page_discovered,task.completed,task.failed"
  hx-swap="none"
  data-task-id="{{ task_id }}"
></div>

The page markup both transports render is shared too:

examples/shared/templates/index.html

{% extends "base.html" %}

{% block content %}
<!-- docs: template-start -->
<main class="stage" data-demo-root>
  <div class="starfield" aria-hidden="true">
    <div class="stars stars-far"></div>
    <div class="stars stars-mid"></div>
    <div class="stars stars-near"></div>
  </div>
  <div class="nebula" aria-hidden="true"></div>

  <a class="brand" href="https://litestar.dev" aria-label="Litestar">
    <svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 64 64" fill="none" role="img" aria-hidden="true">
      <g stroke="#edb641" stroke-width="3.4" stroke-linecap="round">
        <line x1="24.4" y1="16.2" x2="36.6" y2="17.8"/>
        <line x1="39.5" y1="25.4" x2="25.6" y2="41.6"/>
        <line x1="46.3" y1="27.4" x2="47.8" y2="37.6"/>
        <line x1="28.5" y1="47.4" x2="40.5" y2="46.6"/>
      </g>
      <g stroke="#edb641" stroke-width="3.4" fill="none">
        <circle cx="16" cy="15" r="6"/>
        <circle cx="45" cy="19" r="6"/>
        <circle cx="20" cy="48" r="6"/>
        <circle cx="49" cy="46" r="6"/>
      </g>
    </svg>
  </a>

  <!-- litestar-vite JSON-template extension: a whisper-quiet backend-name caption. -->
  <div class="backend-caption" hx-ext="litestar" hx-get="/demo/status" hx-trigger="load" hx-swap="json">
    <template ls-if="backend"><span>${backend}</span></template>
    <template ls-else><span>{{ backend_name }}</span></template>
  </div>

  <div class="crawl-perspective" aria-live="polite" aria-label="Task events">
    <div id="crawl-lines" class="crawl-plane idle">
      <p>Press restart to send a background task and watch its messages arrive.</p>
    </div>
  </div>

  <p class="telemetry-caption"><span class="telemetry-dot" aria-hidden="true"></span>Live messages · each update is pushed from the backend job</p>

  <div id="delivery-status" class="delivery-status" role="status" aria-live="polite">
    <span class="delivery-pulse" aria-hidden="true"></span>
    <span id="delivery-label">Waiting for a backend message</span>
  </div>

  <p id="job-readout" class="readout">Awaiting launch</p>

  <button
    class="planet"
    type="button"
    hx-post="/demo/restart"
    hx-target="#stream-mount"
    hx-swap="outerHTML"
    hx-indicator="this"
    hx-disabled-elt="this"
    hx-sync="this:replace"
    aria-label="Restart the demo queue task"
  >
    <span class="planet-orbit" aria-hidden="true"><span class="planet-satellite"></span></span>
    <span class="planet-label">Restart</span>
    <span class="planet-launching" aria-hidden="true">Launching</span>
  </button>

  <!-- The htmx SSE extension owns this element after each restart swaps it. -->
  <div id="stream-mount" class="stream-mount" aria-hidden="true"></div>
</main>
<!-- docs: template-end -->
{% endblock %}

Backend variants

The gallery includes these runnable source families:

• examples/htmx_realtime_websocket_sqlspec and examples/htmx_realtime_sse_sqlspec

• examples/htmx_realtime_websocket_advanced_alchemy and examples/htmx_realtime_sse_advanced_alchemy

• examples/htmx_realtime_websocket_redis and examples/htmx_realtime_sse_redis

• examples/htmx_realtime_websocket_valkey and examples/htmx_realtime_sse_valkey

The browser test clicks Restart, observes the enqueue request and page updates, then waits for the final event. A curl stream cannot prove that the HTMX extension starts or that the browser opens and closes the connection correctly.