Providers and transports

Import authoring helpers from focused subpaths and lifecycle helpers from the Runtime surface:

import {
  classifySseHttpStatus,
  classifyWebSocketCloseCode,
  createBoundedPushBuffer,
  polling,
  sse,
  sseHttpStatusErrorCode,
  stream,
  webhook,
  websocket,
  webSocketCloseErrorCode,
  type PollingTransport,
  type SseItem,
  type SseTransport,
  type StreamItem,
  type StreamTransport,
  type WebhookTransport,
  type WebSocketItem,
  type WebSocketTransport,
} from "@use-crux/core/signal/transport";
import {
  managedTransportBinding,
  signalProvider,
  type SignalProvider,
} from "@use-crux/core/signal/provider";
import {
  acceptTransportEnvelope,
  createTransportNormalizationRunner,
  ManagedStreamTerminalError,
  projectTransportBindingHealth,
  projectTransportEnvelope,
  replayTransportEnvelope,
  transportBindingHealth,
  transportStatistics,
  type RuntimeAcceptedTransportEnvelope,
  type RuntimeManagedTransportBinding,
  type RuntimeTransportBindingCheckpoint,
  type RuntimeTransportBindingHealth,
  type RuntimeTransportBindingHealthSnapshot,
  type RuntimeTransportEnvelopeProjection,
  type RuntimeTransportEnvelopeRecord,
  type RuntimeTransportEnvelopeState,
} from "@use-crux/core/runtime";

Provider and binding helpers also re-export from the root package where documented.

Authoring

webhook({ handle })

Returns a frozen WebhookTransport (_tag: "WebhookTransport", kind: "webhook").

handle(request) must authenticate and validate before returning:

polling({ poll, intervalMs? })

Returns a frozen PollingTransport (_tag: "PollingTransport", kind: "polling").

poll({ cursor, signal, configRef }) acquires one bounded page:

Optional intervalMs is a positive minimum between poll starts for one binding when more is not pending. When omitted, the Runtime worker maintenance cadence is the poll cadence.

The single Runtime worker leases the binding, calls poll, accepts each event through acceptTransportEnvelope(), and persists nextCursor only after the full batch is accepted, treated as a same-digest duplicate, or progressable after TransportEnvelopeConflictError when durable envelope-store evidence for that identity is confirmed. Other accept failures keep the previous cursor. Partial failure or abort also leaves the previous cursor. Restart resumes from the durable checkpoint without inventing a second worker. signal aborts on worker stop or lease release.

stream({ open })

Returns a frozen StreamTransport (_tag: "StreamTransport", kind: "stream"). This is the managed ingress transport constructor on @use-crux/core/signal/transport. It is distinct from LLM generation stream() helpers on provider packages and @use-crux/ai.

open({ cursor, signal, configRef }) opens one connection and returns an AsyncIterable<StreamItem> (or a Promise of one). Runtime owns lease, fiber, reconnect, accept, and checkpoint. Credentials and live clients stay inside the open closure.

StreamItem

Exactly one protocol item per yield — never a batch:

Cursor law

A yielded cursor asserts that the adapter has finished all provider input through that opaque position. Runtime never advances an envelope cursor before accept resolves for that item. Crash between accept and checkpoint redelivers the same provider event id; the #337 kernel dedupes.

Lifecycle

sse({ open })

Returns a frozen SseTransport (_tag: "SseTransport", kind: "sse"). This is the managed provider-ingress SSE transport on @use-crux/core/signal/transport.

It is not:

• LLM generation stream() helpers on provider packages / @use-crux/ai
• @use-crux/react browser SSE (createSSETransport / cruxSSEHandler), which is egress from Crux RecordStore/state to the browser

Runtime lowers SSE items onto the managed stream fiber. There is no second reconnect loop, worker, checkpoint schema, or Runtime-owned fetch / EventSource / frame parser. Credentials and HTTP clients stay inside the open closure; inert bindings and program JSON remain secret-free.

