# stapel-realtime 0.2.3

Realtime delivery substrate: the L1 library behind the Signal primitive (stapel_core.comm.signal). Ships the Channels/Redis transport for the core's signal-delivery seam, the two consumers every browser socket in the fleet is built from (EphemeralStreamConsumer for at-most-once Signal fan-out; ResumableStreamConsumer for hello/welcome/replay/live journals with seq dedup and a bounded replay window), the versioned v1 wire envelope, the canonical <mod>:<scope_type>:<scope_id>[:<topic>] stream key, a fail-closed per-stream authorize seam with the workspace-capability authorizer, revoke-to-kick, heartbeat with JWT-exp re-check, disconnect-on-overflow backpressure, the fleet close-code canon, build_websocket_application() host assembly with a port-aware origin guard, the fleet's presence oracle, and eight system checks. Presence is a TTL lease in the fleet-shared cache (stapel_core.core.fleet_cache), written by the base consumer on connect, on every heartbeat tick and on disconnect — no model, no migration — and read by two comm Functions: realtime.is_live {user_id, family?} -> {live, sessions, last_seen} and realtime.live_batch {user_ids[<=100], family?} -> {users: {id: {...}}}. That pair is the whole comm surface; there are no models, migrations, views or urls, and no HTTP route of its own, so a peer asks over the bus like any other Function. It is installed as a Django app so the checks and the Functions are registered.

Contract: axes 9 · surface 23 · extension points 6.
Generated from docs/capabilities.json by `stapel-llms-txt` — do not edit; drift-gated by `make contract-check`.

## Configuration axes — what a product switches on
Settings keys; `default` is what you get by saying nothing. Turning an axis off unmounts the operations it gates.
- ALLOWED_ORIGINS [list, default []] — Which web origins may open a realtime socket
  Exact origins WITH port that may open a socket. Empty disables the guard (realtime.W002); an entry that is not a scheme://host[:port] origin is realtime.E003, because a guard that can never match is worse than no guard. Configured-but-all-malformed refuses everything rather than falling open.
- AUTHORIZE_CACHE_S [int, default 30] — How long a subscription verdict is trusted before it is re-asked
  Seconds an authorize() verdict is reused within one socket, and the interval at which an open socket is re-asked uncached: the ceiling on how long a withdrawn right keeps a live stream without revoke().
- HEARTBEAT_S [int, default 25] — How often a live socket is proven alive and still credentialed
  Seconds between server-initiated ping frames. Each tick ALSO re-checks the JWT's exp and closes 4401 if it has passed, so setting this to 0 disables two things, not one (realtime.W003 says so at check time).
- HEARTBEAT_TIMEOUT_S [int, default 10] — Grace period before a silent socket is reaped
  Seconds to wait for the client's pong before closing 4408.
- LAYER_SOCKET_TIMEOUT_MIN [int|null, default null] — Floor for the channel layer's redis socket timeout
  Minimum acceptable redis socket_timeout for the channel layer; null derives it from the layer's expiry + 10s. redis-py >= 8 defaults socket_timeout to 5s, which tears down a consumer parked in BZPOPMIN on an idle stream — realtime.E002.
- MAX_REPLAY [int, default 500] — How far back a reconnecting client is caught up over the socket
  Widest resume gap replayed inline on a resumable stream, and the LIMIT passed to the module's get_replay_rows hook. A wider gap answers a resync frame (not an error) and the client re-hydrates over HTTP pagination — there is no infinite rewind by design.
- PRESENCE_TTL_S [int, default 60] — How long a person counts as watching after their last heartbeat
  Seconds one live session counts for without a refresh — a LEASE, renewed by the base consumer on every heartbeat tick, so a worker killed mid-socket stops counting one TTL later instead of leaving a user online forever. It must stay above HEARTBEAT_S or a healthy socket lets its own lease expire between two beats (realtime.W006). 0 disables the registry and realtime.is_live then answers 'not live' for everyone, which the same check reports.
- SEND_QUEUE_SIZE [int, default 100] — How far a slow browser may fall behind before it is dropped
  Outbound frames buffered per socket before the substrate closes 4413. The producer never waits on a slow client; the client resyncs on reconnect.
