live

The live modes over the one planar comprehension: the same quoted [?for] that runs once as a query is also a delta feed. changes-since is the stateless core — it takes the quoted comprehension, a $bind map (formal source names → open store/journal handles), and a head-set cursor, and answers [changes [head-set …] ∂-frame…]: the edit script that carries a consumer from its cursor to head, plus the new cursor. Within the incremental sub-fragment the ∂ set is exact ([insert]/[retract]/[regroup]); outside it the answer is the honest [recompute reason=…] marker followed by the full relation as inserts — and from the empty cursor [head-set] the exact ∂ is the full relation as inserts for every comprehension, which is why a one-time [?for] and changes-since-from-empty agree (the equivalence quartet, leg 1). observe is the live form of the same answer: it returns a delivery-contract subscription ([live-sub …] — rung reported, retention declared, closeable) whose every [?receive] delivers the next [changes …] batch from the consumer's own cursor — observe ≡ repeated changes-since driven by source advance (leg 2), a fault terminating the stream loud as the final frame. The maintained fold (materialize) and its poll-substrate companions (advance/read) land with their implementation wave per the pack spec.

live:changes-since

[$live:changes-since] -> element — Answer the ∂ set from a head-set cursor to head over a quoted planar comprehension, plus the new cursor.

      [?lib 'cx-stdlib/store']
[?lib 'cx-stdlib/live']
[?let [= $s [$store:open "mem://live-doc"]]
[= $h [$store:put-doc $s [order [region "west"] [amount 100]]]]
  [$live:changes-since [?quote [?for [in $o [$store:source $orders "//order"]] [yield $o]]] {orders: $s} [head-set]]]

      [changes [head-set [s source=orders pos=1]] [insert pos=0 [order [region 'west'] [amount 100]]]]

live:observe

[$live:observe] -> element — Open a live ∂ subscription over a quoted planar comprehension: a delivery-contract handle whose receives deliver the [changes …] batches from the consumer's cursor to head.

      [?lib 'cx-stdlib/store']
[?lib 'cx-stdlib/live']
[?let [= $s [$store:open "mem://live-obs-doc"]]
[= $sub [$live:observe [?quote [?for [in $o [$store:source $orders "//order"]] [yield $o]]] {orders: $s} [head-set]]]
[= $h [$store:put-doc $s [order id=1]]]
  [?receive from=$sub]]

      [changes [head-set [s source=orders pos=1]] [insert pos=0 [order id=1]]]

live:materialize

[$live:materialize] -> element — Create (or re-attach to) a named, store-aliased, checkpointed fold of a quoted planar comprehension; the checkpoint is the durable cursor.

      [?lib 'cx-stdlib/store']
[?lib 'cx-stdlib/live']
[?let [= $s [$store:open "mem://live-mat-src"]]
[= $cs [$store:open "mem://live-mat-ckpt"]]
[= $h [$store:put-doc $s [order region=west amount=100]]]
[= $m [$live:materialize $cs [?quote [?for [in $o [$store:source $orders "//order"]] [group-by $o@region] [yield [rev region=$key n=$count]]]] {orders: $s} "rev-by-region"]]
  [$live:read $m]]

      [snapshot [head-set [s source=orders pos=1]] [rev region=west n=1]]

live:advance

[$live:advance] -> element — The maintenance tick: pull the fold's sources to head, apply the ∂ (or recompute loud), checkpoint, and CAS the alias forward.

      [?lib 'cx-stdlib/store']
[?lib 'cx-stdlib/live']
[?let [= $s [$store:open "mem://live-adv-src"]]
[= $cs [$store:open "mem://live-adv-ckpt"]]
[= $m [$live:materialize $cs [?quote [?for [in $o [$store:source $orders "//order"]] [yield $o]]] {orders: $s} "orders-live"]]
[= $h [$store:put-doc $s [order id=1]]]
  [$live:advance $m]]

      [advanced applied=1 recomputed=false [head-set [s source=orders pos=1]]]

live:read

[$live:read] -> element — A point-in-time read of a materialization: the fold value carrying its head-set coordinate.

      [?lib 'cx-stdlib/store']
[?lib 'cx-stdlib/live']
[?let [= $s [$store:open "mem://live-read-src"]]
[= $cs [$store:open "mem://live-read-ckpt"]]
[= $h1 [$store:put-doc $s [order id=1]]]
[= $m [$live:materialize $cs [?quote [?for [in $o [$store:source $orders "//order"]] [yield $o]]] {orders: $s} "orders"]]
  [$live:read $m]]

      [snapshot [head-set [s source=orders pos=1]] [order id=1]]

live:lower

[$live:lower] -> element — The three-way ingest lowering ladder: JSON text becomes a wrapped [json …] value, CX-parseable text becomes its first element verbatim, anything else becomes the lossless [foreign …] wrapper — never a drop.

      [?lib 'cx-stdlib/live']
([$live:lower "{\"sku\": \"A-1\", \"qty\": 2}"], [$live:lower "[order id=1]"], [$live:lower "plain text, not a record"])

      ([json {sku: A-1, qty: 2}], [order id=1], [foreign 'plain text, not a record'])

live:adapt-poll

[$live:adapt-poll] -> element — The :snapshot-diff floor adapter: owns a declared journal stream and, on every ingest tick, appends the source store's exact delta window as one atomic [ingested from= at= FRAME…] entry — the entry itself carries the resume token.

      [?lib 'cx-stdlib/store']
[?lib 'cx-stdlib/live']
[?let [= $db [$store:open "mem://live-poll-doc"]]
[= $a [$live:adapt-poll "mem://live-poll-doc-j" "acme" "orders" $db [?quote [?for [in $r [$store:source $db2 "//order"]] [yield $r]]] "poll-1"]]
[= $h1 [$store:put-doc $db [order id=1]]]
  [$live:ingest $a]]

      [ingested from=0 at=1 n=1]

live:adapt-watch

[$live:adapt-watch] -> element — The :coalesced-rescan floor adapter: fs watch events between ticks; overflow — or the first tick — rescans the directory; files lower through the three-way ladder into the declared stream.

      [?lib 'cx-stdlib/io']
[?lib 'cx-stdlib/live']
[?let [= $a [$live:adapt-watch "mem://live-watch-doc" "acme" "files" "/tmp/live-watch-doc-dir" "watch-1"]]
[= $w1 [$io:write-file "/tmp/live-watch-doc-dir/a.json" "{\"sku\": \"A-1\"}"]]
  [$live:ingest $a]]

      [ingested rescan=true n=1]

live:ingest

[$live:ingest] -> element — The explicit adapter tick (sched cadence is the deployment driver): pull the adapter's pending window and append it to the declared stream; answers the tick summary.

      [?lib 'cx-stdlib/store']
[?lib 'cx-stdlib/live']
[?let [= $db [$store:open "mem://live-ingest-doc"]]
[= $a [$live:adapt-poll "mem://live-ingest-doc-j" "acme" "orders" $db [?quote [?for [in $r [$store:source $db2 "//order"]] [yield $r]]] "poll-1"]]
[= $h1 [$store:put-doc $db [order id=1]]]
[= $t1 [$live:ingest $a]]
  [$live:ingest $a]]

      [ingested from=1 at=1 n=0]