Shared runtime: connections, state, and convergence

Web and mobile do not share a DOM, storage adapter, or platform lifecycle. They do share the code that answers a harder question: what is the currently justified local view of one server-owned environment? packages/client-runtime is that shared seam. It turns a selected environment into a supervised RPC lease, then builds keyed state machines whose snapshots, cursors, and reducers converge after disconnects and reordered deliveries.

This chapter calls the state “converged” only when the implementation has an authoritative snapshot or has caught a known snapshot up to its cursor. It does not mean every client has already rendered the same frame, nor that a local cache has become a second source of truth.

1. An environment target names a route, not a live connection

The target union has four variants: Primary, Bearer, Relay, and SSH. Primary holds direct HTTP and WebSocket bases and may operate without a bearer when the platform provides none. Bearer and SSH targets hold a connectionId whose profile/credential data is resolved locally. Relay names the target environment and obtains its relay bootstrap through the cloud/device path. SSH asks a platform-provided gateway to prepare a bootstrap, persists the returned SSH target profile, then authorizes the resulting endpoint with the bootstrap bearer.

Those are target-specific preparation paths, not four client data models. Each successful path returns the same PreparedConnection: expected environment id, label, HTTP base, socket URL, optional HTTP authorization, and the original target. The resolver checks that a profile or authorization result names the expected environment; a stored connection id is therefore not treated as proof that it still points at the intended server.

The EnvironmentRegistry owns the next level of multiplicity. It loads persisted targets into a catalog, creates a scoped supervisor on demand per environment, and uses a per-environment lease lock so concurrent consumers acquire the same current supervisor instead of independently opening competing connections. Closing that service scope closes the supervisor and its resources. This is why an atom or RPC helper is keyed by environmentId: the key chooses an authority boundary, not merely a UI filter.

2. One attempt has a deliberately narrow job

