Streams and channels

A stream is a single-consumer sequence that may produce values over time. A channel is a named, session-scoped event bus with a bounded replay ring. They share value methods but solve different ownership problems: streams model pull-driven flow, while channels decouple publishers and subscribers.

Lifecycle first🔗

flowchart LR
accTitle: Lifecycle first
accDescr: Shows the components and relationships described in Lifecycle first.
    SRC["source producer"] -->|"bounded queue"| S["stream<T>"]
    S --> C["lazy combinators"]
    C --> K{"explicit sink"}
    K --> E["each"]
    K --> L["collect"]
    K --> R["render"]
    K --> F["save / append"]
    K --> I["into channel"]

Streams are consumed once. A second attempt raises stream_consumed. A stream can be bounded (known to end) or unbounded. Materializing an unbounded stream raises stream_unbounded instead of buffering forever.

A bare stream at the REPL renders a descriptor. It does not automatically start a live view:

every(1s)                    # stream descriptor
every(1s).take(5).render()   # drives five values

Timer source🔗

every(interval) starts a timer producer and returns stream<datetime>:

every(1s).take(5).collect()
every(250ms).take(20).each(t => echo (t))

The interval must be positive. The producer has a one-slot buffer. If the consumer is slow, additional ticks are dropped/coalesced with no marker; memory remains constant. The retained slot is the earliest undelivered tick, so its timestamp can be up to roughly one interval stale when the consumer resumes.

Construction currently starts the producer thread immediately. “Lazy” describes downstream pulling and combinators, not deferred creation of every underlying OS resource.

Watch filesystem events🔗

watch(path("./src"), recursive: true)
  .where(.kind == "modified")
  .take(10)
  .collect()

watch accepts a path or glob and returns events:

{ path: path, kind: "created" | "modified" | "removed", ts: datetime }

A glob watches its literal directory prefix and filters full paths. The current consumer queue holds 64 events. When a burst overflows it, events are dropped and a summary event is owed:

{ path: watched_root, kind: "modified", ts: datetime, coalesced: true }

Consumers that require an exact audit trail must rescan authoritative state after coalesced: true; a watch stream is a notification channel, not a durable journal.

Tail a file🔗

tail(path("./service.log"))
  .where(.contains("ERROR"))
  .take(20)
  .collect()

tail(file, from_start: false) follows complete lines. It begins at EOF by default or byte zero with from_start: true, tracks truncation/rotation when the file shrinks, and uses native filesystem notifications.

Its queue also holds 64 items. Dropped lines are represented by a marker value:

{ dropped: 37 }

That means the stream can contain either strings or marker records under pressure. Branch on shape instead of assuming every element is text.

Lazy combinators🔗

Current stream-specific combinators are:

Every combinator consumes its input stream and returns a new one. Assign the new stream, not the old handle:

let filtered = watch(path("./src")).where(.kind == "modified")
filtered.take(10).render()

distinct is a deliberate memory tradeoff: source buffers remain bounded, but the set of every distinct value seen is inherently unbounded. Prefer dedupe or a bounded window when long-lived cardinality is unknown.

Sinks🔗

Sinks drive a stream until it ends, errors, or is canceled:

stream.each(item => handle(item))
stream.collect()
stream.render()
stream.save(path("events.ndjson"))
stream.append(path("events.ndjson"))
stream.into(channel("user.events"))

collect() rejects an unbounded stream. each returns null after completion. render sends each value to the evaluator’s statement sink.

For streams, both .save(path) and .append(path) currently open the file in append mode and write one line per item. Strings and bytes are written as their content; other values become JSON per line. .save does not truncate, despite its name—this is an important preview behavior.

buffer(n) currently type-checks and documents intent, but it does not create a queue, background producer, or pacing boundary. Shoal’s stream runtime is synchronously pulled today, so the operation returns the same stream unchanged.

tee(n) returns independently drivable streams. A bounded stream materializes once for exact replay. A live stream uses a queue of at most 64 pending items per fork. When a slow fork falls behind, overflowed values are dropped and later represented in order by a {dropped: n} marker. The marker appears as soon as that fork’s queue has room, or after its buffered items drain; overflow does not raise an error and is never silent.

Streams cannot feed processes yet🔗

The finite-value feed serializer rejects streams. Incremental stream-to-child-stdin is not wired:

# current workaround
let batch = source.take(100).collect()
batch.feed(^consumer)

Likewise, the kernel wire does not currently expose a general stream-ref chunk-pull protocol. Keep live consumption inside language code or bridge through user.* channels.

Named channels🔗

Create a handle with channel(name):

let updates = channel("user.builds")
updates.emit({ id: 42, state: "started" })
updates.latest()
updates.events()
updates.take(timeout: 5s)

Channel methods:

Event records have a common shape:

{ channel: str, seq: int, ts: datetime, payload: value }

Sequence numbers are monotonic per channel. Each channel retains its most recent 1,024 events. events() replays the current ring then goes live; events(since: n) replays only records with seq > n. If a cursor is older than the ring, evicted history cannot be recovered from the channel; persist it separately.

take() subscribes without replay, so it sees only an event published after the call. A timeout raises timeout rather than returning null.

Handlers and stream-to-channel bridges🔗

let task = on(channel("user.builds"), event => {
  echo (event.payload)
})

watch(path("./src"))
  .into(channel("user.files"))

on(channel_or_name, handler) subscribes before spawning its handler task so the startup gap does not lose an event. It is equivalent in intent to spawning .events().each(handler).

There is no on channel(...) { ... } keyword form; on(...) is a function call in the current grammar.

Kernel bridge🔗

In a kernel-hosted session, only channels beginning with user. bridge bidirectionally between language code and the external event bus. This prevents language code from spoofing kernel-owned semantic channels such as approvals, journal notifications, or session transcript events.

flowchart LR
accTitle: Kernel bridge
accDescr: Shows the components and relationships described in Kernel bridge.
    L["language channel user.x"] <-->|"mirrored payloads"| K["kernel EventBus user.x"]
    K <-->|"events subscribe/read"| A["agent client"]
    OWN["kernel-owned channels"] --> K
    L -. "cannot publish as kernel-owned" .-> OWN

The local standalone REPL has no external forwarder; its channels remain session-local.

Cancellation and resources🔗

Dropping/finishing a stream usually disconnects its bounded receiver, causing timer/watch/tail producers to exit and release resources. A timer thread sleeping until its next interval notices disconnection only when it wakes and tries to send. Ctrl-C and task cancellation propagate through host cancellation paths, but code should still bound live workflows explicitly.

For durable event history use .save, journal-aware workflows, or an external store. Channels and source buffers intentionally trade completeness under overload for bounded memory.