Cross-process gateway event bus.
The gateway's live event log + SSE fan-out (gateway.state) is a
PROCESS-LOCAL in-memory registry: append-event! only reaches
subscribers inside the SAME JVM. That is why a turn streaming in the
TUI process is invisible to a web/Telegram process watching the SAME
conversation — each process has its own registry, and the only thing
they share is the persisted DB (which lands whole turns, not the live
token stream). So two watchers never stream together.
This bus closes that gap with the simplest transport that needs no
schema change and no always-on daemon: a shared append-only journal
under ~/.vis/gateway/events/<sid>.ndjson. Every LOCALLY-produced
gateway event is publish!ed there (one JSON line, tagged with this
process's producer id). A background tailer in each process follows
those files and re-delivers FOREIGN events (producer != self) into the
local registry via a delivery fn wired by gateway.state — so every
process's subscribers see the same stream, live.
Ordering/seq: exactly ONE turn runs per session at a time, so at any
moment a single producer owns the stream and its monotonic :seq is
authoritative for every watcher. The producer truncates the journal at
each turn.started, bounding a file to one turn's worth of deltas;
consumers detect the truncation (offset past EOF) and rewind.
Degrades safely: any IO failure is swallowed and the process falls back to today's in-process-only behavior.
Cross-process gateway event bus. The gateway's live event log + SSE fan-out (`gateway.state`) is a PROCESS-LOCAL in-memory registry: `append-event!` only reaches subscribers inside the SAME JVM. That is why a turn streaming in the TUI process is invisible to a web/Telegram process watching the SAME conversation — each process has its own registry, and the only thing they share is the persisted DB (which lands whole turns, not the live token stream). So two watchers never stream together. This bus closes that gap with the simplest transport that needs no schema change and no always-on daemon: a shared append-only journal under `~/.vis/gateway/events/<sid>.ndjson`. Every LOCALLY-produced gateway event is `publish!`ed there (one JSON line, tagged with this process's `producer` id). A background tailer in each process follows those files and re-delivers FOREIGN events (producer != self) into the local registry via a delivery fn wired by `gateway.state` — so every process's subscribers see the same stream, live. Ordering/seq: exactly ONE turn runs per session at a time, so at any moment a single producer owns the stream and its monotonic `:seq` is authoritative for every watcher. The producer truncates the journal at each `turn.started`, bounding a file to one turn's worth of deltas; consumers detect the truncation (offset past EOF) and rewind. Degrades safely: any IO failure is swallowed and the process falls back to today's in-process-only behavior.
(forget! sid)Drop a session's journal (on session close). Never throws.
Drop a session's journal (on session close). Never throws.
(hydrate! sid)Replay a session's CURRENT journal into this process's registry NOW, so a
watcher subscribing mid-turn sees the turn running in a SIBLING process from
its turn.started — not just the deltas that happen to arrive after it
connects.
The producer truncates the journal at each turn.started, so a file holds
exactly one turn. We replay it ONLY while that turn is still running (no
terminal event yet): a finished turn is already covered by the durable DB +
normal history, and re-streaming it would double-render a completed answer.
Rewinds this process's tail cursor to the file's current end first, so the background tailer won't re-deliver what we hand over here. Never throws.
Replay a session's CURRENT journal into this process's registry NOW, so a watcher subscribing mid-turn sees the turn running in a SIBLING process from its `turn.started` — not just the deltas that happen to arrive after it connects. The producer truncates the journal at each `turn.started`, so a file holds exactly one turn. We replay it ONLY while that turn is still running (no terminal event yet): a finished turn is already covered by the durable DB + normal history, and re-streaming it would double-render a completed answer. Rewinds this process's tail cursor to the file's current end first, so the background tailer won't re-deliver what we hand over here. Never throws.
Stable per-process id: foreign events carry a different one.
Stable per-process id: foreign events carry a different one.
(publish! sid event)(publish! sid event {:keys [store? truncate?]})Append one locally-produced event to the shared journal for sid.
truncate? (true on turn.started) resets the file first, bounding it
to the current turn. Never throws.
Append one locally-produced `event` to the shared journal for `sid`. `truncate?` (true on `turn.started`) resets the file first, bounding it to the current turn. Never throws.
(set-deliver-fn! f)Register the fn the tailer calls for every FOREIGN event:
(f sid event store?). Wired by gateway.state.
Register the fn the tailer calls for every FOREIGN event: `(f sid event store?)`. Wired by `gateway.state`.
(start!)Start the background tailer once. Idempotent.
Start the background tailer once. Idempotent.
cljdoc builds & hosts documentation for Clojure/Script libraries
| Ctrl+k | Jump to recent docs |
| ← | Move to previous article |
| → | Move to next article |
| Ctrl+/ | Jump to the search field |