RpcSessionFactory.connect creates connection/disconnection deferreds, opens a WebSocket with a 15-second open timeout, and creates Effect’s socket RPC protocol with retryTransientErrors: false and Schedule.recurs(0). It caches the initial serverGetConfig request; readiness requires both the socket-connect signal and that initial configuration. Its probe either calls serverProbe when the config advertises it or reuses the config request. The session exposes its closure as a transient transport failure.

        [ORCHESTRATION_WS_METHODS.subscribeThread]: (input) =>
          observeRpcStreamEffect(
            ORCHESTRATION_WS_METHODS.subscribeThread,
            Effect.gen(function* () {
              const isThisThreadDetailEvent = (event: OrchestrationEvent) =>
                event.aggregateKind === "thread" &&
                event.aggregateId === input.threadId &&
                isThreadDetailEvent(event);
 
              const liveStream = orchestrationEngine.streamDomainEvents.pipe(
                Stream.filter(isThisThreadDetailEvent),
                Stream.map((event) => ({
                  kind: "event" as const,
                  event: projectActivityEvent(event),
                })),
              );
 
              // Attach live delivery before reading either replay or snapshot state.
              // Otherwise an event published while the snapshot is loading is lost.
              const liveBuffer = yield* Queue.unbounded<OrchestrationThreadStreamItem>();
              yield* Effect.forkScoped(
                liveStream.pipe(Stream.runForEach((item) => Queue.offer(liveBuffer, item))),
              );
              const bufferedLiveStream = Stream.fromQueue(liveBuffer);
 
              // When the client already loaded the snapshot over HTTP it passes
              // that snapshot's sequence, and we resume the live subscription by
              // replaying persisted events after it instead of re-sending the
              // (potentially multi-KB) snapshot frame over the socket.
              //
              // The live PubSub subscription must be attached *before* draining
              // the catch-up replay, otherwise events published during the replay
              // window are dropped (they are past the persisted tail the replay
              // read, but the live stream is not yet subscribed). So fork the
              // live stream into a buffer bound to this stream's scope, then emit
              // catch-up followed by the buffered/ongoing live events. Overlapping
              // events are deduped by sequence on the client.
              //
              // The replay is bounded to the projection head captured below. The
              // catch-up range is normally tiny (a fresh HTTP snapshot sequence),
              // but a stale cached cursor can sit hundreds of thousands of global
              // events behind — replaying that decodes every intervening event
              // (including every other thread's tool payloads) only to discard
              // almost all of them, which has OOM-killed servers on large
              // databases. A truncated replay would silently drop this thread's
              // events, so past the gap cap we reset the client with a fresh
              // thread snapshot instead, exactly like subscribeShell above.
              if (input.afterSequence !== undefined) {
                const afterSequence = input.afterSequence;
                const headSequence = yield* orchestrationEngine.latestSequence;
                const replayGap = headSequence - afterSequence;
                if (replayGap >= 0 && replayGap <= THREAD_RESUME_MAX_GAP) {
                  const catchUpStream = orchestrationEngine
                    .readEvents(afterSequence, replayGap)
                    .pipe(
                      Stream.filter(isThisThreadDetailEvent),
                      Stream.map((event) => ({
                        kind: "event" as const,
                        event: projectActivityEvent(event),
                      })),
                      Stream.mapError(
                        (cause) =>
                          new OrchestrationGetSnapshotError({
                            message: `Failed to replay thread ${input.threadId} events`,
                            cause,
                          }),
                      ),
                    );
                  const afterCatchUp =
                    input.requestCompletionMarker === true
                      ? Stream.concat(
                          Stream.fromEffect(
                            Queue.offer(liveBuffer, { kind: "synchronized" as const }),
                          ).pipe(Stream.drain),
                          bufferedLiveStream,
                        )
                      : bufferedLiveStream;
                  return Stream.concat(catchUpStream, afterCatchUp);
                }
                // Gap too large (or cursor ahead of authoritative state): fall
                // through to the snapshot path so the client converges from a
                // fresh thread detail instead of an unbounded replay.
              }
 
              const snapshot = yield* projectionSnapshotQuery
                .getThreadDetailSnapshot(
                  input.threadId,
                  // Windowing the fallback snapshot is opt-in per subscription:
                  // clients that don't send turnLimit (including all
                  // pre-pagination clients) get the full thread, since they
                  // have no way to load older pages.
                  input.turnLimit === undefined ? undefined : { turnLimit: input.turnLimit },
                )
                .pipe(
                  Effect.mapError(
                    (cause) =>
                      new OrchestrationGetSnapshotError({
                        message: `Failed to load thread ${input.threadId}`,
                        cause,
                      }),
                  ),
                );
 
              if (Option.isNone(snapshot)) {
                return yield* new OrchestrationGetSnapshotError({
                  message: `Thread ${input.threadId} was not found`,
                  cause: input.threadId,
                });
              }
 
              const afterSnapshot =
                input.requestCompletionMarker === true
                  ? Stream.concat(
                      Stream.fromEffect(
                        Queue.offer(liveBuffer, { kind: "synchronized" as const }),
                      ).pipe(Stream.drain),
                      bufferedLiveStream,
                    )
                  : bufferedLiveStream;
              return Stream.concat(
                Stream.make({
                  kind: "snapshot" as const,
                  snapshot: projectThreadDetailSnapshot(snapshot.value),
                }),
                afterSnapshot,
              );
            }),
            { "rpc.aggregate": "orchestration" },

The seemingly strict “no retry” setting is a division of responsibility. The supervisor owns intent (connect/disconnect), network status, session generation, last failure, and the visible phases available, offline, connecting, backoff, connected, and blocked. It reports preparation, opening, and synchronization separately. Transient failures retry on the pinned 3/4/8/16-second ladder; blocked errors wait for an external signal rather than spinning. A stable connection for at least 30 seconds resets retry state. A manual retry resets it; a foreground probe can replace a stale mobile lease and make the immediate first reconnect without the first backoff rung. These are policy choices in the supervisor—not a promise that every failed operation is idempotent.

3. Atom families supply one machine per key

The runtime’s helpers create Effect Atom families around an environment key and, where needed, an input or thread id. They route effects through the registry’s environment lease. Query families watch connected generations, use stale-while- revalidate behavior, and have an idle TTL; subscription families follow the live environment stream. Command helpers execute through an AtomRegistry, and their scheduler can be parallel, FIFO serial, single-flight, or latest-wins per key.

This is not a global “store of all server objects.” The registry is the mounted atom runtime; individual families own the lifetime of a particular observable state machine. createEnvironmentShellAtoms and createEnvironmentThreadStateAtoms, for example, return environment/thread-keyed atoms. The app surface decides how to mount and render them; the shared package decides how their data reaches a justified state.

4. Caches accelerate first paint; snapshots re-establish authority

Shell and eligible settled thread state can be read from EnvironmentCacheStore. That starts the UI in cached, rather than erasing a useful previous view. Active or starting thread sessions are deliberately not persisted on every streamed change: their large, rapid payloads remain server-authoritative until settling. Persist writes travel through a one-item sliding queue and a 500 ms debounce, so cache encoding is not put on the event-stream hot path.

But a cursor alone is insufficient across a new RPC session. For the shell, lastAuthoritativeSession distinguishes a foreground resubscription on the same session (resume from the in-memory snapshot cursor) from a replacement session (fetch HTTP snapshot first). If that refresh fails, it omits the cached cursor so the server must send a complete socket snapshot. Threads similarly seed a warm cache’s lastSequence, fetch an HTTP snapshot if they have no local detail, then subscribe with afterSequence only when they have a current detail.

The server’s paired protocol makes that defensible: it attaches live delivery before loading snapshot/replay state, captures a head for the bounded replay, then emits the buffered tail. A completion marker, when the server advertises support, means the buffered work before that marker has been delivered; it is not a claim that future domain events have stopped.

flowchart LR
  T[Primary / Bearer / Relay / SSH target] --> R[Resolver]
  R --> P[Prepared connection]
  P --> G[Environment registry + supervisor]
  G --> S[One-attempt RPC session
new generation on reconnect]
  S --> H[Shell state machine]
  S --> D[Thread state machine]
  C[(platform cache)] --> H
  C --> D
  H -->|HTTP snapshot if new session| X[authoritative snapshot]
  D -->|HTTP snapshot if needed| X
  X -->|afterSequence| L[bounded replay + live tail]
  L -->|newer only| H
  L -->|newer only| D
  D --> E[history epoch + page watermark]
  E --> D

5. The sequence rule handles snapshot overlap and duplicate delivery

The shell reducer replaces state on a snapshot and applies a shell event only when its sequence is greater than snapshotSequence. Thread state does the analogous check against lastSequence; an event at or below it is ignored. This is expected to discard replay/live overlap, an event already represented in an HTTP snapshot, and duplicate delivery after resubscription. It is not an attempt to make delivery exactly once.

When the cursor is missing, ahead of the server, or beyond the configured bounded gap, the server chooses a fresh snapshot rather than silently truncating replay. On a fresh thread snapshot, the client sets the new sequence and replaces all loaded history, then advances the history epoch. That replacement matters: a client disconnected during a revert has no remaining per-event removal to apply to an older cached page.

6. Pagination is a second stream race, not “just append older rows”

Thread detail may begin with a recent turn window and a keyset beforeCursor. “Load earlier” is serialized through a sliding one-item request queue, no-ops while one page is loading, and is enabled only after the connected server advertises pagination. The client merges only windowed collections from the older snapshot; newer thread metadata stays authoritative. Identity de-duplication protects overlapping rows even if a server/cursor edge returns them.

The awkward cases are intentional design constraints:

• Stale page: a response whose snapshot sequence is below loaded state, or whose epoch differs from the request’s epoch, is discarded under the same lock used for live stream application.
• Page ahead of live cursor: it is parked instead of merged. Its thread-scoped watermark may include out-of-window updates the subscription has not delivered; merging immediately could show them and later replay them again. The parked page merges only once live lastSequence reaches that watermark.
• Snapshot or revert during page fetch: both rewrite the meaningful history and bump the epoch. The in-flight page cannot merge afterwards, so removed turns cannot be resurrected.
• Reconnect to an older server: pagination support is reset on disconnect and re-read from the new session config. A cached window is dropped before resume if the replacement server lacks pagination, because it could not supply the missing older pages.

The lab is intentionally a small model rather than a network mock. Each control advances one visible transition: it can replace a snapshot, reject a duplicate, choose snapshot recovery for a gap, replace the session, invalidate history on a revert, or park/merge/reject an older page. Its rules mirror the boundaries above; it does not claim to emulate Effect scheduling or a real server.

7. Read a reconnect as a proof obligation

The practical question after a disconnect is not “did the socket reopen?” It is:

1. Did a target resolve and a supervisor publish a ready session?
2. Did the relevant state machine obtain an authoritative snapshot for that session or a bounded replay after a known snapshot cursor?
3. Did its reducer reject overlap, and—if it is a thread—did its page merge survive both the sequence/watermark and history-epoch tests?

Only then is it reasonable for the UI to label the state live. That discipline lets web and mobile differ at the platform edge while sharing the same answer to the expensive correctness questions: what may be cached, what may be resumed, and what must be thrown away.