open({ cursor, signal, configRef }) opens one SSE connection and returns an AsyncIterable<SseItem> (or a Promise of one). The open context field is still named cursor — for SSE adapters it is the durable Last-Event-ID resume value (or null). Map non-null cursor to the HTTP Last-Event-ID request header when connecting.

SseItem

Exactly one protocol item per yield — never a batch:

lastEventId is progress through this event inclusive (wire id: / Last-Event-ID after the event). Omitted means no new resume position. null clears the durable resume position only when the provider truly has none. Cursor-only items must never cover unyielded events. Validation reuses the canonical stream cursor contract (byte limit, non-empty trimmed, no ASCII controls).

Pure HTTP status helpers

Core does not perform HTTP. After an adapter observes a failed connect status:

Throw ManagedStreamTerminalError(code, message?) for terminal classifications; throw an ordinary Error for transient. Non-HTTP / network failures before a status remain ordinary errors (transient reconnect). 2xx is success — do not call these helpers for successful connects. Content-Type mismatches after 2xx are adapter policy.

Lifecycle

EOF, transient throw, terminal fault, abort, config invalidation, pull backpressure, accept-before-checkpoint, and no required post-accept ack all reuse the stream({ open }) lifecycle table above.

websocket({ open })

Returns a frozen WebSocketTransport (_tag: "WebSocketTransport", kind: "websocket"). This is the managed provider-ingress WebSocket transport on @use-crux/core/signal/transport.

Runtime lowers WebSocket items onto the managed stream fiber. There is no second reconnect loop, worker, checkpoint schema, or Runtime-owned socket API. Credentials, sockets, ping/pong, and wire codecs stay inside the open closure; inert bindings and program JSON remain secret-free.

open({ cursor, signal, configRef }) opens one WebSocket connection and returns an AsyncIterable<WebSocketItem> (or a Promise of one). Must honor signal and clean up sockets on abort / iterator return.

WebSocketItem

Exactly one protocol item per yield — never a batch:

Optional post-accept acknowledge

Ordinary receive-only WebSocket ingress omits acknowledge. When a provider requires an application ack only after durable progress, attach acknowledge: () => void | Promise<void> on the envelope item.

Runtime invokes it only after durable #337 accept (or same-digest duplicate) and, when cursor is present, after that cursor is successfully checkpointed (or checkpoint is skipped because the store port is absent).

If acknowledge throws:

• acceptance remains accepted (or remains a same-digest duplicate);
• the durable cursor is not cleared or rolled back;
• Runtime writes lastErrorCode (TRANSPORT_ACK_FAILED or a safe provider code) on an active checkpoint;
• the connection outcome is transient so reconnect resumes from the durable cursor; provider redelivery is #337-deduped.

Bounded push buffering

Use createBoundedPushBuffer({ capacity, signal }) (or an equivalent bound). There is no unlimited mode. Overflow fails the consumer with TRANSPORT_PUSH_BUFFER_OVERFLOW instead of dropping messages so Runtime can reconnect from the durable cursor.

Pure close-code helpers

Map "normal" to clean iterator completion, "transient" to an ordinary Error, and "terminal" to ManagedStreamTerminalError.

Lifecycle

EOF, transient throw, terminal fault, abort, config invalidation, pull backpressure, and accept-before-checkpoint reuse the stream({ open }) lifecycle table above.

signalProvider({ id, transport, signals, onEvent })

Returns a frozen SignalProvider. Live definitions retain transport and onEvent as process code. transport may be webhook(), polling(), stream(), sse(), or websocket(). They perform no I/O at construction time and do not register globally.

onEvent(envelope, { signals }) may publish only the declared Signal map. Omitted publish idempotency keys default to the accepted provider/account/event identity when invoked through the transport normalization runner.

managedTransportBinding(provider, options)

Projects a live provider into an inert RuntimeManagedTransportBinding:

The result is suitable for immutable Runtime program generation. It never captures handle, poll, open, onEvent, credentials, or clients.

Acceptance and normalization

acceptTransportEnvelope(options)

