Signal providers

Accept authenticated provider webhooks, poll provider pages, or supervise managed streams, acknowledge only after durable acceptance, and fan out through ordinary Signals.

A Signal provider turns an external webhook, polled provider page, or managed async stream into the ordinary Signal path your application already understands. Use one when a third-party system must push or expose events into Crux, you need accept-before-checkpoint durability, and retries must not create a second logical delivery for the same provider event ID.

Transport status

Transport	Status
`webhook()`	Existing production path (#337 edge accept + worker normalize)
`polling()`	One Runtime worker leases, polls, accepts, checkpoints
`stream({ open })`	Managed async-stream seam: worker owns open, reconnect, accept-before-checkpoint, fault status
`sse({ open })`	Thin managed provider-ingress SSE adapter: `lastEventId` authoring lowers onto the stream fiber
`websocket({ open })`	Thin managed provider-ingress WebSocket adapter: optional post-accept `acknowledge`, bounded push buffering, close-code helpers
Channel exclusive conversation ownership	Not on this path; remains #302

Why providers exist

Publishing a Signal directly from a route handler works for trusted internal events. Provider webhooks are different:

authentication and size checks must fail before durable acceptance;
the host should acknowledge only after the envelope is committed;
crash recovery must resume normalization without losing the event;
duplicate provider deliveries must stay idempotent.

The managed-transport kernel owns that ingress contract. Normalization still publishes through your declared Signals, so Flow waits, Session subscriptions, and local callbacks keep their existing fan-out semantics.

Minimal webhook path

import { signal } from "@use-crux/core";
import { webhook } from "@use-crux/core/signal/transport";
import {
  managedTransportBinding,
  signalProvider,
} from "@use-crux/core/signal/provider";
import {
  acceptTransportEnvelope,
  createTransportNormalizationRunner,
} from "@use-crux/core/runtime";
import { z } from "zod";

const orderSubmitted = signal({
  id: "order.submitted",
  schema: z.object({ orderId: z.string() }),
});

export const orders = signalProvider({
  id: "orders.webhook",
  transport: webhook({
    async handle(request) {
      // Verify signature and enforce size limits before returning.
      const body = await request.json();
      return {
        accountId: body.accountId,
        eventId: body.eventId,
        authenticatedRouting: { source: "webhook" },
        payload: encodeInlinePayload(body),
      };
    },
  }),
  signals: { orderSubmitted },
  async onEvent(envelope, { signals }) {
    const orderId = decodeOrderId(envelope.payload);
    // Omit idempotencyKey: the transport runner defaults to a transport-scoped
    // provider/account/event key so crash recovery cannot double-deliver.
    await signals.orderSubmitted.publish({ orderId });
  },
});

// Inert declaration for Runtime Program generation — no credentials or live clients.
export const ordersBinding = managedTransportBinding(orders, {
  id: "binding.orders",
  configRef: { id: "config.orders", revision: "1" },
  signalId: "order.submitted",
});

Host edge shape:

Call transport.handle(request) for authentication and size validation.
Build a validated accepted envelope and call acceptTransportEnvelope().
Acknowledge the provider only when acceptance resolves with acknowledge: true. Catch TransportEnvelopeConflictError for 409; do not treat conflict as a successful ack.
Let the Runtime worker or createTransportNormalizationRunner() claim and normalize accepted envelopes after the response is sent.

When to use this

Use a Signal provider when:

a vendor webhook is the source of truth for an event identity;
you need crash-safe accept-before-ack;
many independent consumers should observe the same Signal at least once.

Do not use it when:

one conversation owner must exclusively claim the event — that is Channel work under related RFC tracks, not this Signal path;
you need a browser EventSource subscription for Crux UI state — that is @use-crux/react createSSETransport / cruxSSEHandler (egress), not provider-ingress sse({ open }).

Minimal polling path

import { polling } from "@use-crux/core/signal/transport";

export const ordersPoll = signalProvider({
  id: "orders.poll",
  transport: polling({
    intervalMs: 5_000,
    async poll({ cursor, signal }) {
      const page = await fetchProviderPage(cursor, signal);
      return {
        events: page.events.map((event) => ({
          accountId: event.accountId,
          eventId: event.eventId,
          authenticatedRouting: { source: "polling" },
          payload: encodeInlinePayload(event.body),
        })),
        nextCursor: page.cursor,
      };
    },
  }),
  signals: { orderSubmitted },
  async onEvent(envelope, { signals }) {
    await signals.orderSubmitted.publish({
      orderId: decodeOrderId(envelope.payload),
    });
  },
});

The single Runtime worker leases the binding, polls, durably accepts each event, and writes nextCursor only after the full batch is accepted. Restart resumes from the durable checkpoint. No second transport daemon is required.

Polling cursor, more, errors, and shutdown

Concern	Behavior
Cursor	Opaque `nextCursor` is written only after every event in the poll batch is durably accepted or a same-digest duplicate. Failure or abort leaves the previous cursor so restart redelivers without loss; #337 event IDs dedupe redelivery.
`more`	Optional `PollResult.more: true` means more pages are available immediately. After a successful batch, the next worker tick may skip `intervalMs` once.
Interval / backoff	Optional `intervalMs` is a minimum between poll starts when `more` is not pending. Poll failures store a safe `lastErrorCode` and do not advance the cursor. There is no automatic exponential backoff beyond `intervalMs` and the worker cadence.
Shutdown / restart	Worker stop aborts in-flight `poll` via `AbortSignal`, releases binding leases, and preserves the last durable checkpoint. A replacement worker resumes from that cursor.
Deployment	Declare providers + inert bindings on `createRuntimeProgram({ providers, transports })`, run `createRuntimeWorker({ runtime, program })` against Memory or PostgreSQL Runtime storage that implements the transports port (including binding checkpoints). Live `poll` / `onEvent` stay in process code; they never enter generated program JSON.

Minimal managed stream path

import { stream } from "@use-crux/core/signal/transport";
import {
  managedTransportBinding,
  signalProvider,
} from "@use-crux/core/signal/provider";

export const ordersStream = signalProvider({
  id: "orders.stream",
  transport: stream({
    async *open({ cursor, signal }) {
      const connection = await connectProvider({ cursor, signal });
      try {
        for await (const message of connection.messages) {
          yield {
            kind: "envelope",
            accountId: message.accountId,
            eventId: message.eventId,
            authenticatedRouting: { source: "stream" },
            payload: encodeInlinePayload(message.body),
            cursor: message.cursor,
          };
        }
      } finally {
        await connection.close();
      }
    },
  }),
  signals: { orderSubmitted },
  async onEvent(envelope, { signals }) {
    await signals.orderSubmitted.publish({
      orderId: decodeOrderId(envelope.payload),
    });
  },
});

export const ordersStreamBinding = managedTransportBinding(ordersStream, {
  id: "binding.orders.stream",
  configRef: { id: "config.orders.stream", revision: "1" },
  signalId: "order.submitted",
});

This is the managed ingress stream() on @use-crux/core/signal/transport, not LLM generation stream() helpers on provider packages or @use-crux/ai.

The Runtime worker leases the binding, opens one connection fiber, pulls items under serial backpressure, accepts each envelope through the same #337 kernel, and writes a cursor only after that item is durably accepted. Clean EOF and transient errors reconnect with bounded backoff. Terminal faults mark durable faulted status so restart does not silently reopen a dead connection.

Stream cursor, lifecycle, and recovery

Concern	Behavior
Cursor law	A yielded cursor means the adapter has finished all provider input through that position. Runtime checkpoints only after every envelope that position covers is durably accepted (or same-digest duplicate / progressable conflict with durable evidence).
Item shapes	Each yield is either `{ kind: "envelope", …, cursor? }` or `{ kind: "cursor", cursor }` — never a batch of envelopes. Cursor-only items may checkpoint immediately (heartbeats / caught-up markers).
Accept-before-checkpoint	Envelope cursors advance only after accept resolves. Crash between accept and checkpoint redelivers the same event id; #337 dedupe keeps Signal occurrence identity stable.
Config invalidation	Checkpoint stores `configRef`. When live binding `configRef` changes, Runtime over-invalidates the prior cursor (does not inherit faulted status from the stale config).
EOF	Clean iterator completion is disconnect, not terminal success. Runtime reconnects with bounded backoff from the durable cursor.
Transient error	Thrown errors reconnect by default (bounded backoff). Safe `lastErrorCode` may be stored without advancing the cursor.
Terminal fault	`ManagedStreamTerminalError` (or `{ terminal: true, code }`) sets durable `status: "faulted"` and stops automatic reconnect until config identity changes or an operator clears status.
Abort	Worker stop, lease loss, or rebalance aborts `signal`, calls `iterator.return?.()`, and prevents stale accept/checkpoint after the lease fence.
Backpressure	Pull iteration is Runtime backpressure: one item at a time on the fiber. Push adapters (WebSocket) must bound buffers and disconnect on overflow — never drop events silently. Prefer `createBoundedPushBuffer`.
Post-accept ack	Optional process-local `acknowledge` on an envelope item. Runtime invokes it only after durable accept (and cursor checkpoint when present). Ack failure is observable (`TRANSPORT_ACK_FAILED` / safe provider code) and transient; it never rolls back accept or clears the durable cursor.
Deployment	Same as polling: Memory or PostgreSQL with binding checkpoints, `createRuntimeProgram({ providers, transports })`, one `createRuntimeWorker`. Live `open` / `onEvent` / `acknowledge` stay process-local; generated program JSON never embeds closures or secrets.
SSE	Prefer `sse({ open })` when the external system is SSE (distinct Catalog/Index kind + `lastEventId` vocabulary). Lifecycle is this table via pure lowering.
WebSocket	Prefer `websocket({ open })` when the external system is WebSocket (distinct Catalog/Index kind, optional post-accept ack, push-buffer helper, close-code helpers). Lifecycle is this table via pure lowering.

Minimal managed SSE path

import {
  sse,
  classifySseHttpStatus,
  sseHttpStatusErrorCode,
} from "@use-crux/core/signal/transport";
import { ManagedStreamTerminalError } from "@use-crux/core/runtime";
import {
  managedTransportBinding,
  signalProvider,
} from "@use-crux/core/signal/provider";

export const ordersSse = signalProvider({
  id: "orders.sse",
  transport: sse({
    async *open({ cursor, signal, configRef }) {
      // Adapter owns fetch / EventSource / frame parsing — Core does not.
      const response = await connectOrdersSse({
        lastEventId: cursor,
        signal,
        configRef,
      });

      if (!response.ok) {
        // Release an unused body when present; null bodies are a no-op.
        await response.body?.cancel?.();
        const kind = classifySseHttpStatus(response.status);
        if (kind === "terminal") {
          throw new ManagedStreamTerminalError(
            sseHttpStatusErrorCode(response.status),
            `SSE connect rejected with HTTP ${response.status}`,
          );
        }
        throw new Error(`SSE connect transient HTTP ${response.status}`);
      }

      for await (const frame of parseSse(response.body, signal)) {
        if (frame.kind === "comment" && frame.id) {
          yield { kind: "cursor", lastEventId: frame.id };
          continue;
        }
        if (frame.kind === "event") {
          yield {
            kind: "envelope",
            accountId: frame.accountId,
            eventId: frame.eventId,
            authenticatedRouting: {
              source: "sse",
              eventType: frame.eventType ?? "message",
            },
            payload: encodeInlinePayload(frame.body),
            lastEventId: frame.id ?? undefined,
          };
        }
      }
    },
  }),
  signals: { orderSubmitted },
  async onEvent(envelope, { signals }) {
    await signals.orderSubmitted.publish({
      orderId: decodeOrderId(envelope.payload),
    });
  },
});

export const ordersSseBinding = managedTransportBinding(ordersSse, {
  id: "binding.orders.sse",
  configRef: { id: "config.orders.sse", revision: "1" },
  signalId: "order.submitted",
});

This is provider-ingress SSE on @use-crux/core/signal/transport. It is unrelated to @use-crux/react createSSETransport / cruxSSEHandler, which push Crux RecordStore state to the browser.

Runtime lowers lastEventId to the managed stream cursor contract and reuses the same stream fiber, lease, checkpoint, reconnect, fault, and abort laws as stream({ open }). Use the stream lifecycle table above for EOF, transient errors, terminal faults, abort, and backpressure. Prefer sse() when the external system is SSE and tooling should project transportKind: "sse"; prefer stream() when the adapter already speaks generic cursors.

Minimal managed WebSocket path

import {
  websocket,
  createBoundedPushBuffer,
  classifyWebSocketCloseCode,
  webSocketCloseErrorCode,
} from "@use-crux/core/signal/transport";
import { ManagedStreamTerminalError } from "@use-crux/core/runtime";
import {
  managedTransportBinding,
  signalProvider,
} from "@use-crux/core/signal/provider";

export const ordersWs = signalProvider({
  id: "orders.ws",
  transport: websocket({
    async open({ cursor, signal, configRef }) {
      // Adapter owns WebSocket connect / subscribe / ping-pong / close.
      const socket = await connectOrdersWs({ cursor, signal, configRef });
      const buffer = createBoundedPushBuffer({ capacity: 32, signal });

      socket.onmessage = (event) => {
        try {
          const message = mapFrame(event);
          buffer.push({
            kind: "envelope",
            accountId: message.accountId,
            eventId: message.eventId,
            authenticatedRouting: { source: "websocket" },
            payload: message.payload,
            cursor: message.cursor,
            // Optional: only after durable accept + cursor checkpoint.
            acknowledge: () => socket.ack(message.wireId),
          });
        } catch {
          socket.close();
          // Overflow already failed the buffer; Runtime reconnects from cursor.
        }
      };

      socket.onclose = (closeEvent) => {
        const kind = classifyWebSocketCloseCode(closeEvent.code);
        if (kind === "normal") {
          buffer.close();
          return;
        }
        if (kind === "terminal") {
          buffer.fail(
            new ManagedStreamTerminalError(
              webSocketCloseErrorCode(closeEvent.code),
              `WebSocket closed with code ${closeEvent.code}`,
            ),
          );
          return;
        }
        buffer.fail(new Error(`WebSocket closed with code ${closeEvent.code}`));
      };

      signal.addEventListener(
        "abort",
        () => {
          socket.close();
        },
        { once: true },
      );

      return buffer.items;
    },
  }),
  signals: { orderSubmitted },
  async onEvent(envelope, { signals }) {
    await signals.orderSubmitted.publish({
      orderId: decodeOrderId(envelope.payload),
    });
  },
});

export const ordersWsBinding = managedTransportBinding(ordersWs, {
  id: "binding.orders.ws",
  configRef: { id: "config.orders.ws", revision: "1" },
  signalId: "order.submitted",
});

Ordinary receive-only WebSocket ingress omits acknowledge. When a provider requires application acks only after durable progress, attach acknowledge on the envelope item — never before accept, and never as a rollback of accept. Push overflow closes/reconnects from the durable cursor; it never silently drops messages.

Guarantees

Guarantee	Behavior
Accept-before-ack	Hosts acknowledge only after durable acceptance commits.
Idempotency	Same provider/account/event identity with the same digest is a duplicate; a conflicting digest is rejected without mutation.
Fan-out	Normalization publishes ordinary Signals. Every matching independent subscription may receive the occurrence at least once.
Restart safety	Accepted envelopes and namespace transport statistics survive process restart in a capable Runtime store.
Privacy	Operator projections and statistics never retain raw payloads, credentials, or signature material.

Operator surfaces

transportStatistics({ store, namespace }) returns bounded accepted, deduplicated, normalized, delivered, retried, and dead-lettered totals with first-64 adapter/binding attribution.
transportBindingHealth({ store, namespace, program }) projects one secret-free health row per managed binding from program identity, durable checkpoints, and the same statistics ledger. Cursor age and last owner are available when checkpointed; live reconnect backoff and shutdown outcome stay explicitly unavailable when they are not durable. At most 64 bindings are returned. Memory and PostgreSQL implement the transports port; Convex does not claim managed-transport accept/checkpoints and rejects managed-binding workers with CAPABILITY_MISSING.
projectTransportEnvelope(record) exposes envelope state and Signal occurrence lineage without payload bytes. Accept/normalize also emit that projection on the existing observability transport for Devtools Run detail.
Devtools Catalog shows authored binding/adapter/config/target identity. Runtime status includes a Transports tab with live binding health when a generated program is present.
Runtime retention prunes only terminal normalized and dead-letter envelopes through the existing maintenance sweep (transportEnvelopes, default 7d).

Provider recipes for authentication, retries, dead letters, replay, and deployment.
Provider API reference for exact signatures and lifecycle states.
Signals guide for publication and durable Flow waits after the envelope becomes a Signal.