- URL_PREFIX [str, default "ws"] — URL prefix for socket routes
  The edge convention socket routes live under (/ws/<mod>/...), asserted by realtime.W004. Read ONLY from this key, never a bare URL_PREFIX setting (a service's HTTP mount) — one present alone is realtime.W007.

## Usage surface — call these before writing your own
This is the answer to "does Stapel already have something for X?". `instead of` names the outside symbol this one displaces.
### gate_function
- deny — stapel_realtime.authorize.deny
  The default authorize hook: refuse, and log which stream a consumer forgot to gate. Fail-closed is the whole design — an open default is a leak that looks like working code in every test a module writes for itself. Do not call it; know that it is what you get by not overriding authorize().
- revoke — stapel_realtime.delivery.revoke
  instead of: waiting for the client to reconnect before re-checking access
  Kick one user (or everyone) off a stream immediately when their access ends — call it from the module's @on_action subscriber on the membership Action. The subscriber gets a `kick` frame and close 4410. This is the answer to the main leak vector: a socket that keeps streaming after the user was removed. Without it the residual window is the 30-second capability cache.
### predicate
- close_code_name — stapel_realtime.close_codes.close_code_name
  Machine name for a close code (4403 -> 'forbidden'). Use it in logs and in a client's reconnect switch instead of comparing bare numbers; TERMINAL_CLOSE_CODES (4403, 4404, 4410) are the ones a client must not retry; 4401 means refresh the token.
- is_live — stapel_realtime.presence.is_live
  instead of: a module keeping its own socket-scoped presence table, which reads offline for anyone watching a different module's stream
  Does this user have a live realtime session right now — {live, sessions, last_seen}, optionally narrowed to one stream family. The in-process half of the fleet oracle; a PEER service asks the same question as the realtime.is_live comm Function, because the process holding the socket is not the one deciding whether to send a push. Never raises: an unreachable cache answers 'not live', so a caller gating a notification falls back to sending it.
- live_batch — stapel_realtime.presence.live_batch
  instead of: a loop of single is_live calls, one round trip per recipient
  is_live for up to 100 users in one cache round trip — the read a notifier makes before fanning out to a group. Every id asked about comes back, including one nobody has ever seen, so an absence can never be mistaken for an answer.
- user_id_from_scope — stapel_realtime.authorize.user_id_from_scope
  The connecting user's pk, or None for an anonymous scope — the one place an authorizer should read identity from, so 'authenticated' means the same thing in every module's hook.
### factory
- build_websocket_application — stapel_realtime.asgi.build_websocket_application
  instead of: a hand-written ProtocolTypeRouter/URLRouter/AuthMiddlewareStack in a host's asgi.py
  Assemble the host's whole ProtocolTypeRouter in one call: origin guard (compared WITH the port) over core's G14 JWT stack over a URLRouter of every installed module's routing manifest. This is what a host's asgi.py should contain instead of a hand-written stack — the three existing hand-written ones are why the fleet has three different auth and close-code conventions.
- collect_websocket_urlpatterns — stapel_realtime.asgi.collect_websocket_urlpatterns
  instead of: a hand-kept list of module routing imports in the host
  Discover every installed app's `<package>.routing.websocket_urlpatterns`. Libraries carry the manifest and the assembly reads it, so a scaffold can emit an ASGI file without knowing any module by name — and a routing module that fails to import raises instead of silently contributing no routes.
- deliver — stapel_realtime.delivery.deliver
  instead of: a hand-rolled async_to_sync(get_channel_layer().group_send) in a module
  The transport callable the core's STAPEL_COMM['SIGNAL_TRANSPORT'] = 'channels' resolves to: transport(stream_key, frame), called after commit with the complete envelope stapel_core.comm.signal() built, forwarded to subscribers verbatim. At-most-once and best-effort — no layer, no subscriber or a dead redis all mean the frame is dropped and nothing raises. Emit with comm.signal(); call this directly only when wiring a transport by hand.
- deliver_frame — stapel_realtime.delivery.deliver_frame
  instead of: a module's own broadcast_message() over the channel layer
  Fan out one JOURNAL frame (carrying the persisted row's seq) to a resumable stream, after the row is committed. Use this — not deliver() — whenever the frame corresponds to a row a client could replay: the seq is what lets ResumableStreamConsumer deduplicate the replay/live overlap.
- error_frame — stapel_realtime.envelope.error_frame
  Build the `error` envelope — a machine `code` (bad_envelope / bad_type / unauthorized) plus a human `message`. Refusals only; the resume-gap verdict is a `resync` FRAME, not an error, because re-hydrating is a normal instruction and the socket stays open.
- frame — stapel_realtime.envelope.frame
  instead of: a hand-written dict literal sent with send_json
  Build a v1 wire envelope {v, type, payload, seq?, stream?}. Every frame on a Stapel socket goes through this, so the version and the optional-field discipline (no seq on ephemeral frames) hold everywhere.
- group_name — stapel_realtime.streams.group_name
  Translate a stream key into its Channels group name (':' is not a legal group character). Long keys fold into a digest rather than being truncated, because truncation would make two streams share one group. Only a host writing its own transport needs this; deliver()/revoke() call it for you.
- normalize_origin — stapel_realtime.asgi.normalize_origin
  Canonicalize an Origin header or allowlist entry to scheme://host[:port], lower-cased with the scheme's default port elided. The port is identity: an allowlist entry of 'studio.localhost' never matching 'http://studio.localhost:8600' is the incident the origin guard and realtime.E003 exist for.
- open_stream — stapel_realtime.testing.open_stream
  instead of: a hand-wired channels.testing.WebsocketCommunicator with a hand-built scope
  Connect a test socket to a consumer with an already-authenticated scope, returning a StreamClient that speaks envelopes and swallows heartbeat noise. A module testing its consumer should use this instead of wiring a WebsocketCommunicator by hand — with expect_accept=False it is also how a fail-closed authorize is asserted.
- parse_frame — stapel_realtime.envelope.parse_frame
  Validate an inbound envelope into a Frame dataclass, raising InvalidEnvelope rather than returning a half-trusted dict. Note the asymmetry it encodes: an unknown frame TYPE parses (the consumer answers bad_type), an unknown envelope VERSION does not (the shape underneath is unknown).
- parse_stream_key — stapel_realtime.streams.parse_stream_key
  Parse a canonical key into its module/scope_type/scope_id/topic parts, raising InvalidStreamKey on anything else. This is how an authorizer learns which workspace it is being asked about — a consumer that cannot parse its own stream key must close the socket, not guess a scope.
- record_connect — stapel_realtime.presence.record_connect
  Count one live session for a user, keyed by the socket's channel name. The base consumer calls it after accept(); write your own only if you serve a socket this library's consumers do not. Two tabs are two sessions and one person.
- record_disconnect — stapel_realtime.presence.record_disconnect
  Stop counting one session. The document survives with an empty session map so last_seen stays answerable for one more TTL; presence is a liveness lease, not a history.
- record_heartbeat — stapel_realtime.presence.record_heartbeat
  Renew a session's presence lease. This is what makes a crashed worker stop counting — nothing runs its disconnect, and one PRESENCE_TTL_S later the session is simply not live. Which is why HEARTBEAT_S must stay below that TTL (realtime.W006).
- register_transport — stapel_realtime.delivery.register_transport
  Register deliver() into the core's signal seam under the name 'channels'. Called from this package's AppConfig.ready(); a host that composes app config by hand and wants the axis to resolve calls it itself. Registration is not activation — the host still selects the transport, and the default stays 'none'.
- site_registry_origins — stapel_realtime.asgi.site_registry_origins
  instead of: re-listing the fleet's brand hosts in STAPEL_REALTIME['ALLOWED_ORIGINS']
  https://<host> for every host and alias in the stapel_core.sites registry — the origins OriginGuard unions into its allowlist so one image answers every registered brand host without a hand-list per socket. Empty when the registry is empty, broken (reported by stapel_core.sites.E001, not here) or the installed core predates it. Call it when assembling an allowlist by hand; the guard and realtime.W002 already read it.
- workspace_stream — stapel_realtime.streams.workspace_stream
  Shorthand for the common shape, <module>:ws:<workspace_id>[:<topic>] — the key form the WorkspaceCapability authorizer understands.

## Extension points — what a product replaces, fork-free
- ChannelsSignalTransport [transport]
  The v1 signal-delivery backend the core's STAPEL_COMM['SIGNAL_TRANSPORT'] axis names. Callable and .send()-able, so either resolution convention in the core works. Swap it for a dotted path to your own object to move Signal onto a different bus.
- EphemeralStreamConsumer [base_class]
  Subclass for Signal delivery: set module/scope_type/stream_key_kwarg (or override get_stream_key) plus an authorizer, and the socket is done. No hooks required — there is no journal.
- OriginGuard [middleware]
  The ASGI origin allowlist build_websocket_application() installs. Instantiable directly with explicit `allowed_origins` when a host composes its own stack.
- ResumableStreamConsumer [base_class]
  Subclass for a journal with catch-up. Two async hooks: get_server_seq() and get_replay_rows(after_seq, limit) -> Sequence[JournalRow]. The welcome, the bounded replay, the resync verdict and the replay/live dedup are the base class's.
- authorize() [required_hook]
  The per-stream authorization seam on every consumer. FAIL-CLOSED: a subclass that does not override it (or set `authorizer`) subscribes nobody. Set `authorizer = WorkspaceCapability('<mod>.read')` for a ws-scoped stream, or write an async (scope, stream_key) -> bool of your own (MODULE.md 'Extension points').
- stapel_realtime.testing [test_harness]
  open_stream()/StreamClient — the envelope-aware Channels test client a module uses to test its own consumer, instead of the fourth hand-wired WebsocketCommunicator.

## Fits with — fleet dependencies
- stapel-core (required) — AppSettings config layer (conf.py), the G14 Channels JWT middleware reused by build_websocket_application and its close-code constant (asgi.py, close_codes.py), and require_capability for the workspace authorizer (authorize.py). The non-fleet dependencies are extras: `channels` to SERVE a socket (the envelope, stream keys and the delivery seam work without it — delivery degrades to a silent no-op, which is the contract) and `channels-redis` for cross-replica fan-out (more than one worker without it is realtime.E001).