Call only after edge authentication succeeds.

Catch only TransportEnvelopeConflictError for conflict responses and rethrow other errors. Conflict is never a resolved acknowledge: false result.

Envelope lifecycle states

RuntimeTransportEnvelopeState:

createTransportNormalizationRunner({ store, namespace, providers })

Claims a bounded batch and runs provider normalization. Hosts typically invoke runOnce() from the single Runtime worker maintenance path.

replayTransportEnvelope(options)

Returns a dead-letter envelope to accepted. Automatic workers never invent replay.

Observability

transportStatistics({ store, namespace })

Returns TransportEnvelopeStats:

• exact totals: accepted, deduplicated, normalized, delivered, retried, deadLettered
• first-64 byIdentity attribution keyed by transportStatisticsIdentity(adapterId, bindingId) (structured JSON pair, unambiguous when either id contains /)
• overflow identities roll into otherIdentities with identityAttribution: "truncated"

Statistics use the shared statistics ledger export persisted on the transport store. They are restart-safe where the Runtime store is durable.

projectTransportEnvelope(record)

Privacy-safe operator view: identity, state, attempts, safe lastFailure, config ref ids, target Signal id, and lineage entries of { signalId, occurrenceId }.

Projections omit payload bytes, payload refs, credentials, and raw routing secrets. Occurrence identities join existing Signal / Work / Session / Flow observability without a parallel execution registry.

When observability sinks are active, accept and normalize also emit a custom.operation run whose attributes carry { kind: "transport.envelope", outcome, envelope } so Devtools Run detail can render accepted-envelope lineage without a second event store.

transportBindingHealth({ store, namespace, program, now? })

Bounded, restart-safe per-binding health snapshot for operators and Devtools. Derives only from the immutable Runtime program, durable binding checkpoints, and the existing transport statistics ledger — no second registry or metrics store.

Each binding row includes:

At most 64 bindings are projected (coverage.bindings: "truncated" beyond the bound). Missing checkpoint methods, empty statistics ledgers, and unsupported facets are reported with explicit coverage — never fabricated counters.

projectTransportBindingHealth(options)

Pure projection of one binding health row from already-loaded durable facts. Useful in tests and hosts that already hold checkpoints and statistics.

Retention

Runtime retention accepts transportEnvelopes (default 7d, or false to keep forever). Only terminal normalized and dead-letter rows are eligible. Accepted and claimed envelopes are never pruned by retention.

Binding checkpoints

Polling and stream supervision store opaque cursors through the transport store:

Checkpoints never retain credentials, clients, sockets, Requests, raw payloads, or unbounded event identity lists. Lease ownership uses the existing Runtime lease port with resource transport-binding:{namespace}:{bindingId}. putBindingCheckpoint is lease-fenced (accepted / rejected).

Deployment requirements

• Memory or PostgreSQL Runtime storage with the transports port, including getBindingCheckpoint / putBindingCheckpoint (PostgreSQL table transport_binding_checkpoints via store setup).
• Convex Runtime storage does not implement managed-transport accept, checkpoints, or statistics. Worker start with managed bindings fails with CAPABILITY_MISSING rather than fabricating transport state.
• createRuntimeProgram({ providers, transports }) with live providers in process and inert bindings in the program.
• createRuntimeWorker({ runtime, program }) as the only supervised owner of polling and stream acquisition, envelope drain, and Work maintenance.
• Shutdown: worker stop aborts in-flight poll/open via AbortSignal, releases binding leases, and preserves the last durable checkpoint for restart recovery. Process-local reconnect attempt history is not durable; see health coverage.

Guarantees and non-goals

• No second transport daemon, queue, or global provider registry.
• No Channel exclusive conversation ownership on this path (#302).
• Webhook edge accept remains host-driven; polling and managed stream are worker-supervised.
• Provider-ingress sse({ open }) and websocket({ open }) ship as thin adapters over the managed stream fiber on this path.
• Competing supervisors coordinate through Runtime leases on Memory and PostgreSQL.