conference: reliable roster re-broadcast (fixes browser 'can't see each other') + scene roadmap

examples/ariadne/documents/conference_roster.ari

@@ -144,13 +144,13 @@ windows:
     id: cam_{i}
     bind: peer.{i}.name
     repeat: (state-get "peer.count")
-    pos: [780, 40]
-    size: [400, 360]
+    pos: [612, 66]
+    size: [330, 290]
     closable: true
     children:
       - type: stream_view
         bind: peer.{i}.cam
-        size: 360
+        size: 300
 
 on_tick: |
   (begin
@@ -159,6 +159,20 @@ on_tick: |
       (begin
         (state-set "joined" "1")
         (state-set "name_error" "") ; no name conflict yet (the web client's launch name is random)
+        ; ── pre-seed each peer camera window at a distinct TILED position ──
+        ; every cam_<i> declares the SAME authored pos:, so without this the per-peer windows land
+        ; exactly on top of one another and look like a single window — the #1 reason a user reports
+        ; "I can't see the other cameras". Tile them in a 2-column grid to the RIGHT of the Users
+        ; window (which sits at x≈390–610): the roster is NEVER covered, and up to four cameras show
+        ; side-by-side without overlapping on a laptop screen. Seeded ONCE here, before any peer window
+        ; exists, so each window's first-use geometry seed lands at its own spot; the user's drag then
+        ; persists and wins. Covers 6 peers; a 7th+ falls back to the authored pos.
+        (state-set "tree.cam_0.geometry" "612,66,330,290")
+        (state-set "tree.cam_1.geometry" "952,66,330,290")
+        (state-set "tree.cam_2.geometry" "612,364,330,290")
+        (state-set "tree.cam_3.geometry" "952,364,330,290")
+        (state-set "tree.cam_4.geometry" "612,662,330,290")
+        (state-set "tree.cam_5.geometry" "952,662,330,290")
         (state-set "cam_on" "1") ; on/off toggles start ON (streams open unmuted)
         (state-set "mic_on" "1")
         (room-join "" (state-get "conf.room") (dict "name" (state-get "conf.self")))

inc/cvc/state/state_transport_ws.h

@@ -128,6 +128,13 @@ class state_transport_ws final : public state_transport, public media_endpoint {
   // server's main loop. No-op under wasm. Returns how many connections were erased this call.
   std::size_t reap_idle(std::chrono::milliseconds timeout);
 
+  // Relay reliability: push the full current shard snapshot to EVERY connection (not just a new
+  // one, the way admit does). A client-to-client live re-fan is single-shot with no retry, so a
+  // join that races an already-connected peer's socket can be missed and never recovers; calling
+  // this periodically on the server's main loop lets every client re-converge on the room/scene
+  // tree within one interval (idempotent bulk apply). Returns how many connections it wrote to.
+  std::size_t broadcast_snapshot();
+
   // ── Media plane (bring-up: media over this WS control channel) ──
   // publish_media_frame sends a frame to this transport's connections: a browser CLIENT has one
   // connection (the relay), so it pushes its camera/mic UP; a relay's media_router fans each

src/arigl/main.cpp

@@ -466,11 +466,24 @@ struct ConferenceSession {
     const char *synthEnv = std::getenv("CVC_ARIADNE_SYNTHETIC_CAM");
     const std::string synth = (synthEnv && *synthEnv && std::string(synthEnv) != "0") ? "1" : "0";
     std::vector<std::string> ce;
-    ari::run_init(app, prefix,
-                  "(state-set \"conf.room\" \"" + room + "\") (state-set \"conf.self\" \"" + self +
-                      "\") (state-set \"cam_synthetic\" \"" + synth +
-                      "\") (state-set \"self.pid\" \"" + nodeId + "\")",
-                  &ce);
+    ari::run_init(
+        app, prefix,
+        "(state-set \"conf.room\" \"" + room + "\") (state-set \"conf.self\" \"" + self +
+            "\") (state-set \"cam_synthetic\" \"" + synth + "\") (state-set \"self.pid\" \"" +
+            nodeId +
+            // Seed peer.count to a valid integer BEFORE the first render, so the roster's
+            // `repeat: (state-get "peer.count")` never evaluates an empty node on frame 0
+            // (which warns "did not yield an integer" until the first on_tick sets it).
+            "\") (state-set \"peer.count\" \"0\")"
+            // Seed the device-selection fields to EMPTY STRINGS. state-get on an UNSET node
+            // returns nil, not "", so the real-camera on_tick guard (= cam.want_device "")
+            // was false before the host first populated it, running the device-switch body
+            // and evaluating (int nil) — which THROWS and permanently kills the on_tick
+            // resident, so the roster never builds (synthetic skips that block, so it was
+            // only ever broken for real cameras). With "" seeded the guard is correct.
+            " (state-set \"cam.want_device\" \"\") (state-set \"mic.want_device\" \"\")"
+            " (state-set \"cam.cur_device\" \"\") (state-set \"mic.cur_device\" \"\")",
+        &ce);
     for (const std::string &e : ce)
       std::fprintf(stderr, "arigl: conference init: %s\n", e.c_str());
     // NOTE: device enumeration is deferred to pump() (see devices_enumerated_), NOT done here —

src/arigl/web/cyprty.html

@@ -138,7 +138,7 @@ <h1>cyberpcangel &middot; cyprty</h1>
     },
   };
 
-  const V = 'cyprty-ui4'; // bump on every redeploy so caches can't serve a stale wasm/js
+  const V = 'cyprty-ui7'; // bump on every redeploy so caches can't serve a stale wasm/js
   async function boot() {
     try {
       const resp = await fetch(DEMO + '.wasm?v=' + V, {cache: 'no-cache'});

src/arisrv/main.cpp

@@ -156,8 +156,15 @@ int main(int argc, char **argv) {
     // close stops streaming media; reap_idle shuts such a connection down after ~10 s so its
     // on-close retraction fires and the roster does not accumulate ghosts. A live client streams
     // continuously.
-    if (++reapticks % 5 == 0)
+    if (++reapticks % 5 == 0) {
       ws_tp.reap_idle(std::chrono::seconds(10));
+      // Re-broadcast the room/scene snapshot to every client (~1s). A client-to-client live re-fan
+      // is single-shot with no retry, so a join that races an already-connected peer's socket is
+      // otherwise lost for good (an established client never sees a later joiner). This idempotent
+      // re-sync makes every client re-converge within ~1s — and is the reliable push the
+      // server-authoritative room+scene tree rides on.
+      ws_tp.broadcast_snapshot();
+    }
     // Apply any queued departures (a browser tab closed) before pumping, so each retraction goes
     // out to the remaining clients this tick.
     {

src/cvc/ariadne/ariadne.cpp

@@ -687,7 +687,18 @@ class ReactiveEngine {
 
   static constexpr uint64_t kMaxSteps = 100'000;       // per-eval step cap (a predicate is tiny)
   static constexpr double kMaxSeconds = 0.005;         // per-eval time cap (5 ms)
-  static constexpr double kFrameBudgetSeconds = 0.010; // aggregate across all predicates/frame
+  // Aggregate reactive-eval time across ALL predicates in a frame. This is a runaway guard, not a
+  // normal operating point: a handful of scalar predicates finish in well under a millisecond. It
+  // was 10 ms, but a live conference re-evaluates ~5-7 predicates PER PEER every frame (roster
+  // rows' visible_when, the per-peer camera-window count/title/texture binds), and the stackless
+  // evaluator is markedly slower under wasm while the browser's single thread is also decoding
+  // video — so at a realistic peer count the 10 ms guard tripped mid-roster, skipping the rest and
+  // blinking the participant list / peer windows empty. 30 ms clears a ~6-7 peer conference with
+  // margin and is still a hard bound on a pathological document (eval time the thread spends
+  // descheduled doesn't count against it, so a merely-loaded machine never trips it). See
+  // last_good_count for the structural backstop that keeps a repeated list stable across any
+  // residual skip.
+  static constexpr double kFrameBudgetSeconds = 0.030; // aggregate across all predicates/frame
   static constexpr std::size_t kMaxOptions = 1024;     // cap on computed combo options
 
   std::size_t data_step_nodes_ = 0;            // see resolve_data_settings()
@@ -780,6 +791,14 @@ struct Runtime::Impl {
   std::vector<std::string> reactive_warnings;
   std::set<std::string> reactive_warned;
   int frame_instances = 0; // §3: repeat instances emitted this frame (bounds many repeats)
+  // §3: last successful repeat-count per expression. A reactive repeat-count eval fail-safes to 0
+  // on a transient per-frame budget exhaustion (reactive eval is slow in wasm, and video decode can
+  // eat the frame); collapsing a repeated list — the conference roster, the per-peer camera windows
+  // — to EMPTY for that frame and back flickers it, and under sustained load leaves an apparently
+  // empty conference even though the peers are present in state. Reusing the last-good count
+  // bridges those skips so the list stays stable; a genuine count change is picked up on the next
+  // unskipped frame.
+  std::unordered_map<std::string, int> last_good_count;
 
   // §4.7 action lane: a unique owner tag for this Runtime's processes on the app-wide
   // scheduler (app.exec_scheduler()), so teardown reaps exactly this document's process
@@ -993,8 +1012,16 @@ std::vector<std::string> Runtime::Impl::eval_options(const std::string &expr) {
 int Runtime::Impl::eval_count(const std::string &expr) {
   constexpr int64_t kMaxRepeat = 4096; // a UI with thousands of repeated widgets is pathological
   const ReactiveEngine::IntOutcome o = ensure_reactive().eval_int(expr, /*dflt=*/0);
-  if (!o.error.empty())
+  if (!o.error.empty()) {
     warn_once(o.error);
+    // The eval fail-safed (budget exhausted / broken). Rather than collapse the repeated list to 0
+    // this frame, reuse the last count this expression yielded successfully (if any) — a transient
+    // skip then leaves the roster/peer-window list unchanged instead of blinking empty. With no
+    // prior success the fallback is still 0, so a genuinely-never-valid expr behaves as before.
+    const auto it = last_good_count.find(expr);
+    if (it != last_good_count.end())
+      return it->second;
+  }
   int64_t n = o.value;
   if (n < 0)
     n = 0;
@@ -1003,6 +1030,10 @@ int Runtime::Impl::eval_count(const std::string &expr) {
               std::to_string(kMaxRepeat) + ") — clamped");
     n = kMaxRepeat;
   }
+  // Remember this frame's good value only when the eval actually succeeded (o.error empty): a
+  // fail-safe 0 must never be cached as "last good", or it would pin the list empty.
+  if (o.error.empty())
+    last_good_count[expr] = static_cast<int>(n);
   return static_cast<int>(n);
 }
 

src/cvc/state/state_transport_ws.cpp

@@ -33,8 +33,13 @@
 #include <boost/asio/ip/tcp.hpp>
 #include <boost/beast/core.hpp>
 #include <boost/beast/websocket.hpp>
+#ifdef __linux__
+#include <linux/sockios.h> // SIOCOUTQ — unsent bytes in a socket's send buffer (send_backlog)
+#include <sys/ioctl.h>
+#endif
 #else
 #include <cstdint>
+#include <cstdlib> // std::getenv — CVC_WS_BROADCAST_DEBUG diagnostic gate
 #include <cstring>
 #include <emscripten/websocket.h>
 #endif
@@ -280,6 +285,25 @@ class state_transport_ws::connection {
     std::lock_guard<std::mutex> lk(write_mu);
     return write_locked(f);
   }
+
+  // Bytes still unsent in this connection's OS send buffer (Linux only; 0 elsewhere). A large value
+  // means the peer is not draining fast enough and the next write would BLOCK. The fan paths check
+  // this before writing and skip a backed-up connection — dropping a loss-tolerant media frame, or
+  // deferring the periodic roster re-broadcast to the next tick — instead of blocking the whole
+  // relay (and every other client's roster) on one slow peer. The continuous video flood otherwise
+  // fills a slow client's buffer, the synchronous ws->write() blocks, and the broadcast loop stalls
+  // for all.
+  std::size_t send_backlog() {
+#if !defined(__EMSCRIPTEN__) && defined(__linux__)
+    if (!ws)
+      return 0;
+    int unsent = 0;
+    const int fd = ws->next_layer().native_handle();
+    if (fd >= 0 && ::ioctl(fd, SIOCOUTQ, &unsent) == 0 && unsent > 0)
+      return static_cast<std::size_t>(unsent);
+#endif
+    return 0;
+  }
 };
 
 #ifndef __EMSCRIPTEN__
@@ -789,6 +813,11 @@ state_transport::publish_stats state_transport_ws::publish(const state_mutation
       _peers.note_delivery_filtered(c->remote_node_id);
       continue;
     }
+    // Don't block the caller (the arisrv main loop drains local writes here) on a backed-up peer —
+    // it would stall the loop and its periodic broadcast for everyone. The ~1s broadcast
+    // re-delivers.
+    if (c->send_backlog() > 2 * 1024 * 1024)
+      continue;
     if (c->write(frame)) {
       _sent_frames.fetch_add(1, std::memory_order_relaxed);
       ++stats.delivered;
@@ -881,6 +910,14 @@ std::size_t state_transport_ws::write_media(const media_frame &m, const void *ex
   for (auto &c : conns) {
     if (!c || c.get() == exclude || !c->alive.load() || !c->can_write())
       continue;
+    // Media is loss-tolerant: if this peer's send buffer is backed up (it can't keep pace with the
+    // video), DROP the frame rather than let the synchronous write block the relay. A LOW threshold
+    // is deliberate — it keeps each peer's buffer small so the small, must-deliver STATE writes
+    // (roster join/leave + the periodic re-broadcast below) are never crowded out or deferred.
+    // Dropping video to a slow peer is invisible; a stalled roster makes clients unable to SEE each
+    // other at all.
+    if (c->send_backlog() > 32 * 1024)
+      continue;
     if (c->write(frame)) {
       _sent_frames.fetch_add(1, std::memory_order_relaxed);
       ++n;
@@ -980,6 +1017,9 @@ void state_transport_ws::on_inbound_mutation(const state_mutation &m, connection
     for (auto &c : conns) {
       if (!c || c.get() == origin || !c->alive.load() || !c->can_write())
         continue;
+      if (c->send_backlog() > 2 * 1024 * 1024)
+        continue; // don't block the relay on a backed-up peer; the ~1s roster broadcast backstops
+                  // it
       if (c->write(frame))
         _sent_frames.fetch_add(1, std::memory_order_relaxed);
     }
@@ -1017,6 +1057,9 @@ void state_transport_ws::on_inbound_message(const state_message &m, connection *
     for (auto &c : conns) {
       if (!c || c.get() == origin || !c->alive.load() || !c->can_write())
         continue;
+      if (c->send_backlog() > 2 * 1024 * 1024)
+        continue; // don't block the relay on a backed-up peer; the ~1s roster broadcast backstops
+                  // it
       if (c->write(frame))
         _sent_frames.fetch_add(1, std::memory_order_relaxed);
     }
@@ -1121,6 +1164,93 @@ void state_transport_ws::send_snapshot_locked(connection &c,
   }
 }
 
+std::size_t state_transport_ws::broadcast_snapshot() {
+  // Build each shard's snapshot frame ONCE, then write it to every live connection. The receivers
+  // apply it idempotently (ingest_snapshot_entry is a bulk load), so re-sending the same
+  // roster/scene each interval is a no-op for a client that is already converged and a recovery for
+  // one that missed a live re-fan. Uses connection::write() (locks per connection) — do NOT hold a
+  // connection write lock here. Snapshotting the shard reads the tree under its own lock.
+  std::vector<state_cluster_shard *> shards;
+  {
+    std::lock_guard<std::mutex> lk(_shards_mu);
+    shards = _shards;
+  }
+  std::vector<pb::Frame> frames;
+  for (auto *s : shards) {
+    if (s == nullptr)
+      continue;
+    const auto snap = s->snapshot();
+    if (snap.empty())
+      continue;
+    pb::Frame f;
+    auto *rsp = f.mutable_snapshot_response();
+    rsp->set_request_id(0);
+    rsp->set_final(true);
+    for (const auto &e : snap) {
+      auto *pe = rsp->add_entries();
+      pe->set_path(e.path);
+      pe->set_string_value(e.string_value);
+      pe->set_comment(e.comment);
+      pe->set_hidden(e.hidden);
+      pe->set_read_only(e.read_only);
+      for (const auto &p : e.properties)
+        (*pe->mutable_properties())[p.first] = p.second;
+      pe->set_type_name(e.type_name);
+      pe->set_origin_node_id(e.origin_node_id);
+      pe->set_sequence(e.sequence);
+      pe->set_data_type_name(e.data_type_name);
+      if (e.data_payload.kind != state_payload_kind::none)
+        set_pb_payload(pe->mutable_data_payload(), e.data_payload);
+    }
+    frames.push_back(std::move(f));
+  }
+  if (frames.empty())
+    return 0;
+  std::vector<std::shared_ptr<connection>> conns;
+  {
+    std::lock_guard<std::mutex> lk(_conns_mu);
+    conns = _conns;
+  }
+  // Diagnostic (off unless CVC_WS_BROADCAST_DEBUG is set): per-tick summary of how many peers the
+  // roster broadcast reached vs. skipped, and the worst backlog observed. This is the decisive
+  // signal for "clients can't see each other" — a skipped peer is one the periodic roster never
+  // reaches.
+  static const bool dbg = std::getenv("CVC_WS_BROADCAST_DEBUG") != nullptr;
+  std::size_t wrote = 0, skipped = 0, max_backlog = 0;
+  for (auto &c : conns) {
+    if (!c || !c->alive.load() || !c->can_write())
+      continue;
+    const std::size_t bl = c->send_backlog();
+    if (bl > max_backlog)
+      max_backlog = bl;
+    // Skip a peer whose send buffer is badly backed up: writing the (multi-entry) snapshot would
+    // block the whole broadcast loop on that one slow peer, starving every other client. This is
+    // periodic
+    // (~1s), so a skipped peer simply catches up on the next tick once its buffer drains. The media
+    // drop keeps a peer's buffer from filling with video, so at 2 MB this only trips for a peer
+    // whose socket is effectively dead (not draining state at all) — exactly when we must not
+    // block.
+    if (bl > 2 * 1024 * 1024) {
+      ++skipped;
+      if (dbg)
+        std::fprintf(stderr, "[ws-bcast] SKIP node='%s' backlog=%zuKB\n", c->remote_node_id.c_str(),
+                     bl / 1024);
+      continue;
+    }
+    bool ok = true;
+    for (const auto &f : frames)
+      ok = c->write(f) && ok;
+    if (ok) {
+      ++wrote;
+      _sent_frames.fetch_add(frames.size(), std::memory_order_relaxed);
+    }
+  }
+  if (dbg)
+    std::fprintf(stderr, "[ws-bcast] wrote=%zu skipped=%zu maxbacklog=%zuKB conns=%zu frames=%zu\n",
+                 wrote, skipped, max_backlog / 1024, conns.size(), frames.size());
+  return wrote;
+}
+
 void state_transport_ws::unregister_connection(connection *conn) {
   std::lock_guard<std::mutex> lk(_conns_mu);
   _conns.erase(