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 another 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 another 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.
A non-terminal journal has TWO causes, told apart by "_pid" liveness: the
producer is still alive (a real in-flight sibling turn — mirror it) OR the
producer PROCESS is gone (a daemon crash/restart mid-turn). An orphaned turn
will never emit a terminal, so resurrecting it pins this process's
:current-turn to a dead turn — wedging the session queue (new sends pile up
queued, nothing drains) and spinning every watcher forever. For an orphan we
instead land a synthetic turn.failed so the queue drains, clients get
closure, and no later hydrate replays it again.
Runs under the sid's [[tail-lock]] and moves this process's tail cursor past the COMPLETE lines it read, so neither the background tailer nor a concurrent hydrate re-delivers 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. A non-terminal journal has TWO causes, told apart by `"_pid"` liveness: the producer is still alive (a real in-flight sibling turn — mirror it) OR the producer PROCESS is gone (a daemon crash/restart mid-turn). An orphaned turn will never emit a terminal, so resurrecting it pins this process's `:current-turn` to a dead turn — wedging the session queue (new sends pile up `queued`, nothing drains) and spinning every watcher forever. For an orphan we instead land a synthetic `turn.failed` so the queue drains, clients get closure, and no later hydrate replays it again. Runs under the sid's [[tail-lock]] and moves this process's tail cursor past the COMPLETE lines it read, so neither the background tailer nor a concurrent hydrate re-delivers what we hand over here. Never throws.
(journal-high-water-seq sid)Highest "seq" persisted in sid's journal file, or 0 when there is none.
A daemon restart resets its in-memory :seq counter to 0, but a client
(TUI) keeps its replay cursor as a monotonic MAX across reconnects — so
events from the fresh daemon (seq 1, 2, …) fall UNDER the client's stale
cursor and its seq > cursor replay filter silently drops the whole new
turn (the orphan-reap terminal included). Seeding a fresh registry entry
from this high-water keeps the restarted daemon numbering ABOVE what the
client already saw. Never throws.
Highest `"seq"` persisted in `sid`'s journal file, or 0 when there is none. A daemon restart resets its in-memory `:seq` counter to 0, but a client (TUI) keeps its replay cursor as a monotonic MAX across reconnects — so events from the fresh daemon (seq 1, 2, …) fall UNDER the client's stale cursor and its `seq > cursor` replay filter silently drops the whole new turn (the orphan-reap terminal included). Seeding a fresh registry entry from this high-water keeps the restarted daemon numbering ABOVE what the client already saw. 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?] :as opts})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. Transient (store? false) live deltas are
queued to a writer thread so ndjson I/O cannot stall the gateway hot path;
durable events are flushed before this fn returns.
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. Transient (`store? false`) live deltas are queued to a writer thread so ndjson I/O cannot stall the gateway hot path; durable events are flushed before this fn returns.
(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`.
(set-relevant-sid-fn! f)Register a predicate (f sid) -> truthy naming the sessions this process has a
LOCAL consumer for. The tailer skips draining journals for any other sid: a
foreign event for a session with no local registry entry is dropped by the
delivery fn anyway (gateway.state/ingest-mirrored-event! no-ops on an unknown
sid), so opening + reading + JSON-parsing — and even stat'ing — those journals
on every poll is pure waste, the CPU an otherwise-idle daemon burns re-scanning
every sibling's journal forever. Absent a wired predicate (tests, early boot)
every sid is relevant, preserving the drain-everything behavior. Wired by
gateway.state.
Register a predicate `(f sid) -> truthy` naming the sessions this process has a LOCAL consumer for. The tailer skips draining journals for any other sid: a foreign event for a session with no local registry entry is dropped by the delivery fn anyway (gateway.state/ingest-mirrored-event! no-ops on an unknown sid), so opening + reading + JSON-parsing — and even stat'ing — those journals on every poll is pure waste, the CPU an otherwise-idle daemon burns re-scanning every sibling's journal forever. Absent a wired predicate (tests, early boot) every sid is relevant, preserving the drain-everything behavior. Wired by `gateway.state`.
(set-relevant-sids-fn! f)Register a 0-arg fn returning the COLLECTION of sids this process has a local
consumer for. When wired, the tailer drains ONLY those sessions' journal files
directly — it never lists/stats the whole events dir on a poll, so an idle
daemon (empty set) does zero per-poll directory work. Falls back to the
relevant-sid? directory scan when this is not wired (tests).
Register a 0-arg fn returning the COLLECTION of sids this process has a local consumer for. When wired, the tailer drains ONLY those sessions' journal files directly — it never lists/stats the whole events dir on a poll, so an idle daemon (empty set) does zero per-poll directory work. Falls back to the `relevant-sid?` directory scan when this is not wired (tests).
(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 |