@panphora/hyper-wire

Stateless local message bus. Routes opaque JSON envelopes between SSE subscribers and HTTP senders on named channels. Zero dependencies.

npm install @panphora/hyper-wire

Built for Hyperclay™ Local: HTML pages publish and subscribe to channels, and small user-run handler scripts (holding their own secrets and OS access) do the same from the other side. The bus executes nothing, stores nothing, and knows nothing about payloads: capability lives entirely in the handlers the user chooses to run.

What it is not

• Not a router and not auth. The host wires the HTTP endpoints, SSE headers, keep-alive, body limits, and any origin or host checks. hyperclay-local, for example, adds a loopback Origin check on send and a Host allowlist on both routes.
• Not persistent. Best-effort delivery: no acks, no retries, no replay. If nobody is subscribed, the message evaporates (send() returns 0).
• Not filtered. A sender subscribed to its own channel receives its own messages and filters by sender id.

Protocols built on top must be self-correcting: a final message carries the whole truth, so lost intermediate messages cannot corrupt state.

API

const { messageBus, isValidChannel } = require('@panphora/hyper-wire');

messageBus.subscribe(channel, res);    // res: an SSE-ready ServerResponse
messageBus.unsubscribe(channel, res);
messageBus.send(envelope);             // → number of subscribers delivered to
messageBus.getStats();                 // → { channels, connections }
isValidChannel(name);                  // → boolean, /^[a-z0-9/_-]{1,64}$/

subscribe() and send() throw a TypeError on an invalid channel name, and send() also throws on an empty type. Validation is the only thing that throws; everything after it is best-effort.

send() stamps a monotonic seq, defaults v to 1, and passes everything else through opaque:

{
  "channel": "ai-edit",
  "type": "ai-edit/delta",
  "v": 1,
  "payload": { "id": "req-3", "index": 12, "text": "<p>…" },
  "sender": "page-a1b2",
  "origin": "hyperclay.html",
  "seq": 1751400000000
}

origin is advisory transport metadata supplied by the host, never authentication.

Dead connections are cleaned up centrally: any subscriber whose write throws is removed, and send() counts only successful writes. A channel emptied by unsubscribe or by that cleanup is dropped from the registry. getStats() reports the live channel count and total subscriptions, so one response on two channels counts as two connections.

Wires

A wire is a process's connection to a bus: it owns a sender identity, stamps it on outgoing envelopes, and filters its own echoes on the way in. Both implementations expose the same interface — send({channel, type, payload}) and subscribe(channel, onEnvelope) → unsubscribe() — so protocol code runs unchanged in-process or remote:

const { messageBus, localBus, httpBus } = require('@panphora/hyper-wire');

const inProcess = localBus(messageBus);                          // inside the host server
const remote = httpBus('http://localhost:4321/_/bus');           // a standalone process

httpBus.subscribe reconnects forever (2s backoff) and accepts {onConnect, onRetry} callbacks as a third argument.

httpBus requires two things from the host it points at: GET /subscribe?channel=<name> must stream SSE, and POST /send must reply with JSON { "delivered": <number> }. httpBus.send() returns that number and throws on any non-2xx response.

serve() — streamed request/reply

The handler-side kit for the one protocol pattern the bus was built to carry: a request streams back as deltas and finishes with exactly one terminal frame. Any wire plus an onRequest function becomes a handler:

const { serve, httpBus } = require('@panphora/hyper-wire');

serve(httpBus(), 'ai-edit', async (payload, reply, signal) => {
  reply.delta('<p>str');                 // batched ~50ms, indexes ordered
  reply.delta('eamed</p>');
  await reply.done({ html: '<p>streamed</p>' });   // or reply.error(message)
});

Conventions (all types namespaced under the channel): <channel>/request {id, …} in; <channel>/ack {id} out immediately on receipt, even if queued; <channel>/delta {id, index, text} out; one terminal <channel>/done or <channel>/error; <channel>/cancel {id} in aborts the request's signal and is answered with silence.

A handler that throws, returns without replying, or outlives timeoutMs becomes an error frame — a stuck handler can't strand the requester. Options: maxConcurrent (2, excess requests queue FIFO but ack immediately), timeoutMs (120000), batchMs (50), subscribeOptions (passed through to wire.subscribe). Returns { close() }.

Used by hyperclay-local's built-in plugins (over localBus) and by standalone handlers like hyperclay-pages' ai-edit handler (over httpBus) — same shape, different host.

Wire format

One SSE data: frame per envelope. curl is a first-class participant:

curl -N 'http://localhost:4321/_/bus/subscribe?channel=ai-edit'

curl -X POST 'http://localhost:4321/_/bus/send' \
  -H 'Content-Type: application/json' \
  -d '{"channel":"ai-edit","type":"ai-edit/request","payload":{"id":"req-1"},"sender":"curl"}'

serve() handlers require payload.id and silently drop a request without one. The bus itself still reports it as delivered, so an id-less request looks successful while nothing runs.

Test

npm test