feat(mobile): page-side RpcClient over the web shell bridge (OTA phase C, C0.4) (#21467)

* feat(mobile): RN bridge host for the web shell page (OTA phase C, C0.3)

One page document's end of the bridge: page frames in through the C0.1
reader, one RpcClient behind it, host frames out. Requests forward with the
arity the page used and answer with the verbatim RpcResponse, chunked when it
is over the frame cap; a rejection crosses as the five-field capture instead.
Subscriptions carry a seq and an unacked window, and end with `overflow`
rather than dropping frames a reader cannot see are missing.

The fence is structural: the protocol names no host, so the client is
whichever this host was built with, and the in-flight caps the page is told
about in `init` are enforced here rather than trusted from there.

Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb

* feat(mobile): wire the bridge host to B4's hybrid shell screen (OTA phase C, C0.3)

The channel opens on the session B4 put on screen and closes with it. The
session id is B4's: nothing new is minted, and a remount is a new one, which
is what makes a dead page's frames fail the native origin check.

Both halves are stamped with the session they belong to, because React swaps
refs during the commit and runs the retiring effect's cleanup after it — a
host disposing on a remount would otherwise post its teardown into the page
that replaced it. `bridgeEnabled` is derived from the session step alone,
since the native side treats a prop change as a reload.

Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb

* test(mobile): prove the bridge fence holds for traffic, not just for answers

A mutation that dropped the post-teardown guard in `receive` survived: the
teardown case only fed a frame whose answer the outbound guard already
swallowed, so nothing observed that a dead page could still reach a live
client. Both teardown paths now feed a request, a subscribe and a notify,
and assert the client saw none of them.

Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb

* test(mobile): read the hook's frames through the page's own reader

`JSON.parse` returns `any`, and taming it with an assertion is a cast the
gate refuses and a check nobody gets. Reading each posted frame through
`readBridgeHostMessage` types it and proves the same thing the host's own
suite does: a frame the page would refuse is a frame that never arrives.

Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb

* chore(mobile): list the bridge host as a raw request port owner

The boundary ratchet reads a `.sendRequest` access as a call site, and the
host has three: one per arity the page can use. It is not a call site. It
picks no method, reads no reply and decides no acceptance — the page names
the method and runs the typed operation over the client this carries, which
is what the C0 design put page-side so `runRpcOperation` stays unchanged
there. That makes it an owner, beside the socket and relay senders, not a
migration backlog entry.

Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb

* feat(mobile): page-side RpcClient over the web shell bridge (OTA phase C, C0.4)

Every member of the native contract, carried over the C0.1 envelope so the
screens above it cannot tell a bridge from a socket: requests keep the arity the
caller used, a host RpcFailure resolves as data while a rejection is rebuilt with
its class and its delivery-unknown mark, subscriptions stream with periodic acks,
and the synchronous getters read a cache primed by init rather than answering
before they know.

A state whose generation went backwards is refused and re-asked for, because a
shell rebuilt under the page makes what the page holds the newer of the two.
close settles what the page owns and never touches the shell's client, which the
native screens and the host catalog still share.

Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb

* test(mobile): run the page client against the shell host over an in-memory port pair

One FIFO per direction and delivery on a microtask, which is what C0.5's golden
replay needs: a subscribe that overtook a sendRequest would move the recorder's
shared ordinal, and anything stronger than a microtask moves a virtual
millisecond. Every member round-trips through the real host over a fake client;
the frame-level suite covers what no pair can reach, including the handshake
backoff, refusals and the binary lane C6 will fill.

Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb

* chore(mobile): list the page bridge client as a raw request port owner

Both ends of the bridge hold the port as a transport: one forwards raw requests
and the other offers them, and neither picks a method or reads a reply.

Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb

* refactor(mobile): drop the assembler discard no abandoned request can reach

A request is only abandoned when its frame never left the page, so the shell was
never told the id and no part can have arrived under it. Says what actually keeps
an omitted param omitted while it is here: JSON drops an undefined value, so the
spread states the intent rather than producing the result.

Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb

* test(mobile): close the three gaps a mutation sweep found in the page client

A settled id has to give its assembler slot back, or 64 replies that were cut
short before an error leave the page unable to read the next chunked one. Close
says goodbye once rather than cancelling each stream first. And the read guard is
only observable through a port that ignores its own unsubscribe, which is what
the harness can now be.

Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb

* test(mobile): prove a stream that overflows inside subscribe is unsubscribed

A client that emits synchronously from `subscribe` can retire a stream before
its unsubscribe exists to be stored. The identity check that calls it instead
had no test; deleting it left the suite green while the client's stream leaked.

Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb

* fix(mobile): hand the bridge host over in the commit, not after it

A client swap that keeps the session id leaves the handler's own fence inert:
until the passive effect ran, a native frame reached the retiring host and the
client it closed over. A layout effect swaps both inside the commit.

Teardown on unmount now runs while the view is still attached, so a pending
request is answered delivery-unknown instead of being dropped on the floor.

Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb

* fix(mobile): hold a refused page frame to one warning per page

A page that sends one bad frame usually sends many, and a line each buries the
first — the one that says why. Same bound the host already keeps on a failing
post, applied per kind and reset when a new page gets a new host.

Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb

* fix(mobile): bound the page's terminal viewport at the bridge contract

A viewport crossing the bridge is written into the cached subscribe params of
every stream naming that terminal, including the native terminal screen's, and
the desktop refuses cols over 1000 or rows over 500 when those streams
resubscribe. Unbounded, one page could kill streams it never opened; the frame
is refused instead, and the bound is pinned to the desktop's own.

Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb

* fix(mobile): stop the page's close from latching the bridge host shut

One view carries every document the shell loads, so the page that says `close`
is not the last one. A latched host dropped the next document's `ready` in
silence, and a page that re-sends `ready` on a backoff would retry forever with
nothing posted and nothing logged. Close now cancels what the page owned and
leaves the host live; only dispose shuts it, and a frame arriving after that is
diagnosed rather than dropped.

Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb

* fix(mobile): keep a throwing client or post inside the bridge host

The `state` frame is sent from inside the client's own state-change fan-out and
a notify runs on the native event handler that delivered the page's frame, so a
synchronous throw from either escapes into a loop the bridge does not own and
takes unrelated listeners with it. Both are fenced and reported once.

Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb

* test(mobile): prove an ack releases the stream's unacked bytes

The frame window reopens on ack through the splice, so deleting the byte
release left every existing test green while a long-lived stream of large
frames would end with overflow on its first frame after an ack.

Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb

* fix(mobile): pass the commit-window harness its children as a prop

`createElement`'s variadic children do not satisfy a props type that declares
`children`, so the file dropped out of the tests typecheck ratchet.

Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb

* fix(mobile): render the harness from the commit-window wrapper, not as children

A props type that declares `children` is what `createElement`'s variadic form
does not satisfy, and passing it as a prop instead trips the react rule. The
wrapper renders the harness itself, which is the parent position the layout
effect ordering needs anyway.

Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb

* fix(mobile): pin the desktop viewport bound by reading it, not importing it

Mobile may not pull an rpc-contract *value* into its bundle, and the boundary
test that enforces that scans this test file too. The pin reads the schema's
own source instead, so drift in either bound still fails loudly.

Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb

* fix(mobile): settle a refused subscribe as the stream it was

The shell answers a refused `subscribe` with `error` on the stream's id.
Routing that to the pending requests dropped it, because no request is
open under that id: the page heard nothing and kept the slot forever.

Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb

* fix(mobile): report a reply or an error the page has no id for

Silently dropped before. Nothing recovers it in place, but a frame the
page cannot place means the two ledgers disagree, which is worth a line.

Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb

* fix(mobile): say disconnected on close instead of going silent

Every native client publishes the transition and keeps answering its last
snapshot; the screens read both. The page's client cleared the cache
instead, so a closing page left its listeners on a dot that never moved
and every getter throwing underneath it.

Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb

* fix(mobile): let a closed page client go inert, not throw

An unmounting screen still calls, and nothing on a teardown path catches.
Subscribe hands back a no-op dispose and the notifies do nothing, as the
native client's do, and a request rejects rather than throwing past the
caller's catch. A call before init still throws: that one is a bug.

Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb

* refactor(mobile): lift the init handshake out of the page client

The backoff that asks the shell for a session is its own concern, and the
client had grown past the file's line budget holding it.

Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb

* test(mobile): pin the cancel a page owes for a stream already ended

A screen unmounts on its own schedule, routinely after the shell gave up
on the stream. Only the double-dispose order was covered.

Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb

* docs(mobile): state what the page client does after close

The doc gave the pre-init rule and stopped; the after-close rule is the
opposite one, and subscription failures have no channel but a diagnostic.

Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb

* feat(mobile): read the shell's page channel as a client transport

The document-start installer leaves `postMessage` and one `onmessage`
slot, the intersection of what the two platforms inject. A page opened
outside the shell has no global at all, so reading it answers null rather
than throwing: the bundle still has to open in a browser.

Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb

* feat(mobile): give the page its bridge client instead of a placeholder

The web provider now builds BridgeRpcClient over the shell channel and
mounts nothing until `init` lands: every member throws before a session,
and a screen that rendered first would record its first frame against a
client that has none. Outside the shell there is no session coming, so
the placeholder stays and the route tree mounts at once, which is what
the Route A render check exercises.

Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb

* fix(mobile): declare the page provider test's probe instead of casting it

Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb

* fix(mobile): serve the bridge to one document at a time

A page's `close` now ends that document's turn: until the next `ready`
claims the view, every other frame is dropped and diagnosed instead of
reaching the client, and nothing is posted. Without the fence a straggler
from the closed document was still forwarded, and a `state` frame from the
still-running client landed in the replacement document before its `init`.

Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb

* fix(mobile): hold the request cap against the calls, not the page's ledger

`sendRequest` has no cancel, so a request the page cancelled or closed out
keeps running on the desktop until it answers. The cap now counts those
calls until each settles; counting the pending map let a page interleaving
`close` with batches hold many more than the cap `init` advertises.

Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb

* test(mobile): pin the ack ratio to the shell's window, not a copy of it

The ack interval test held 256 and 4 MiB as literals, so narrowing the
shell's window would have left the page acking too late with the test still
green. The comment naming the test that pins the ratio pointed at the wrong
file.

Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb

* fix(mobile): give back the slot of a subscribe that never left the page

A post that threw left the stream in the page's ledger with nothing open on
the shell's side, so 32 of them exhausted the subscription budget for the
life of the document. The slot goes back and the listener hears a terminal
error result, which is what the native client does with the same failure.

Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb

* fix(mobile): end a page stream through its listener, not only the log

A stream the shell ends or fails now reaches its listener as a terminal
error result, the way the native client's emitError does. A consumer reads
that result: host-worktree-refresh clears the flag that says the event
stream is live, and without it the worktree list stops updating for the life
of the document. A dispose the page asked for stays silent.

Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb

* fix(mobile): settle the old shell's work before adopting a new session

A second `init` naming a different sessionId is a rebuilt host with empty
tables: every pending request and every open stream the page still held
belonged to the shell that is gone. They now settle delivery-unknown and end
through their listeners before the new session is adopted. A second `init`
for the same session is what a re-asked `ready` earns, and keeps everything.

Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb

* fix(mobile): take the page's streams out of the ledger before failing them

A listener that resubscribes while the old shell's streams are being ended
is opening one against the shell that is arriving; draining the map first is
what keeps this loop from tearing that one down too. Fixes the lint the
previous commit left behind.

Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb

* docs(mobile): say how long a reply assembler's refusal actually lives

The tombstone is not kept forever: the request ledger discards the id as it
settles the caller, so it normally outlives only the rest of the reply that
raised it. The bounded map is there for the ids nothing settles.

Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb

* test(mobile): pin the ceiling the ready backoff stops widening at

An unclamped backoff reads the same for the first minute and then leaves a
page asking once an hour into a shell that is still booting behind it.

Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb

* docs(mobile): say why the document fence carries no epoch

Page frames reach the shell through one native listener per platform, so a
straggler from the closed document lands before the next document's `ready`
and the flag alone catches it. An echoed epoch would be a wire change for
nothing.

Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb

mobile/src/mobile-web-shell/bridge-host.ts

@@ -95,6 +95,8 @@ export function createBridgeHost(options: BridgeHostOptions): BridgeHost {
   let inFlight = 0
   // One document's turn at the bridge. `close` ends it and the next `ready` begins the next one;
   // between the two the view belongs to no document, so nothing is served and nothing is posted.
+  // No epoch rides along: one native listener delivers page frames in order, so a straggler from
+  // the closed document is always behind it and ahead of the next document's `ready`.
   let serving = true
   let postFailureReported = false
   let notifyFailureReported = false

mobile/src/mobile-web-shell/bridge/bridge-client-connection-cache.ts

@@ -0,0 +1,85 @@
+import type { ConnectionState } from '../../transport/types'
+import type { BridgeConnectionSnapshot } from './bridge-envelope'
+
+/** Why a `state` frame did not land. `unprimed` is a frame that beat `init`, `stale` one that lost to it. */
+export type BridgeSnapshotOutcome = 'applied' | 'stale' | 'unprimed'
+
+/**
+ * What the page's synchronous `RpcClient` getters read.
+ *
+ * Screens read `getState()` during render, so the answer has to already be here when the first one
+ * mounts: `init` primes it, every `state` refreshes it, and nothing is ever derived or guessed. A
+ * cache that answered `connecting` because it had not heard yet would move a golden.
+ */
+export class BridgeConnectionCache {
+  private held: BridgeConnectionSnapshot | null = null
+  private readonly listeners = new Set<(state: ConnectionState) => void>()
+
+  read(): BridgeConnectionSnapshot | null {
+    return this.held
+  }
+
+  /** From `init`. Re-priming with the same state is not a transition, so no listener hears one. */
+  prime(snapshot: BridgeConnectionSnapshot): void {
+    const changed = this.held?.state !== snapshot.state
+    this.held = snapshot
+    if (changed) {
+      this.fanOut(snapshot.state)
+    }
+  }
+
+  /**
+   * From `state`, one frame per transition on the shell's side, so every accepted one is fanned out.
+   *
+   * A snapshot whose generation went backwards is refused: the shell was rebuilt over a newer
+   * client and the page missed the `init` that would have said so, which makes what the page holds
+   * newer than what just arrived. Applying it would walk the cache backwards and leave every getter
+   * answering for a client that no longer exists.
+   */
+  apply(snapshot: BridgeConnectionSnapshot): BridgeSnapshotOutcome {
+    const previous = this.held
+    if (previous === null) {
+      return 'unprimed'
+    }
+    if (
+      previous.generation !== null &&
+      snapshot.generation !== null &&
+      snapshot.generation < previous.generation
+    ) {
+      return 'stale'
+    }
+    this.held = snapshot
+    this.fanOut(snapshot.state)
+    return 'applied'
+  }
+
+  onStateChange(listener: (state: ConnectionState) => void): () => void {
+    this.listeners.add(listener)
+    return () => {
+      this.listeners.delete(listener)
+    }
+  }
+
+  /**
+   * The page said goodbye. Every native client publishes `disconnected` when it closes and keeps
+   * answering its last snapshot afterwards, and the screens above this one are written to that: a
+   * getter that threw here, or a listener that never heard the transition, would leave a closing
+   * page rendering a dot that is still connected.
+   */
+  close(): void {
+    const held = this.held
+    if (held !== null && held.state !== 'disconnected') {
+      this.held = { ...held, state: 'disconnected' }
+      this.fanOut('disconnected')
+    }
+    this.listeners.clear()
+  }
+
+  // Walked in place: a listener that unsubscribes a sibling during the fan-out is what a `Set`
+  // iterator is specified to survive.
+  private fanOut(state: ConnectionState): void {
+    for (const listener of this.listeners) {
+      listener(state)
+    }
+  }
+}

mobile/src/mobile-web-shell/bridge/bridge-client-errors.ts

@@ -0,0 +1,50 @@
+import type { BridgeRefusal } from './bridge-caps'
+
+/** Everything the page's own client raises, as opposed to what it reconstructs from the shell. */
+
+/** A call that needs a session the page is not in yet. Always a mount-order bug, never a retry. */
+export class BridgeClientNotReadyError extends Error {
+  constructor() {
+    super('the page bridge has no session yet; wait for init before calling the client')
+    this.name = 'BridgeClientNotReadyError'
+  }
+}
+
+export class BridgeClientClosedError extends Error {
+  constructor() {
+    super('the page bridge was closed')
+    this.name = 'BridgeClientClosedError'
+  }
+}
+
+/** A second `init` naming a different session: whatever the page still held belonged to the shell
+ *  that is now gone, and the one that replaced it has never heard of any of it. */
+export class BridgeShellReplacedError extends Error {
+  constructor() {
+    super('the shell behind this page was replaced')
+    this.name = 'BridgeShellReplacedError'
+  }
+}
+
+/** The page's copy of the shell's in-flight caps, refusing before the round trip rather than after. */
+export class BridgeClientCapExceededError extends Error {
+  constructor(message: string) {
+    super(message)
+    this.name = 'BridgeClientCapExceededError'
+  }
+}
+
+export class BridgeReplyRefusedError extends Error {
+  constructor(refusal: BridgeRefusal) {
+    super(`the reply could not be read (${refusal})`)
+    this.name = 'BridgeReplyRefusedError'
+  }
+}
+
+/** The frame never left the page, so this is a definite send failure and carries no delivery mark. */
+export class BridgeSendFailedError extends Error {
+  constructor() {
+    super('the request could not be posted to the shell')
+    this.name = 'BridgeSendFailedError'
+  }
+}

mobile/src/mobile-web-shell/bridge/bridge-client-init-handshake.ts

@@ -0,0 +1,48 @@
+/** The page asks again until the shell answers; a session has no other way to start. */
+export const BRIDGE_READY_RETRY_MIN_MS = 50
+export const BRIDGE_READY_RETRY_MAX_MS = 2000
+
+export type BridgeInitHandshake = {
+  /** Posts `ready` now, and again on a widening backoff until `stop`. */
+  start: () => void
+  stop: () => void
+  /** For a shell rebuilt under the page: the wait starts over from the floor. */
+  restart: () => void
+}
+
+/**
+ * How the page gets a session.
+ *
+ * The shell posts `init` when it is ready, but a page that loaded first, or reloaded after the shell
+ * had already sent one, would wait forever for a frame that has been and gone. Asking on a widening
+ * backoff costs one frame at a time and needs nothing remembered on the shell's side.
+ */
+export function createBridgeInitHandshake(ask: () => void): BridgeInitHandshake {
+  let timer: ReturnType<typeof setTimeout> | null = null
+  let delayMs = BRIDGE_READY_RETRY_MIN_MS
+
+  function start(): void {
+    ask()
+    timer = setTimeout(() => {
+      delayMs = Math.min(delayMs * 2, BRIDGE_READY_RETRY_MAX_MS)
+      start()
+    }, delayMs)
+  }
+
+  function stop(): void {
+    if (timer !== null) {
+      clearTimeout(timer)
+      timer = null
+    }
+  }
+
+  return {
+    start,
+    stop,
+    restart: (): void => {
+      stop()
+      delayMs = BRIDGE_READY_RETRY_MIN_MS
+      start()
+    }
+  }
+}

mobile/src/mobile-web-shell/bridge/bridge-client-requests.ts

@@ -0,0 +1,89 @@
+import { markRpcDeliveryUnknown } from '../../transport/rpc-delivery-ambiguity'
+import type { RpcResponse } from '../../transport/types'
+import { BridgeClientClosedError, BridgeReplyRefusedError } from './bridge-client-errors'
+import type { BridgeReplyMessage } from './bridge-envelope'
+import { BridgeReplyAssembler } from './bridge-reply-chunking'
+
+export type PendingRequest = {
+  resolve: (response: RpcResponse) => void
+  reject: (error: unknown) => void
+}
+
+/**
+ * The page's in-flight requests, and the replies that settle them.
+ *
+ * Nothing here expires an id on its own, so every id this opens is discarded from the assembler the
+ * moment it settles or is abandoned: a reply whose last part never arrives would otherwise hold a
+ * slot until the page closes, and 64 of those are the whole in-flight budget.
+ */
+export class BridgeClientRequests {
+  private readonly pending = new Map<string, PendingRequest>()
+  private readonly assembler = new BridgeReplyAssembler()
+
+  get size(): number {
+    return this.pending.size
+  }
+
+  has(id: string): boolean {
+    return this.pending.has(id)
+  }
+
+  open(id: string, request: PendingRequest): void {
+    this.pending.set(id, request)
+  }
+
+  /** For a frame that never left the page: the caller settles it, and no part can have arrived for
+   *  an id the shell was never told about, so there is no assembler slot to give back. */
+  abandon(id: string): void {
+    this.pending.delete(id)
+  }
+
+  acceptReply(message: BridgeReplyMessage): void {
+    const assembly = this.assembler.accept(message)
+    if (assembly.status === 'pending') {
+      // A part for an id nobody is waiting on still costs a slot until it is discarded.
+      if (!this.pending.has(message.id)) {
+        this.assembler.discard(message.id)
+      }
+      return
+    }
+    if (assembly.status === 'failed') {
+      this.fail(message.id, new BridgeReplyRefusedError(assembly.refusal))
+      return
+    }
+    // A host `RpcFailure` resolves: it is data the caller reads, and the goldens record it.
+    this.settle(message.id, (request) => {
+      request.resolve(assembly.payload)
+    })
+  }
+
+  fail(id: string, error: unknown): void {
+    this.settle(id, (request) => {
+      request.reject(error)
+    })
+  }
+
+  /**
+   * Every pending request reaches its caller before this returns, and each one rejects
+   * delivery-unknown: the desktop may already have run it, and a caller told this was a definite
+   * send failure would offer to retry something that already happened.
+   */
+  closeAll(reason: Error = new BridgeClientClosedError()): void {
+    const error = markRpcDeliveryUnknown(reason)
+    for (const request of this.pending.values()) {
+      request.reject(error)
+    }
+    this.pending.clear()
+    this.assembler.clear()
+  }
+
+  private settle(id: string, settleWith: (request: PendingRequest) => void): void {
+    this.assembler.discard(id)
+    const request = this.pending.get(id)
+    if (request === undefined) {
+      return
+    }
+    this.pending.delete(id)
+    settleWith(request)
+  }
+}

mobile/src/mobile-web-shell/bridge/bridge-client-subscriptions.ts

@@ -0,0 +1,183 @@
+import type { BrowserScreencastFrame } from '../../transport/browser-screencast-protocol'
+import {
+  BRIDGE_PROTOCOL_VERSION,
+  type BridgeClientMessage,
+  type BridgeHostMessage
+} from './bridge-envelope'
+import { decodeBridgeScreencastFrame, type BridgeBinaryEvent } from './bridge-screencast-binary'
+
+type BridgeEventMessage = Extract<BridgeHostMessage, { type: 'event' }>
+
+/** Derived from the envelope's closed list, the same way the shell's ledger derives it: a reason
+ *  added there is a compile error here rather than one this side silently never sees. */
+export type BridgeStreamEndReason = Extract<BridgeHostMessage, { type: 'end' }>['reason']
+
+/**
+ * How far behind the page lets itself fall before it acks.
+ *
+ * The shell ends a stream at 256 unacked frames or 4 MiB. A quarter of each leaves room for the
+ * frames already in flight when an ack is posted, so a page that is keeping up never walks the
+ * shell's window down to the point where it ends a stream. `bridge-rpc-client-frames.test.ts` pins
+ * the ratio against the shell's own numbers.
+ */
+export const BRIDGE_ACK_INTERVAL_FRAMES = 64
+export const BRIDGE_ACK_INTERVAL_BYTES = 1024 * 1024
+
+/**
+ * What a listener is handed when its stream dies under it, in the shape the native client's
+ * `emitError` uses. Consumers read `type` and act on it — `host-worktree-refresh.ts` clears the flag
+ * that says the event stream is live — so a stream that merely stops delivering leaves them waiting
+ * on a replay that is never coming.
+ */
+export type BridgeStreamErrorResult = { type: 'error'; message: string; error?: unknown }
+
+export function bridgeStreamError(message: string, error?: unknown): BridgeStreamErrorResult {
+  return error === undefined ? { type: 'error', message } : { type: 'error', message, error }
+}
+
+type OpenStream = {
+  onData: (result: unknown) => void
+  onBinaryFrame?: (frame: BrowserScreencastFrame) => void
+  lastSeq: number
+  unackedFrames: number
+  unackedBytes: number
+}
+
+type SubscriptionsOptions = {
+  /** False when the frame never left the page. */
+  send: (frame: BridgeClientMessage) => boolean
+  /** A binary frame with no listener or no decodable image. Neither is recoverable in place. */
+  onDroppedBinaryFrame: () => void
+}
+
+/** Every stream the page opened, and the ack it owes the shell for each one. */
+export class BridgeClientSubscriptions {
+  private streams = new Map<string, OpenStream>()
+
+  constructor(private readonly options: SubscriptionsOptions) {}
+
+  get size(): number {
+    return this.streams.size
+  }
+
+  has(id: string): boolean {
+    return this.streams.has(id)
+  }
+
+  /** False when the `subscribe` never left the page. The shell has not heard of the stream, so
+   *  nothing will ever end it: the slot goes back here and the listener is told, which is what the
+   *  native client does with a subscribe it could not send. */
+  open(
+    id: string,
+    method: string,
+    params: unknown,
+    onData: (result: unknown) => void,
+    onBinaryFrame?: (frame: BrowserScreencastFrame) => void
+  ): boolean {
+    this.streams.set(id, { onData, onBinaryFrame, lastSeq: 0, unackedFrames: 0, unackedBytes: 0 })
+    const sent = this.options.send({
+      v: BRIDGE_PROTOCOL_VERSION,
+      type: 'subscribe',
+      id,
+      method,
+      params,
+      // Asked for only when there is something to hand the frames to, so a shell that pays to
+      // encode binary is one a listener is waiting on.
+      ...(onBinaryFrame === undefined ? {} : { wantsBinary: true })
+    })
+    if (sent) {
+      return true
+    }
+    this.streams.delete(id)
+    onData(bridgeStreamError('the subscribe could not be posted to the shell'))
+    return false
+  }
+
+  /** `bytes` is the raw frame as the shell measured it, so both sides' windows agree exactly. */
+  deliver(message: BridgeEventMessage, bytes: number): void {
+    const stream = this.streams.get(message.id)
+    if (stream === undefined) {
+      return
+    }
+    stream.lastSeq = message.seq
+    stream.unackedFrames += 1
+    stream.unackedBytes += bytes
+    // Acked before the listener runs: the frame was received and read either way, and a listener
+    // that throws must not also wedge the stream by stranding the ack behind it.
+    this.ackIfDue(message.id, stream)
+    if ('binary' in message) {
+      this.deliverBinary(stream, message.binary)
+      return
+    }
+    stream.onData(message.payload)
+  }
+
+  /** The shell already retired this stream, so nothing is posted back for it. The listener is told
+   *  before the record goes: frames that merely stop arriving are indistinguishable from a quiet
+   *  stream, and a consumer waiting on a replay would wait for the life of the document. */
+  end(id: string, message: string, error?: unknown): void {
+    const stream = this.streams.get(id)
+    if (stream === undefined) {
+      return
+    }
+    this.streams.delete(id)
+    stream.onData(bridgeStreamError(message, error))
+  }
+
+  /** The page is done with the stream. Idempotent: a second dispose posts nothing. */
+  cancel(id: string): void {
+    if (!this.streams.delete(id)) {
+      return
+    }
+    this.options.send({
+      v: BRIDGE_PROTOCOL_VERSION,
+      type: 'cancel',
+      id,
+      target: 'subscription'
+    })
+  }
+
+  /** For `close`, which is the shell's authority to tear down both sides: a cancel per stream
+   *  ahead of it would say the same thing twice. Silent, because the page asked for this one. */
+  closeAll(): void {
+    this.streams.clear()
+  }
+
+  /** For a shell replaced under the page: every stream it was serving died with it, and the
+   *  listeners are the only ones in a position to do anything about that. */
+  failAll(message: string): void {
+    // Out of the ledger before any listener runs: one that resubscribes on the way down is opening
+    // a stream against the shell that is arriving, and this loop must not take that one with it.
+    const ended = this.streams
+    this.streams = new Map()
+    for (const stream of ended.values()) {
+      stream.onData(bridgeStreamError(message))
+    }
+  }
+
+  private deliverBinary(stream: OpenStream, binary: BridgeBinaryEvent): void {
+    const onBinaryFrame = stream.onBinaryFrame
+    if (onBinaryFrame === undefined) {
+      this.options.onDroppedBinaryFrame()
+      return
+    }
+    const frame = decodeBridgeScreencastFrame(binary)
+    if (frame === null) {
+      this.options.onDroppedBinaryFrame()
+      return
+    }
+    onBinaryFrame(frame)
+  }
+
+  private ackIfDue(id: string, stream: OpenStream): void {
+    if (
+      stream.unackedFrames < BRIDGE_ACK_INTERVAL_FRAMES &&
+      stream.unackedBytes < BRIDGE_ACK_INTERVAL_BYTES
+    ) {
+      return
+    }
+    stream.unackedFrames = 0
+    stream.unackedBytes = 0
+    this.options.send({ v: BRIDGE_PROTOCOL_VERSION, type: 'ack', id, seq: stream.lastSeq })
+  }
+}

mobile/src/mobile-web-shell/bridge/bridge-port-pair-test-harness.ts

@@ -0,0 +1,160 @@
+import { createBridgeHost, type BridgeHost, type BridgeHostDiagnostic } from '../bridge-host'
+import { createFakeRpcClient, type FakeRpcClient } from '../bridge-host-test-fakes'
+import {
+  readBridgeClientMessage,
+  readBridgeHostMessage,
+  type BridgeClientMessage,
+  type BridgeHostMessage
+} from './bridge-envelope'
+import {
+  createBridgeRpcClient,
+  type BridgeRpcClient,
+  type BridgeRpcClientDiagnostic
+} from './bridge-rpc-client'
+
+/**
+ * The page and the shell wired to each other through the weakest transport that is still a
+ * transport, so a test of either one is a test of the pair.
+ *
+ * Two properties are the whole point. One FIFO per direction, because a `subscribe` that overtook a
+ * `sendRequest` would move the recorder's shared ordinal, which is what `write-ordinal.ts` exists to
+ * catch. And delivery on a microtask, the weakest async the golden runner's zero-time drains flush
+ * and the only one that moves no virtual millisecond.
+ */
+export type BridgePortPair = {
+  client: BridgeRpcClient
+  host: BridgeHost
+  rpc: FakeRpcClient
+  /** Everything each side posted, in the order it was posted, raw. */
+  toShell: string[]
+  toPage: string[]
+  diagnostics: BridgeRpcClientDiagnostic[]
+  hostDiagnostics: BridgeHostDiagnostic[]
+  /** Runs both lanes until a full round moves nothing. */
+  flush: () => Promise<void>
+  /** Read back through the reader on the receiving side, so a frame this returns is one that lands. */
+  readToShell: () => BridgeClientMessage[]
+  readToPage: () => BridgeHostMessage[]
+}
+
+export type BridgePortPairOptions = {
+  rpc?: FakeRpcClient
+  sessionId?: string
+  buildId?: string
+}
+
+type Lane = {
+  sent: string[]
+  push: (json: string) => void
+  readonly depth: number
+}
+
+function createLane(deliver: (json: string) => void): Lane {
+  const sent: string[] = []
+  const queue: string[] = []
+  let scheduled = false
+  function drain(): void {
+    scheduled = false
+    const next = queue.shift()
+    if (next === undefined) {
+      return
+    }
+    deliver(next)
+    schedule()
+  }
+  function schedule(): void {
+    if (scheduled || queue.length === 0) {
+      return
+    }
+    scheduled = true
+    void Promise.resolve().then(drain)
+  }
+  return {
+    sent,
+    push(json: string): void {
+      sent.push(json)
+      queue.push(json)
+      schedule()
+    },
+    get depth(): number {
+      return queue.length
+    }
+  }
+}
+
+function readAll<TMessage>(
+  frames: readonly string[],
+  read: (json: string) => { ok: true; message: TMessage } | { ok: false; refusal: string }
+): TMessage[] {
+  return frames.map((json) => {
+    const parsed = read(json)
+    if (!parsed.ok) {
+      throw new Error(`the other side would have refused this frame: ${parsed.refusal}`)
+    }
+    return parsed.message
+  })
+}
+
+export function createBridgePortPair(options: BridgePortPairOptions = {}): BridgePortPair {
+  const rpc = options.rpc ?? createFakeRpcClient()
+  const diagnostics: BridgeRpcClientDiagnostic[] = []
+  const hostDiagnostics: BridgeHostDiagnostic[] = []
+  let receiveOnPage: ((json: string) => void) | null = null
+
+  const toPage = createLane((json) => {
+    receiveOnPage?.(json)
+  })
+  const host = createBridgeHost({
+    client: rpc,
+    post: (json) => {
+      toPage.push(json)
+      return Promise.resolve()
+    },
+    buildId: options.buildId ?? 'build-a',
+    sessionId: options.sessionId ?? 'session-a',
+    onDiagnostic: (diagnostic) => hostDiagnostics.push(diagnostic)
+  })
+  const toShell = createLane((json) => {
+    host.receive(json)
+  })
+  const client = createBridgeRpcClient({
+    send: (json) => {
+      toShell.push(json)
+    },
+    onMessage: (handler) => {
+      receiveOnPage = handler
+      return () => {
+        receiveOnPage = null
+      }
+    },
+    onDiagnostic: (diagnostic) => diagnostics.push(diagnostic)
+  })
+
+  return {
+    client,
+    host,
+    rpc,
+    toShell: toShell.sent,
+    toPage: toPage.sent,
+    diagnostics,
+    hostDiagnostics,
+    async flush(): Promise<void> {
+      for (let round = 0; round < 64; round += 1) {
+        const moved = toShell.sent.length + toPage.sent.length
+        for (let turn = 0; turn < 8; turn += 1) {
+          await Promise.resolve()
+        }
+        const quiet =
+          toShell.depth === 0 &&
+          toPage.depth === 0 &&
+          moved === toShell.sent.length + toPage.sent.length
+        if (quiet) {
+          return
+        }
+      }
+      throw new Error('the port pair never went quiet')
+    },
+    readToShell: () => readAll(toShell.sent, readBridgeClientMessage),
+    readToPage: () => readAll(toPage.sent, readBridgeHostMessage)
+  }
+}

mobile/src/mobile-web-shell/bridge/bridge-reply-chunking.ts

@@ -127,10 +127,12 @@ const BRIDGE_MAX_ASSEMBLING_BYTES = BRIDGE_MAX_REPLY_BYTES * 4
 /**
  * Parts may arrive in any order, so they are held by index rather than appended.
  *
- * A failed id stays failed. Dropping it and starting over on the next part is what lets a sender
+ * A failed id stays failed while the page still holds it. Dropping the refusal and starting over on
- * walk past the ceiling one refusal at a time, so the refusal is remembered and every later part
+ * the next part is what would let a sender walk past the ceiling one refusal at a time, so every
- * for that id gets the same answer. `discard` is how the page says the id is finished with, which
+ * later part for that id gets the same answer instead. Nothing is remembered for long: `discard`
- * is also how it becomes usable again.
+ * reopens the id, and the page's request ledger calls it as it settles the caller, so the tombstone
+ * normally lives no longer than the rest of the reply that raised it. The bound below is for the
+ * ids nothing settles.
  *
  * The number of ids held at once is bounded by the in-flight request cap, since a reply only exists
  * for a request the page made, and their bytes together by `BRIDGE_MAX_ASSEMBLING_BYTES`. Nothing

mobile/src/mobile-web-shell/bridge/bridge-rpc-client-frames.test.ts

@@ -0,0 +1,759 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+import { BrowserScreencastOpcode } from '../../transport/browser-screencast-protocol'
+import { isRpcDeliveryUnknown } from '../../transport/rpc-delivery-ambiguity'
+import {
+  BRIDGE_MAX_MESSAGE_BYTES,
+  BRIDGE_MAX_PENDING_REQUESTS,
+  BRIDGE_MAX_SUBSCRIPTIONS
+} from './bridge-caps'
+import { BRIDGE_MAX_UNACKED_BYTES, BRIDGE_MAX_UNACKED_FRAMES } from '../bridge-host-subscriptions'
+import {
+  BRIDGE_ACK_INTERVAL_BYTES,
+  BRIDGE_ACK_INTERVAL_FRAMES
+} from './bridge-client-subscriptions'
+import {
+  BRIDGE_PROTOCOL_VERSION,
+  readBridgeClientMessage,
+  type BridgeClientMessage,
+  type BridgeHostMessage
+} from './bridge-envelope'
+import {
+  BRIDGE_READY_RETRY_MAX_MS,
+  BRIDGE_READY_RETRY_MIN_MS
+} from './bridge-client-init-handshake'
+import {
+  BridgeClientCapExceededError,
+  BridgeClientClosedError,
+  BridgeClientNotReadyError,
+  createBridgeRpcClient,
+  type BridgeRpcClientDiagnostic
+} from './bridge-rpc-client'
+
+const CONNECTION = {
+  state: 'connected',
+  reconnectAttempt: 2,
+  lastConnectedAt: 1700,
+  lastInboundAt: 1800,
+  generation: 5
+} as const
+
+const INIT: BridgeHostMessage = {
+  v: BRIDGE_PROTOCOL_VERSION,
+  type: 'init',
+  sessionId: 'session-a',
+  buildId: 'build-a',
+  connection: CONNECTION,
+  grants: { rpc: { maxPendingRequests: 64, maxSubscriptions: 32 }, native: [] }
+}
+
+type PageClientOptions = {
+  send?: (json: string) => void
+  /** A port that ignores its own unsubscribe, which is the only way to observe the read guard. */
+  keepDeliveringAfterUnsubscribe?: boolean
+}
+
+function createPageClient(options: PageClientOptions = {}) {
+  const sent: string[] = []
+  const diagnostics: BridgeRpcClientDiagnostic[] = []
+  let handler: ((json: string) => void) | null = null
+  const client = createBridgeRpcClient({
+    send: (json) => {
+      sent.push(json)
+      options.send?.(json)
+    },
+    onMessage: (received) => {
+      handler = received
+      return () => {
+        if (options.keepDeliveringAfterUnsubscribe !== true) {
+          handler = null
+        }
+      }
+    },
+    onDiagnostic: (diagnostic) => {
+      diagnostics.push(diagnostic)
+    }
+  })
+  return {
+    client,
+    sent,
+    diagnostics,
+    deliver(frame: unknown): void {
+      handler?.(JSON.stringify(frame))
+    },
+    deliverRaw(json: string): void {
+      handler?.(json)
+    },
+    frames(): BridgeClientMessage[] {
+      return sent.map((json) => {
+        const read = readBridgeClientMessage(json)
+        if (!read.ok) {
+          throw new Error(`the shell would have refused this frame: ${read.refusal}`)
+        }
+        return read.message
+      })
+    },
+    start(): void {
+      this.deliver(INIT)
+    }
+  }
+}
+
+/** Narrows what a rejection handed back, so a test reads an error rather than asserting one. */
+function readError(thrown: unknown): Error {
+  if (!(thrown instanceof Error)) {
+    throw new Error(`expected an Error, got ${typeof thrown}`)
+  }
+  return thrown
+}
+
+function eventFrame(id: string, seq: number, payload: unknown): BridgeHostMessage {
+  return { v: BRIDGE_PROTOCOL_VERSION, type: 'event', id, seq, payload }
+}
+
+/** The id the client minted for the nth exchange it opened, read back off its own frame. */
+function idOf(page: ReturnType<typeof createPageClient>, index: number): string {
+  const frame = page.frames().filter((message) => 'id' in message)[index]
+  if (frame === undefined || !('id' in frame)) {
+    throw new Error('the page opened no such exchange')
+  }
+  return frame.id
+}
+
+beforeEach(() => {
+  vi.useFakeTimers()
+})
+
+afterEach(() => {
+  vi.useRealTimers()
+})
+
+describe('bridge client handshake', () => {
+  it('asks for a session as soon as it exists', () => {
+    const page = createPageClient()
+    expect(page.frames()).toEqual([{ v: BRIDGE_PROTOCOL_VERSION, type: 'ready' }])
+  })
+
+  it('keeps asking on a widening backoff until init answers', () => {
+    const page = createPageClient()
+    vi.advanceTimersByTime(BRIDGE_READY_RETRY_MIN_MS)
+    expect(page.sent).toHaveLength(2)
+    vi.advanceTimersByTime(BRIDGE_READY_RETRY_MIN_MS)
+    expect(page.sent).toHaveLength(2)
+    vi.advanceTimersByTime(BRIDGE_READY_RETRY_MIN_MS)
+    expect(page.sent).toHaveLength(3)
+    vi.advanceTimersByTime(BRIDGE_READY_RETRY_MAX_MS * 4)
+    expect(page.sent.length).toBeGreaterThan(3)
+  })
+
+  it('asks no less often than the ceiling, however long the shell stays quiet', () => {
+    const page = createPageClient()
+    // Past the ceiling: doubling from the floor reaches it in six steps. An unclamped backoff is
+    // the same thing for a minute and then a page that gives up on a shell booting behind it.
+    vi.advanceTimersByTime(BRIDGE_READY_RETRY_MAX_MS * 4)
+    const asked = page.sent.length
+    vi.advanceTimersByTime(BRIDGE_READY_RETRY_MAX_MS)
+    expect(page.sent).toHaveLength(asked + 1)
+    vi.advanceTimersByTime(BRIDGE_READY_RETRY_MAX_MS * 10)
+    expect(page.sent).toHaveLength(asked + 11)
+  })
+
+  it('stops asking once init lands', () => {
+    const page = createPageClient()
+    page.start()
+    vi.advanceTimersByTime(BRIDGE_READY_RETRY_MAX_MS * 10)
+    expect(page.sent).toHaveLength(1)
+  })
+
+  it('keeps what it holds when the same shell answers a second time', async () => {
+    const page = createPageClient()
+    page.start()
+    const onData = vi.fn()
+    const answer = page.client.sendRequest('worktree.ps')
+    page.client.subscribe('terminal.stream', {}, onData)
+    const id = idOf(page, 1)
+    // Every `ready` is answered, so a page that re-asked before the first init landed hears two.
+    page.deliver(INIT)
+    page.deliver(eventFrame(id, 1, 'still live'))
+    expect(onData.mock.calls).toEqual([['still live']])
+    page.deliver({
+      v: BRIDGE_PROTOCOL_VERSION,
+      type: 'reply',
+      id: idOf(page, 0),
+      payload: { id: 'wire-1', ok: true, result: 'ok', _meta: { runtimeId: 'runtime-a' } }
+    })
+    await expect(answer).resolves.toMatchObject({ ok: true })
+  })
+
+  it('settles everything the shell it lost was holding before adopting the new one', async () => {
+    const page = createPageClient()
+    page.start()
+    const onData = vi.fn()
+    const answer = page.client.sendRequest('worktree.ps')
+    page.client.subscribe('terminal.stream', {}, onData)
+    // A rebuilt host under the same page: its tables are empty, so nothing the page still holds
+    // would ever be answered or ended from there.
+    page.deliver({ ...INIT, sessionId: 'session-b' })
+    const error = await answer.catch((thrown: unknown) => thrown)
+    expect(readError(error).name).toBe('BridgeShellReplacedError')
+    expect(isRpcDeliveryUnknown(error)).toBe(true)
+    expect(onData.mock.calls).toEqual([[{ type: 'error', message: expect.any(String) }]])
+    expect(page.client.getShellSession()?.sessionId).toBe('session-b')
+  })
+
+  it('reads the connection snapshot init primed it with', () => {
+    const page = createPageClient()
+    page.start()
+    expect(page.client.getState()).toBe('connected')
+    expect(page.client.getReconnectAttempt()).toBe(2)
+    expect(page.client.getLastConnectedAt()).toBe(1700)
+    expect(page.client.getLastInboundAt?.()).toBe(1800)
+    expect(page.client.getGeneration?.()).toBe(5)
+    expect(page.client.getShellSession()).toEqual({
+      sessionId: 'session-a',
+      buildId: 'build-a',
+      grants: INIT.grants
+    })
+  })
+
+  it('answers a generation the shell does not keep with a constant epoch', () => {
+    const page = createPageClient()
+    page.deliver({ ...INIT, connection: { ...CONNECTION, generation: null } })
+    expect(page.client.getGeneration?.()).toBe(0)
+  })
+
+  it('tells a waiting listener once, and a late one immediately', () => {
+    const page = createPageClient()
+    const early = vi.fn()
+    const dropped = vi.fn()
+    const release = page.client.onReady(dropped)
+    page.client.onReady(early)
+    release()
+    page.start()
+    expect(early).toHaveBeenCalledTimes(1)
+    expect(dropped).not.toHaveBeenCalled()
+    const late = vi.fn()
+    page.client.onReady(late)
+    expect(late).toHaveBeenCalledTimes(1)
+    page.deliver(INIT)
+    expect(early).toHaveBeenCalledTimes(1)
+  })
+})
+
+describe('bridge client before a session', () => {
+  it('refuses every member that would have to answer for one', () => {
+    const page = createPageClient()
+    expect(() => page.client.getState()).toThrow(BridgeClientNotReadyError)
+    expect(() => page.client.getReconnectAttempt()).toThrow(BridgeClientNotReadyError)
+    expect(() => page.client.getLastConnectedAt()).toThrow(BridgeClientNotReadyError)
+    expect(() => page.client.getLastInboundAt?.()).toThrow(BridgeClientNotReadyError)
+    expect(() => page.client.getGeneration?.()).toThrow(BridgeClientNotReadyError)
+    expect(() => page.client.sendRequest('worktree.ps')).toThrow(BridgeClientNotReadyError)
+    expect(() => page.client.subscribe('terminal.stream', {}, vi.fn())).toThrow(
+      BridgeClientNotReadyError
+    )
+    expect(() => page.client.notifyForeground()).toThrow(BridgeClientNotReadyError)
+    expect(() =>
+      page.client.updateTerminalSubscriptionViewport('t', { cols: 80, rows: 24 })
+    ).toThrow(BridgeClientNotReadyError)
+    expect(page.sent).toHaveLength(1)
+  })
+
+  it('still registers a state listener and still closes', () => {
+    const page = createPageClient()
+    const listener = vi.fn()
+    expect(() => page.client.onStateChange(listener)()).not.toThrow()
+    expect(() => {
+      page.client.close()
+    }).not.toThrow()
+  })
+
+  it('drops a state frame that beat init rather than priming from it', () => {
+    const page = createPageClient()
+    page.deliver({
+      v: BRIDGE_PROTOCOL_VERSION,
+      type: 'state',
+      connection: { ...CONNECTION, state: 'reconnecting' }
+    })
+    expect(() => page.client.getState()).toThrow(BridgeClientNotReadyError)
+    expect(page.diagnostics).toEqual([])
+  })
+})
+
+describe('bridge client after close', () => {
+  it('goes inert instead of throwing into a teardown, and posts nothing more', async () => {
+    const page = createPageClient()
+    page.start()
+    page.client.close()
+    expect(page.frames().at(-1)).toEqual({ v: BRIDGE_PROTOCOL_VERSION, type: 'close' })
+    const refused = page.client.sendRequest('worktree.ps')
+    await expect(refused).rejects.toThrow(BridgeClientClosedError)
+    expect(() => page.client.subscribe('terminal.stream', {}, vi.fn())()).not.toThrow()
+    expect(() => page.client.notifyForeground()).not.toThrow()
+    expect(() => {
+      page.client.updateTerminalSubscriptionViewport('t', { cols: 80, rows: 24 })
+    }).not.toThrow()
+    page.client.close()
+    page.deliver(INIT)
+    expect(page.sent).toHaveLength(2)
+  })
+
+  it('publishes disconnected and keeps answering the snapshot it last held', () => {
+    const page = createPageClient()
+    page.start()
+    const listener = vi.fn()
+    page.client.onStateChange(listener)
+    page.client.close()
+    expect(listener).toHaveBeenCalledWith('disconnected')
+    expect(page.client.getState()).toBe('disconnected')
+    expect(page.client.getReconnectAttempt()).toBe(CONNECTION.reconnectAttempt)
+    expect(page.client.getLastConnectedAt()).toBe(CONNECTION.lastConnectedAt)
+    expect(page.client.getLastInboundAt?.()).toBe(CONNECTION.lastInboundAt)
+    expect(page.client.getGeneration?.()).toBe(CONNECTION.generation)
+  })
+
+  it('answers nothing it never heard: a close before init leaves the getters unready', () => {
+    const page = createPageClient()
+    const listener = vi.fn()
+    page.client.onStateChange(listener)
+    page.client.close()
+    expect(listener).not.toHaveBeenCalled()
+    expect(() => page.client.getState()).toThrow(BridgeClientNotReadyError)
+  })
+
+  it('reads nothing more, even from a port that kept delivering', () => {
+    const page = createPageClient({ keepDeliveringAfterUnsubscribe: true })
+    page.start()
+    page.client.subscribe('terminal.stream', {}, vi.fn())
+    const id = idOf(page, 0)
+    page.client.close()
+    page.deliver(INIT)
+    page.deliver(eventFrame(id, 1, 'late'))
+    page.deliverRaw('{ not json')
+    expect(page.diagnostics).toEqual([])
+    expect(page.client.getState()).toBe('disconnected')
+  })
+
+  it('says goodbye once, without a cancel for each stream it owned', () => {
+    const page = createPageClient()
+    page.start()
+    page.client.subscribe('terminal.stream', {}, vi.fn())
+    page.client.subscribe('terminal.stream', {}, vi.fn())
+    page.client.close()
+    expect(page.frames().filter((frame) => frame.type === 'cancel')).toEqual([])
+    expect(page.frames().filter((frame) => frame.type === 'close')).toHaveLength(1)
+  })
+})
+
+describe('bridge client replies', () => {
+  it('rejects with the class and the delivery mark the shell captured', async () => {
+    const page = createPageClient()
+    page.start()
+    const answer = page.client.sendRequest('worktree.ps')
+    page.deliver({
+      v: BRIDGE_PROTOCOL_VERSION,
+      type: 'error',
+      id: idOf(page, 0),
+      error: {
+        category: 'RpcTimeoutError',
+        message: 'timed out',
+        isRpcDeliveryUnknown: true,
+        code: 'ETIMEDOUT',
+        cause: { category: 'Error', message: 'socket closed', isRpcDeliveryUnknown: false }
+      }
+    })
+    const error = await answer.catch((thrown: unknown) => thrown)
+    expect(error).toBeInstanceOf(Error)
+    expect(readError(error).name).toBe('RpcTimeoutError')
+    expect(isRpcDeliveryUnknown(error)).toBe(true)
+    expect(readError(readError(error).cause).message).toBe('socket closed')
+  })
+
+  it('rejects a reply the assembler refuses', async () => {
+    const page = createPageClient()
+    page.start()
+    const answer = page.client.sendRequest('worktree.ps')
+    const id = idOf(page, 0)
+    const part = {
+      v: BRIDGE_PROTOCOL_VERSION,
+      type: 'reply',
+      id,
+      part: { i: 0, of: 2 },
+      chunk: '{'
+    }
+    page.deliver(part)
+    page.deliver(part)
+    await expect(answer).rejects.toThrow('duplicate-part')
+  })
+
+  it('drops a reply or an error for an id it never opened, and says so', () => {
+    const page = createPageClient()
+    page.start()
+    const stranger = 'z'.repeat(22)
+    page.deliver({
+      v: BRIDGE_PROTOCOL_VERSION,
+      type: 'reply',
+      id: stranger,
+      payload: { id: stranger, ok: true, result: 1, _meta: { runtimeId: 'runtime-a' } }
+    })
+    page.deliver({
+      v: BRIDGE_PROTOCOL_VERSION,
+      type: 'error',
+      id: stranger,
+      error: { category: 'Error', message: 'gone', isRpcDeliveryUnknown: false }
+    })
+    expect(page.diagnostics).toEqual([{ kind: 'unknown-id' }, { kind: 'unknown-id' }])
+  })
+
+  it('frees the assembler slot of every id nobody is waiting on', async () => {
+    const page = createPageClient()
+    page.start()
+    const answer = page.client.sendRequest('worktree.ps')
+    const id = idOf(page, 0)
+    for (let index = 0; index < BRIDGE_MAX_PENDING_REQUESTS * 2; index += 1) {
+      page.deliver({
+        v: BRIDGE_PROTOCOL_VERSION,
+        type: 'reply',
+        id: index.toString(36).padStart(22, 'z'),
+        part: { i: 0, of: 2 },
+        chunk: '{"a":'
+      })
+    }
+    const payload = { id, ok: true, result: 7, _meta: { runtimeId: 'runtime-a' } }
+    const serialized = JSON.stringify(payload)
+    const cut = Math.floor(serialized.length / 2)
+    page.deliver({
+      v: BRIDGE_PROTOCOL_VERSION,
+      type: 'reply',
+      id,
+      part: { i: 0, of: 2 },
+      chunk: serialized.slice(0, cut)
+    })
+    page.deliver({
+      v: BRIDGE_PROTOCOL_VERSION,
+      type: 'reply',
+      id,
+      part: { i: 1, of: 2 },
+      chunk: serialized.slice(cut)
+    })
+    await expect(answer).resolves.toEqual(payload)
+  })
+
+  it('gives back the assembler slot of every id it settles', async () => {
+    const page = createPageClient()
+    page.start()
+    const settled: Promise<unknown>[] = []
+    for (let index = 0; index < BRIDGE_MAX_PENDING_REQUESTS; index += 1) {
+      const abandoned = page.client.sendRequest('worktree.ps')
+      const id = idOf(page, index)
+      page.deliver({
+        v: BRIDGE_PROTOCOL_VERSION,
+        type: 'reply',
+        id,
+        part: { i: 0, of: 2 },
+        chunk: '{"a":'
+      })
+      page.deliver({
+        v: BRIDGE_PROTOCOL_VERSION,
+        type: 'error',
+        id,
+        error: { category: 'Error', message: 'gone', isRpcDeliveryUnknown: false }
+      })
+      settled.push(abandoned.catch(() => undefined))
+    }
+    const answer = page.client.sendRequest('worktree.ps')
+    const id = idOf(page, BRIDGE_MAX_PENDING_REQUESTS)
+    const payload = { id, ok: true, result: 'assembled', _meta: { runtimeId: 'runtime-a' } }
+    const serialized = JSON.stringify(payload)
+    const cut = Math.floor(serialized.length / 2)
+    page.deliver({
+      v: BRIDGE_PROTOCOL_VERSION,
+      type: 'reply',
+      id,
+      part: { i: 0, of: 2 },
+      chunk: serialized.slice(0, cut)
+    })
+    page.deliver({
+      v: BRIDGE_PROTOCOL_VERSION,
+      type: 'reply',
+      id,
+      part: { i: 1, of: 2 },
+      chunk: serialized.slice(cut)
+    })
+    await expect(answer).resolves.toEqual(payload)
+    await Promise.all(settled)
+  })
+})
+
+describe('bridge client refusals and send failures', () => {
+  it('reports a frame its own reader will not take, and changes nothing', () => {
+    const page = createPageClient()
+    page.start()
+    page.deliverRaw('{ not json')
+    page.deliverRaw(JSON.stringify({ v: 99, type: 'state' }))
+    page.deliverRaw(`"${'z'.repeat(BRIDGE_MAX_MESSAGE_BYTES)}"`)
+    expect(page.diagnostics).toEqual([
+      { kind: 'refused', refusal: 'malformed-json' },
+      { kind: 'refused', refusal: 'unrecognised-message' },
+      { kind: 'refused', refusal: 'oversized' }
+    ])
+    expect(page.client.getState()).toBe('connected')
+  })
+
+  it('fails a request whose frame never left the page, without the delivery mark', async () => {
+    let live = true
+    const page = createPageClient({
+      send: () => {
+        if (!live) {
+          throw new Error('the port is gone')
+        }
+      }
+    })
+    page.start()
+    live = false
+    const answer = page.client.sendRequest('worktree.ps')
+    const error = await answer.catch((thrown: unknown) => thrown)
+    expect(readError(error).name).toBe('BridgeSendFailedError')
+    expect(isRpcDeliveryUnknown(error)).toBe(false)
+    expect(page.diagnostics.at(-1)).toEqual({
+      kind: 'send-failed',
+      error: expect.any(Error)
+    })
+  })
+})
+
+describe('bridge client caps', () => {
+  it('refuses the request past the shell grant without a round trip', async () => {
+    const page = createPageClient()
+    page.start()
+    const answers: Promise<unknown>[] = []
+    for (let index = 0; index < BRIDGE_MAX_PENDING_REQUESTS; index += 1) {
+      answers.push(page.client.sendRequest('worktree.ps'))
+    }
+    const refused = page.client.sendRequest('worktree.ps')
+    await expect(refused).rejects.toThrow(BridgeClientCapExceededError)
+    expect(page.sent).toHaveLength(1 + BRIDGE_MAX_PENDING_REQUESTS)
+    page.client.close()
+    await Promise.allSettled(answers)
+  })
+
+  it('frees the page slot when the subscribe frame never left the page', () => {
+    let live = true
+    const page = createPageClient({
+      send: () => {
+        if (!live) {
+          throw new Error('the port is gone')
+        }
+      }
+    })
+    page.start()
+    live = false
+    const onData = vi.fn()
+    // Every one of these is a slot the shell was never told about, and nothing will ever end it.
+    for (let index = 0; index < BRIDGE_MAX_SUBSCRIPTIONS; index += 1) {
+      page.client.subscribe('terminal.stream', {}, onData)
+    }
+    expect(onData).toHaveBeenCalledTimes(BRIDGE_MAX_SUBSCRIPTIONS)
+    expect(onData.mock.calls.at(-1)?.[0]).toEqual({ type: 'error', message: expect.any(String) })
+    expect(page.diagnostics).toHaveLength(BRIDGE_MAX_SUBSCRIPTIONS)
+    live = true
+    // Short of this, the page is at its cap for the life of the document: only a reload clears it.
+    const dispose = page.client.subscribe('terminal.stream', {}, vi.fn())
+    dispose()
+    expect(page.frames().filter((frame) => frame.type === 'cancel')).toHaveLength(1)
+  })
+
+  it('refuses the subscription past the shell grant at the call site', () => {
+    const page = createPageClient()
+    page.start()
+    for (let index = 0; index < BRIDGE_MAX_SUBSCRIPTIONS; index += 1) {
+      page.client.subscribe('terminal.stream', {}, vi.fn())
+    }
+    expect(() => page.client.subscribe('terminal.stream', {}, vi.fn())).toThrow(
+      BridgeClientCapExceededError
+    )
+    expect(page.sent).toHaveLength(1 + BRIDGE_MAX_SUBSCRIPTIONS)
+  })
+})
+
+describe('bridge client acks', () => {
+  it('stays well inside the window the shell ends a stream at', () => {
+    // The shell's own numbers, not a copy of them: a window narrowed there has to fail here.
+    expect(BRIDGE_ACK_INTERVAL_FRAMES * 4).toBeLessThanOrEqual(BRIDGE_MAX_UNACKED_FRAMES)
+    expect(BRIDGE_ACK_INTERVAL_BYTES * 4).toBeLessThanOrEqual(BRIDGE_MAX_UNACKED_BYTES)
+  })
+
+  it('acks the last seq it read once the frame interval is due', () => {
+    const page = createPageClient()
+    page.start()
+    page.client.subscribe('terminal.stream', {}, vi.fn())
+    const id = idOf(page, 0)
+    for (let seq = 1; seq < BRIDGE_ACK_INTERVAL_FRAMES; seq += 1) {
+      page.deliver(eventFrame(id, seq, seq))
+    }
+    expect(page.frames().filter((frame) => frame.type === 'ack')).toEqual([])
+    page.deliver(eventFrame(id, BRIDGE_ACK_INTERVAL_FRAMES, 'last'))
+    expect(page.frames().filter((frame) => frame.type === 'ack')).toEqual([
+      { v: BRIDGE_PROTOCOL_VERSION, type: 'ack', id, seq: BRIDGE_ACK_INTERVAL_FRAMES }
+    ])
+  })
+
+  it('acks early when the bytes are due before the frames are', () => {
+    const page = createPageClient()
+    page.start()
+    page.client.subscribe('terminal.stream', {}, vi.fn())
+    const id = idOf(page, 0)
+    const heavy = 'z'.repeat(BRIDGE_MAX_MESSAGE_BYTES - 1024)
+    page.deliver(eventFrame(id, 1, heavy))
+    page.deliver(eventFrame(id, 2, heavy))
+    expect(page.frames().filter((frame) => frame.type === 'ack')).toEqual([
+      { v: BRIDGE_PROTOCOL_VERSION, type: 'ack', id, seq: 2 }
+    ])
+  })
+
+  it('acks a frame whose listener throws, so a listener bug cannot wedge the stream', () => {
+    const page = createPageClient()
+    page.start()
+    page.client.subscribe('terminal.stream', {}, () => {
+      throw new Error('listener bug')
+    })
+    const id = idOf(page, 0)
+    for (let seq = 1; seq <= BRIDGE_ACK_INTERVAL_FRAMES; seq += 1) {
+      expect(() => page.deliver(eventFrame(id, seq, seq))).toThrow('listener bug')
+    }
+    expect(page.frames().filter((frame) => frame.type === 'ack')).toHaveLength(1)
+  })
+
+  it('ignores an event for a stream it already disposed', () => {
+    const page = createPageClient()
+    page.start()
+    const onData = vi.fn()
+    const dispose = page.client.subscribe('terminal.stream', {}, onData)
+    const id = idOf(page, 0)
+    dispose()
+    dispose()
+    page.deliver(eventFrame(id, 1, 'late'))
+    expect(onData).not.toHaveBeenCalled()
+    expect(page.frames().filter((frame) => frame.type === 'cancel')).toHaveLength(1)
+  })
+
+  it('posts no cancel for a stream the shell ended before the page let go', () => {
+    const page = createPageClient()
+    page.start()
+    const dispose = page.client.subscribe('terminal.stream', {}, vi.fn())
+    const id = idOf(page, 0)
+    page.deliver({ v: BRIDGE_PROTOCOL_VERSION, type: 'end', id, reason: 'closed' })
+    // The screen unmounts on its own schedule, which is routinely after the shell gave up.
+    dispose()
+    expect(page.frames().filter((frame) => frame.type === 'cancel')).toEqual([])
+  })
+
+  it('retires a stream the shell ended, tells the listener, and reports why', () => {
+    const page = createPageClient()
+    page.start()
+    const onData = vi.fn()
+    page.client.subscribe('terminal.stream', {}, onData)
+    const id = idOf(page, 0)
+    page.deliver({ v: BRIDGE_PROTOCOL_VERSION, type: 'end', id, reason: 'overflow' })
+    page.deliver(eventFrame(id, 1, 'after the end'))
+    // The terminal result is the only thing a consumer hears. `host-worktree-refresh.ts` reads it
+    // to clear the flag that says the event stream is live; without it the list never refreshes
+    // again, because frames that stop arriving look exactly like a stream with nothing to say.
+    expect(onData.mock.calls).toEqual([[{ type: 'error', message: expect.any(String) }]])
+    expect(page.diagnostics).toEqual([{ kind: 'stream-ended', reason: 'overflow' }])
+    expect(page.frames().filter((frame) => frame.type === 'cancel')).toEqual([])
+  })
+
+  it('tells the listener nothing when the page itself let the stream go', () => {
+    const page = createPageClient()
+    page.start()
+    const onData = vi.fn()
+    const dispose = page.client.subscribe('terminal.stream', {}, onData)
+    dispose()
+    // The caller that disposed is the one that would hear it, and it has already moved on.
+    expect(onData).not.toHaveBeenCalled()
+    expect(page.frames().filter((frame) => frame.type === 'cancel')).toHaveLength(1)
+  })
+})
+
+describe('bridge client binary frames', () => {
+  const image = Uint8Array.from([137, 80, 78, 71, 13, 10, 26, 10])
+  const b64 = btoa(String.fromCharCode(...image))
+
+  function binaryFrame(id: string, b64Image: string): BridgeHostMessage {
+    return {
+      v: BRIDGE_PROTOCOL_VERSION,
+      type: 'event',
+      id,
+      seq: 1,
+      binary: { b64: b64Image, format: 'png', frameSeq: 41, metadata: { imageWidth: 8 } }
+    }
+  }
+
+  it('asks for binary only when a listener is there to read it', () => {
+    const page = createPageClient()
+    page.start()
+    page.client.subscribe('browser.screencast', {}, vi.fn())
+    page.client.subscribe('browser.screencast', {}, vi.fn(), { onBinaryFrame: vi.fn() })
+    const opened = page.frames().filter((frame) => frame.type === 'subscribe')
+    expect(opened[0]).not.toHaveProperty('wantsBinary')
+    expect(opened[1]).toHaveProperty('wantsBinary', true)
+  })
+
+  it('decodes to the frame a native listener would have been handed', () => {
+    const page = createPageClient()
+    page.start()
+    const onBinaryFrame = vi.fn()
+    page.client.subscribe('browser.screencast', {}, vi.fn(), { onBinaryFrame })
+    page.deliver(binaryFrame(idOf(page, 0), b64))
+    expect(onBinaryFrame).toHaveBeenCalledWith({
+      opcode: BrowserScreencastOpcode.Frame,
+      seq: 41,
+      format: 'png',
+      metadata: { imageWidth: 8 },
+      image
+    })
+  })
+
+  it('carries every metadata field the shell measured', () => {
+    const page = createPageClient()
+    page.start()
+    const onBinaryFrame = vi.fn()
+    page.client.subscribe('browser.screencast', {}, vi.fn(), { onBinaryFrame })
+    const metadata = {
+      offsetTop: 1,
+      pageScaleFactor: 2,
+      deviceWidth: 3,
+      deviceHeight: 4,
+      imageWidth: 5,
+      imageHeight: 6,
+      scrollOffsetX: 7,
+      scrollOffsetY: 8,
+      timestamp: 9
+    }
+    page.deliver({
+      v: BRIDGE_PROTOCOL_VERSION,
+      type: 'event',
+      id: idOf(page, 0),
+      seq: 1,
+      binary: { b64, format: 'jpeg', frameSeq: 0, metadata }
+    })
+    expect(onBinaryFrame).toHaveBeenCalledWith(
+      expect.objectContaining({ format: 'jpeg', seq: 0, metadata })
+    )
+  })
+
+  it('drops a frame with no listener and one it cannot decode', () => {
+    const page = createPageClient()
+    page.start()
+    page.client.subscribe('browser.screencast', {}, vi.fn())
+    page.deliver(binaryFrame(idOf(page, 0), b64))
+    const onBinaryFrame = vi.fn()
+    page.client.subscribe('browser.screencast', {}, vi.fn(), { onBinaryFrame })
+    page.deliver(binaryFrame(idOf(page, 1), '!!not base64!!'))
+    expect(onBinaryFrame).not.toHaveBeenCalled()
+    expect(page.diagnostics).toEqual([
+      { kind: 'binary-frame-dropped' },
+      { kind: 'binary-frame-dropped' }
+    ])
+  })
+})

mobile/src/mobile-web-shell/bridge/bridge-rpc-client.test.ts

@@ -0,0 +1,264 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+import {
+  isRpcDeliveryUnknown,
+  markRpcDeliveryUnknown
+} from '../../transport/rpc-delivery-ambiguity'
+import type { RpcResponse } from '../../transport/types'
+import { createFakeRpcClient } from '../bridge-host-test-fakes'
+import { BRIDGE_MAX_MESSAGE_BYTES, BRIDGE_MAX_SUBSCRIPTIONS } from './bridge-caps'
+import { createBridgePortPair, type BridgePortPair } from './bridge-port-pair-test-harness'
+
+/**
+ * The page's client and the shell's host, over one FIFO per direction.
+ *
+ * That `createBridgeRpcClient` returns an `RpcClient` is the type system's job and it is already
+ * done; what a test has to prove is that each member still means the same thing after a round trip,
+ * because the screens above it cannot tell which client they are holding.
+ */
+
+function success(id: string, result: unknown): RpcResponse {
+  return { id, ok: true, result, _meta: { runtimeId: 'runtime-a' } }
+}
+
+/** Narrows what a rejection handed back, so a test reads an error rather than asserting one. */
+function readError(thrown: unknown): Error {
+  if (!(thrown instanceof Error)) {
+    throw new Error(`expected an Error, got ${typeof thrown}`)
+  }
+  return thrown
+}
+
+async function ready(pair: BridgePortPair): Promise<BridgePortPair> {
+  await pair.flush()
+  return pair
+}
+
+beforeEach(() => {
+  vi.useFakeTimers()
+})
+
+afterEach(() => {
+  vi.useRealTimers()
+})
+
+describe('bridge round trip: requests', () => {
+  it('reaches the shell with the arity the page called with', async () => {
+    const pair = await ready(createBridgePortPair())
+    void pair.client.sendRequest('worktree.ps')
+    void pair.client.sendRequest('worktree.ps', { host: 'a' })
+    void pair.client.sendRequest('worktree.ps', { host: 'a' }, { timeoutMs: 50 })
+    await pair.flush()
+    expect(pair.rpc.requests.map((request) => request.args)).toEqual([
+      ['worktree.ps'],
+      ['worktree.ps', { host: 'a' }],
+      ['worktree.ps', { host: 'a' }, { timeoutMs: 50 }]
+    ])
+  })
+
+  it('resolves the response the shell answered with, field for field', async () => {
+    const pair = await ready(createBridgePortPair())
+    const answer = pair.client.sendRequest('worktree.ps')
+    await pair.flush()
+    const response: RpcResponse = {
+      id: 'shell-side-id',
+      ok: true,
+      result: { rows: [1, 2, 3] },
+      streaming: true,
+      _meta: { runtimeId: 'runtime-a' }
+    }
+    pair.rpc.requests[0]?.resolve(response)
+    await pair.flush()
+    await expect(answer).resolves.toEqual(response)
+  })
+
+  it('resolves a host failure, because a failure is data and not a rejection', async () => {
+    const pair = await ready(createBridgePortPair())
+    const answer = pair.client.sendRequest('worktree.ps')
+    await pair.flush()
+    const failure: RpcResponse = {
+      id: 'shell-side-id',
+      ok: false,
+      error: { code: 'not_found', message: 'no such worktree', data: { host: 'a' } },
+      _meta: { runtimeId: 'runtime-a' }
+    }
+    pair.rpc.requests[0]?.resolve(failure)
+    await pair.flush()
+    await expect(answer).resolves.toEqual(failure)
+  })
+
+  it('rejects with the class, the code and the delivery mark the shell captured', async () => {
+    const pair = await ready(createBridgePortPair())
+    const answer = pair.client.sendRequest('worktree.ps')
+    await pair.flush()
+    class RpcTimeoutError extends Error {
+      code = 'ETIMEDOUT'
+    }
+    const thrown = markRpcDeliveryUnknown(new RpcTimeoutError('timed out after 50ms'))
+    thrown.cause = new Error('socket closed')
+    pair.rpc.requests[0]?.reject(thrown)
+    await pair.flush()
+    const error = await answer.catch((caught: unknown) => caught)
+    expect(error).toBeInstanceOf(Error)
+    expect(readError(error).name).toBe('RpcTimeoutError')
+    expect(readError(error).message).toBe('timed out after 50ms')
+    expect(isRpcDeliveryUnknown(error)).toBe(true)
+    expect(readError(readError(error).cause).message).toBe('socket closed')
+  })
+
+  it('reassembles a reply too big for one frame', async () => {
+    const pair = await ready(createBridgePortPair())
+    const answer = pair.client.sendRequest('source-control.diff')
+    await pair.flush()
+    const result = { diff: 'z'.repeat(BRIDGE_MAX_MESSAGE_BYTES + 60_000) }
+    pair.rpc.requests[0]?.resolve(success('shell-side-id', result))
+    await pair.flush()
+    await expect(answer).resolves.toEqual(success('shell-side-id', result))
+    expect(pair.toPage.length).toBeGreaterThan(2)
+  })
+})
+
+describe('bridge round trip: subscriptions', () => {
+  it('streams what the shell emits and stops when the page disposes', async () => {
+    const pair = await ready(createBridgePortPair())
+    const onData = vi.fn()
+    const dispose = pair.client.subscribe('terminal.stream', { terminal: 't' }, onData)
+    await pair.flush()
+    expect(pair.rpc.streams[0]?.method).toBe('terminal.stream')
+    expect(pair.rpc.streams[0]?.params).toEqual({ terminal: 't' })
+    pair.rpc.streams[0]?.emit({ type: 'data', chunk: 'hello' })
+    await pair.flush()
+    expect(onData).toHaveBeenCalledWith({ type: 'data', chunk: 'hello' })
+    dispose()
+    await pair.flush()
+    expect(pair.rpc.streams[0]?.unsubscribes).toBe(1)
+    pair.rpc.streams[0]?.emit({ type: 'data', chunk: 'after' })
+    await pair.flush()
+    expect(onData).toHaveBeenCalledTimes(1)
+  })
+
+  it('frees the page slot when the shell refuses the subscribe', async () => {
+    const pair = await ready(createBridgePortPair())
+    const onData = vi.fn()
+    const refuse = vi.spyOn(pair.rpc, 'subscribe').mockImplementation(() => {
+      throw new Error('the terminal is gone')
+    })
+    pair.client.subscribe('terminal.stream', { terminal: 't' }, onData)
+    await pair.flush()
+    refuse.mockRestore()
+    expect(pair.diagnostics).toEqual([
+      { kind: 'stream-failed', error: expect.objectContaining({ message: 'the terminal is gone' }) }
+    ])
+    // The shell's own message reaches the listener, the way the native client passes one through.
+    expect(onData.mock.calls).toEqual([
+      [{ type: 'error', message: 'the terminal is gone', error: expect.any(Error) }]
+    ])
+    // A leaked slot is invisible until the page reaches its own cap, so that is where it is read.
+    for (let index = 0; index < BRIDGE_MAX_SUBSCRIPTIONS; index += 1) {
+      pair.client.subscribe('terminal.stream', {}, vi.fn())
+    }
+    await pair.flush()
+    expect(pair.rpc.streams).toHaveLength(BRIDGE_MAX_SUBSCRIPTIONS)
+  })
+
+  it('keeps a long stream alive, because the acks free the shell window', async () => {
+    const pair = await ready(createBridgePortPair())
+    const onData = vi.fn()
+    pair.client.subscribe('terminal.stream', {}, onData)
+    await pair.flush()
+    for (let batch = 0; batch < 8; batch += 1) {
+      for (let frame = 0; frame < 50; frame += 1) {
+        pair.rpc.streams[0]?.emit(`frame-${batch}-${frame}`)
+      }
+      await pair.flush()
+    }
+    expect(onData).toHaveBeenCalledTimes(400)
+    expect(pair.diagnostics).toEqual([])
+  })
+
+  it('ends the stream when the page never gets a chance to ack', async () => {
+    const pair = await ready(createBridgePortPair())
+    const onData = vi.fn()
+    pair.client.subscribe('terminal.stream', {}, onData)
+    await pair.flush()
+    for (let frame = 0; frame < 400; frame += 1) {
+      pair.rpc.streams[0]?.emit(`frame-${frame}`)
+    }
+    await pair.flush()
+    expect(pair.diagnostics).toEqual([{ kind: 'stream-ended', reason: 'overflow' }])
+    expect(onData.mock.calls.length).toBeLessThan(400)
+  })
+})
+
+describe('bridge round trip: notifications and state', () => {
+  it('carries both notifies to the shell client, with the arity each was called with', async () => {
+    const pair = await ready(createBridgePortPair())
+    pair.client.notifyForeground()
+    pair.client.notifyForeground('app-resume')
+    pair.client.updateTerminalSubscriptionViewport('terminal-a', { cols: 120, rows: 40 })
+    await pair.flush()
+    expect(pair.rpc.foregroundCalls).toEqual([[], ['app-resume']])
+    expect(pair.rpc.viewports).toEqual([{ terminal: 'terminal-a', cols: 120, rows: 40 }])
+  })
+
+  it('reads the shell client through init and fans out every change after it', async () => {
+    const rpc = createFakeRpcClient({
+      getState: () => 'reconnecting',
+      getReconnectAttempt: () => 3,
+      getLastConnectedAt: () => 1234,
+      getLastInboundAt: () => 5678,
+      getGeneration: () => 9
+    })
+    const pair = await ready(createBridgePortPair({ rpc }))
+    expect(pair.client.getState()).toBe('reconnecting')
+    expect(pair.client.getReconnectAttempt()).toBe(3)
+    expect(pair.client.getLastConnectedAt()).toBe(1234)
+    expect(pair.client.getLastInboundAt?.()).toBe(5678)
+    expect(pair.client.getGeneration?.()).toBe(9)
+    const listener = vi.fn()
+    const release = pair.client.onStateChange(listener)
+    rpc.pushState('connected')
+    await pair.flush()
+    expect(listener).toHaveBeenCalledWith('connected')
+    expect(pair.client.getState()).toBe('connected')
+    release()
+    rpc.pushState('disconnected')
+    await pair.flush()
+    expect(listener).toHaveBeenCalledTimes(1)
+  })
+
+  it('refuses a snapshot from a shell that was rebuilt, and asks for a fresh init', async () => {
+    let generation = 5
+    const rpc = createFakeRpcClient({ getGeneration: () => generation })
+    const pair = await ready(createBridgePortPair({ rpc }))
+    const listener = vi.fn()
+    pair.client.onStateChange(listener)
+    const asked = pair.readToShell().filter((frame) => frame.type === 'ready').length
+    generation = 2
+    rpc.pushState('reconnecting')
+    await pair.flush()
+    expect(pair.diagnostics).toEqual([{ kind: 'state-out-of-order' }])
+    expect(listener).not.toHaveBeenCalled()
+    expect(pair.readToShell().filter((frame) => frame.type === 'ready').length).toBe(asked + 1)
+    // The fresh init is what re-primes the cache; the refused frame never touched it.
+    expect(pair.client.getState()).toBe('connected')
+    expect(pair.client.getGeneration?.()).toBe(2)
+  })
+})
+
+describe('bridge round trip: close', () => {
+  it('settles pendings delivery-unknown, retires the streams, and leaves the shell client open', async () => {
+    const pair = await ready(createBridgePortPair())
+    const closeShellClient = vi.spyOn(pair.rpc, 'close')
+    const answer = pair.client.sendRequest('worktree.ps')
+    pair.client.subscribe('terminal.stream', {}, vi.fn())
+    await pair.flush()
+    pair.client.close()
+    const error = await answer.catch((caught: unknown) => caught)
+    expect(readError(error).name).toBe('BridgeClientClosedError')
+    expect(isRpcDeliveryUnknown(error)).toBe(true)
+    await pair.flush()
+    expect(pair.rpc.streams[0]?.unsubscribes).toBe(1)
+    expect(closeShellClient).not.toHaveBeenCalled()
+    expect(pair.readToShell().at(-1)).toEqual({ v: 1, type: 'close' })
+  })
+})

mobile/src/mobile-web-shell/bridge/bridge-rpc-client.ts

@@ -0,0 +1,382 @@
+import type { BrowserScreencastFrame } from '../../transport/browser-screencast-protocol'
+import type { RpcClient, SendRequestOptions } from '../../transport/rpc-client'
+import type { ConnectionState, ForegroundNudgeReason, RpcResponse } from '../../transport/types'
+import {
+  BRIDGE_MAX_PENDING_REQUESTS,
+  BRIDGE_MAX_SUBSCRIPTIONS,
+  utf8ByteLength,
+  type BridgeRefusal
+} from './bridge-caps'
+import { BridgeConnectionCache } from './bridge-client-connection-cache'
+import { createBridgeInitHandshake } from './bridge-client-init-handshake'
+import {
+  BridgeClientCapExceededError,
+  BridgeClientClosedError,
+  BridgeClientNotReadyError,
+  BridgeSendFailedError,
+  BridgeShellReplacedError
+} from './bridge-client-errors'
+import { BridgeClientRequests } from './bridge-client-requests'
+import {
+  BridgeClientSubscriptions,
+  type BridgeStreamEndReason
+} from './bridge-client-subscriptions'
+import {
+  BRIDGE_PROTOCOL_VERSION,
+  readBridgeHostMessage,
+  type BridgeClientMessage,
+  type BridgeConnectionSnapshot,
+  type BridgeGrants,
+  type BridgeHostMessage
+} from './bridge-envelope'
+import { reconstructBridgeError } from './bridge-error-capture'
+
+export {
+  BridgeClientCapExceededError,
+  BridgeClientClosedError,
+  BridgeClientNotReadyError,
+  BridgeReplyRefusedError,
+  BridgeSendFailedError,
+  BridgeShellReplacedError
+} from './bridge-client-errors'
+
+/** Base64url, and the length the envelope's id pattern requires. Base36 digits are a subset of it. */
+const BRIDGE_ID_CHARS = 22
+
+/** Nothing here is recoverable in place; each is worth a line in a log and none is retried. */
+export type BridgeRpcClientDiagnostic =
+  | { kind: 'refused'; refusal: BridgeRefusal }
+  | { kind: 'send-failed'; error: unknown }
+  | { kind: 'stream-ended'; reason: BridgeStreamEndReason }
+  | { kind: 'stream-failed'; error: unknown }
+  | { kind: 'state-out-of-order' }
+  | { kind: 'binary-frame-dropped' }
+  | { kind: 'unknown-id' }
+
+/** What `init` said this page is attached to. `grants` is what a call site checks before it posts. */
+export type BridgeShellSession = {
+  sessionId: string
+  buildId: string
+  grants: BridgeGrants
+}
+
+export type BridgeRpcClientOptions = {
+  /** Posts one frame to the shell. May throw; nothing about returning proves delivery. */
+  send: (json: string) => void
+  onMessage: (handler: (json: string) => void) => () => void
+  onDiagnostic?: (diagnostic: BridgeRpcClientDiagnostic) => void
+}
+
+export type BridgeRpcClient = RpcClient & {
+  /** Fires once `init` has landed, immediately if it already has. Mount no screen before it. */
+  onReady: (listener: () => void) => () => void
+  getShellSession: () => BridgeShellSession | null
+}
+
+/**
+ * The page's `RpcClient`, which is a bridge and not a socket.
+ *
+ * Every member of the native contract is here, so `runRpcOperation` and the screens above it never
+ * learn which one they hold. Two properties make that honest. The getters are synchronous reads of a
+ * cache primed by `init`, because screens read them during render and an async read changes what the
+ * first render sees. And `close` never closes the shell's client: that one is shared with the native
+ * screens and the host catalog, so the page settles what it owns and says goodbye.
+ *
+ * Nothing may be called before `init`. The alternative is a stub answering `connecting` to a screen
+ * that then records the wrong first render, so a call arriving early throws instead. After `close`
+ * the opposite rule holds: every member goes inert and the getters keep answering the snapshot the
+ * page last held, marked `disconnected`, because an unmounting screen calls into a path with no
+ * catch on it. A stream the shell refuses or ends is not thrown anywhere either; it arrives as a
+ * diagnostic, which is the only channel `subscribe` leaves open once it has handed back a dispose.
+ */
+export function createBridgeRpcClient(options: BridgeRpcClientOptions): BridgeRpcClient {
+  const requests = new BridgeClientRequests()
+  const cache = new BridgeConnectionCache()
+  const readyListeners = new Set<() => void>()
+  let session: BridgeShellSession | null = null
+  let closed = false
+  let idCounter = 0
+
+  function report(diagnostic: BridgeRpcClientDiagnostic): void {
+    options.onDiagnostic?.(diagnostic)
+  }
+
+  /** False when the frame never left. Every value in a page frame is one the caller handed in, so
+   *  the throw this catches is the port's, never `JSON.stringify`'s. */
+  function sendFrame(frame: BridgeClientMessage): boolean {
+    try {
+      options.send(JSON.stringify(frame))
+      return true
+    } catch (error) {
+      report({ kind: 'send-failed', error })
+      return false
+    }
+  }
+
+  // Counted rather than random: a recorded run replays the same ids, and one page holds one client,
+  // so a counter is already unique across everything the shell is asked to keep in flight.
+  function nextId(): string {
+    idCounter += 1
+    return idCounter.toString(36).padStart(BRIDGE_ID_CHARS, '0')
+  }
+
+  const subscriptions = new BridgeClientSubscriptions({
+    send: (frame) => sendFrame(frame),
+    onDroppedBinaryFrame: () => {
+      report({ kind: 'binary-frame-dropped' })
+    }
+  })
+
+  const handshake = createBridgeInitHandshake(() => {
+    sendFrame({ v: BRIDGE_PROTOCOL_VERSION, type: 'ready' })
+  })
+
+  /**
+   * A call before `init` is a mount-order bug and throws. A call after `close` is not: an unmounting
+   * screen posts one more nudge on its way out, and the native clients answer those inertly rather
+   * than throwing into a teardown path nobody wrote a catch for. Each member below says what inert
+   * means for its own return type.
+   */
+  function requireSession(): void {
+    if (session === null && !closed) {
+      throw new BridgeClientNotReadyError()
+    }
+  }
+
+  // Answers after `close` as well: what it holds is then the last snapshot, marked `disconnected`.
+  function snapshot(): BridgeConnectionSnapshot {
+    const held = cache.read()
+    if (held === null) {
+      throw new BridgeClientNotReadyError()
+    }
+    return held
+  }
+
+  /** A second `init` is ordinary: the shell answers every `ready`, and a page that re-asked hears
+   *  its own session again. A different id is not, and nothing the page held survives it. */
+  function acceptInit(message: Extract<BridgeHostMessage, { type: 'init' }>): void {
+    handshake.stop()
+    if (session !== null && session.sessionId !== message.sessionId) {
+      const replaced = new BridgeShellReplacedError()
+      requests.closeAll(replaced)
+      subscriptions.failAll(replaced.message)
+    }
+    session = { sessionId: message.sessionId, buildId: message.buildId, grants: message.grants }
+    cache.prime(message.connection)
+    for (const listener of readyListeners) {
+      listener()
+    }
+    readyListeners.clear()
+  }
+
+  /** A shell rebuilt under the page: what the cache holds is for a client that is already gone. */
+  function acceptState(snapshotFromShell: BridgeConnectionSnapshot): void {
+    if (cache.apply(snapshotFromShell) !== 'stale') {
+      return
+    }
+    report({ kind: 'state-out-of-order' })
+    handshake.restart()
+  }
+
+  /** The shell's own words where it had any, the way the native client passes an RPC error message
+   *  through to the listener it ends. */
+  function describeStreamFailure(error: unknown): string {
+    return error instanceof Error ? error.message : 'the shell could not keep this stream open'
+  }
+
+  /** The shell answers a refused `subscribe` with `error` on the stream's id. Nothing is pending to
+   *  reject there, so routing it to the requests would drop it and hold the page's slot forever. */
+  function failExchange(id: string, error: unknown): void {
+    if (subscriptions.has(id)) {
+      // Reported before the listener runs, so a listener that throws cannot swallow the diagnostic.
+      report({ kind: 'stream-failed', error })
+      subscriptions.end(id, describeStreamFailure(error), error)
+      return
+    }
+    if (!requests.has(id)) {
+      report({ kind: 'unknown-id' })
+    }
+    // Still routed: an id with a half-assembled reply behind it holds a slot until it is discarded.
+    requests.fail(id, error)
+  }
+
+  function dispatch(message: BridgeHostMessage, json: string): void {
+    switch (message.type) {
+      case 'init':
+        acceptInit(message)
+        return
+      case 'state':
+        acceptState(message.connection)
+        return
+      case 'reply':
+        if (!requests.has(message.id)) {
+          report({ kind: 'unknown-id' })
+        }
+        requests.acceptReply(message)
+        return
+      case 'error':
+        failExchange(message.id, reconstructBridgeError(message.error))
+        return
+      case 'event':
+        subscriptions.deliver(message, utf8ByteLength(json))
+        return
+      case 'end':
+        report({ kind: 'stream-ended', reason: message.reason })
+        subscriptions.end(message.id, `the shell ended this stream (${message.reason})`)
+        return
+    }
+  }
+
+  function receive(json: string): void {
+    if (closed) {
+      return
+    }
+    const read = readBridgeHostMessage(json)
+    if (!read.ok) {
+      report({ kind: 'refused', refusal: read.refusal })
+      return
+    }
+    dispatch(read.message, json)
+  }
+
+  function sendRequest(...args: [string, unknown?, SendRequestOptions?]): Promise<RpcResponse> {
+    // A call with no session is a page bug and throws; a call over the in-flight cap is the answer
+    // the shell would have posted back, so it arrives the way the shell's does, as a rejection.
+    requireSession()
+    if (closed) {
+      // Rejected, not thrown: `bindDeferredRpcOperation` hands this promise straight back, so a
+      // synchronous throw would escape past the caller's `catch` on the promise.
+      return Promise.reject(new BridgeClientClosedError())
+    }
+    if (requests.size >= BRIDGE_MAX_PENDING_REQUESTS) {
+      return Promise.reject(
+        new BridgeClientCapExceededError(`over ${BRIDGE_MAX_PENDING_REQUESTS} requests in flight`)
+      )
+    }
+    const [method, params, requestOptions] = args
+    const id = nextId()
+    return new Promise<RpcResponse>((resolve, reject) => {
+      requests.open(id, { resolve, reject })
+      const sent = sendFrame({
+        v: BRIDGE_PROTOCOL_VERSION,
+        type: 'request',
+        id,
+        method,
+        // Absent stays absent, because the shell replays whichever arity crossed. JSON drops an
+        // `undefined` value on its own, so an explicit `sendRequest(m, undefined)` reaches the shell
+        // as `sendRequest(m)`; no call site passes one, and no wire that carries `undefined` exists
+        // to carry it. The spread is what states the intent for a carrier that would.
+        ...(args.length > 1 ? { params } : {}),
+        ...(requestOptions === undefined ? {} : { options: requestOptions })
+      })
+      if (!sent) {
+        requests.abandon(id)
+        reject(new BridgeSendFailedError())
+      }
+    })
+  }
+
+  function subscribe(
+    method: string,
+    params: unknown,
+    onData: (result: unknown) => void,
+    subscribeOptions?: { onBinaryFrame?: (frame: BrowserScreencastFrame) => void }
+  ): () => void {
+    requireSession()
+    if (closed) {
+      return () => undefined
+    }
+    // Thrown rather than reported: `subscribe` hands back an unsubscribe and nothing else, so a
+    // refusal the caller could read does not exist on this member. A refusal the shell posts back
+    // arrives too late to throw at all, and reaches the page as a `stream-failed` diagnostic.
+    if (subscriptions.size >= BRIDGE_MAX_SUBSCRIPTIONS) {
+      throw new BridgeClientCapExceededError(`over ${BRIDGE_MAX_SUBSCRIPTIONS} subscriptions`)
+    }
+    const id = nextId()
+    // A frame that never left already told the listener and gave the slot back; the caller still
+    // gets a dispose, because it has no way to know which of the two it is holding.
+    if (!subscriptions.open(id, method, params, onData, subscribeOptions?.onBinaryFrame)) {
+      return () => undefined
+    }
+    let disposed = false
+    return () => {
+      if (disposed) {
+        return
+      }
+      disposed = true
+      subscriptions.cancel(id)
+    }
+  }
+
+  function close(): void {
+    if (closed) {
+      return
+    }
+    closed = true
+    handshake.stop()
+    subscriptions.closeAll()
+    sendFrame({ v: BRIDGE_PROTOCOL_VERSION, type: 'close' })
+    requests.closeAll()
+    cache.close()
+    session = null
+    readyListeners.clear()
+    unsubscribeFromMessages()
+  }
+
+  const unsubscribeFromMessages = options.onMessage(receive)
+  handshake.start()
+
+  return {
+    sendRequest,
+    subscribe,
+    updateTerminalSubscriptionViewport: (terminal, viewport) => {
+      requireSession()
+      if (closed) {
+        return
+      }
+      sendFrame({
+        v: BRIDGE_PROTOCOL_VERSION,
+        type: 'notify',
+        name: 'terminalViewport',
+        terminal,
+        cols: viewport.cols,
+        rows: viewport.rows
+      })
+    },
+    getState: (): ConnectionState => snapshot().state,
+    getReconnectAttempt: () => snapshot().reconnectAttempt,
+    getLastConnectedAt: () => snapshot().lastConnectedAt,
+    getLastInboundAt: () => snapshot().lastInboundAt,
+    // A shell client with no generation of its own never migrates, so its epoch is a constant and
+    // zero is as true as any other. The page still answers a number, because the member it stands in
+    // for is one the native screens read without asking whether it exists.
+    getGeneration: () => snapshot().generation ?? 0,
+    // Not gated on the session: it registers a listener and reads nothing, so it cannot answer
+    // wrongly, and a provider that subscribes before `init` is how a screen hears the first change.
+    onStateChange: (listener) => cache.onStateChange(listener),
+    notifyForeground: (reason?: ForegroundNudgeReason) => {
+      requireSession()
+      if (closed) {
+        return
+      }
+      sendFrame({
+        v: BRIDGE_PROTOCOL_VERSION,
+        type: 'notify',
+        name: 'foreground',
+        ...(reason === undefined ? {} : { reason })
+      })
+    },
+    close,
+    onReady: (listener) => {
+      if (session !== null) {
+        listener()
+        return () => undefined
+      }
+      readyListeners.add(listener)
+      return () => {
+        readyListeners.delete(listener)
+      }
+    },
+    getShellSession: () => session
+  }
+}

mobile/src/mobile-web-shell/bridge/bridge-screencast-binary.ts

@@ -0,0 +1,53 @@
+import {
+  BrowserScreencastOpcode,
+  type BrowserScreencastFrame
+} from '../../transport/browser-screencast-protocol'
+import type { BridgeHostMessage } from './bridge-envelope'
+
+/**
+ * The binary lane's page-side half: base64 in, the same `BrowserScreencastFrame` a native listener
+ * is handed out.
+ *
+ * There is no wire header to parse here. `decodeBrowserScreencastFrame` reads one because the
+ * socket carries a frame as a single buffer; the envelope already carries `format`, `frameSeq` and
+ * the metadata as JSON beside the image, so only the image is base64. C6 owns the encoder that
+ * produces this shape, and this is the inverse it has to satisfy.
+ */
+export type BridgeBinaryEvent = Extract<
+  Extract<BridgeHostMessage, { type: 'event' }>,
+  { binary: unknown }
+>['binary']
+
+/** `null` when the image is not base64: an undecodable frame is dropped, never guessed at. */
+export function decodeBridgeScreencastFrame(
+  event: BridgeBinaryEvent
+): BrowserScreencastFrame | null {
+  const image = decodeBase64(event.b64)
+  if (image === null) {
+    return null
+  }
+  return {
+    opcode: BrowserScreencastOpcode.Frame,
+    // The screencast's own counter. The event frame's `seq` is the bridge's backpressure ordinal,
+    // and handing that one over would renumber every frame the page reports.
+    seq: event.frameSeq,
+    format: event.format,
+    metadata: event.metadata,
+    image
+  }
+}
+
+/** Metro ships no `Buffer`; `atob` is what the pairing and E2EE paths already decode with. */
+function decodeBase64(value: string): Uint8Array | null {
+  let binary: string
+  try {
+    binary = atob(value)
+  } catch {
+    return null
+  }
+  const bytes = new Uint8Array(binary.length)
+  for (let index = 0; index < binary.length; index += 1) {
+    bytes[index] = binary.charCodeAt(index)
+  }
+  return bytes
+}

mobile/src/mobile-web-shell/bridge/orca-bridge-page-channel.test.ts

@@ -0,0 +1,53 @@
+import { afterEach, describe, expect, it, vi } from 'vitest'
+import {
+  createOrcaBridgePageTransport,
+  readOrcaBridgePageChannel,
+  type OrcaBridgePageChannel
+} from './orca-bridge-page-channel'
+
+function installChannel(channel: unknown): void {
+  Object.defineProperty(globalThis, 'orcaBridge', { value: channel, configurable: true })
+}
+
+function createChannel(): OrcaBridgePageChannel {
+  return { postMessage: vi.fn(), onmessage: null }
+}
+
+afterEach(() => {
+  Reflect.deleteProperty(globalThis, 'orcaBridge')
+})
+
+describe('the page channel the shell installs', () => {
+  it('is absent in a browser, which is a page the bundle still has to open', () => {
+    expect(readOrcaBridgePageChannel()).toBeNull()
+  })
+
+  it('refuses a global of another shape rather than posting into it', () => {
+    installChannel({ postMessage: 'not a function', onmessage: null })
+    expect(readOrcaBridgePageChannel()).toBeNull()
+  })
+
+  it('reads the installed object itself, so the page posts through the real sink', () => {
+    const channel = createChannel()
+    installChannel(channel)
+    expect(readOrcaBridgePageChannel()).toBe(channel)
+  })
+})
+
+describe('the page channel as a client transport', () => {
+  it('posts what the client sends', () => {
+    const channel = createChannel()
+    createOrcaBridgePageTransport(channel).send('{"v":1}')
+    expect(channel.postMessage).toHaveBeenCalledWith('{"v":1}')
+  })
+
+  it('hands the client the frame off the event, and gives the slot back', () => {
+    const channel = createChannel()
+    const handler = vi.fn()
+    const release = createOrcaBridgePageTransport(channel).onMessage(handler)
+    channel.onmessage?.({ data: '{"v":1,"type":"init"}' })
+    expect(handler).toHaveBeenCalledWith('{"v":1,"type":"init"}')
+    release()
+    expect(channel.onmessage).toBeNull()
+  })
+})

mobile/src/mobile-web-shell/bridge/orca-bridge-page-channel.ts

@@ -0,0 +1,51 @@
+import type { BridgeRpcClientOptions } from './bridge-rpc-client'
+
+/**
+ * The page's half of the native channel, as the document-start installer leaves it.
+ *
+ * `postMessage` and an `onmessage` assignment are the whole surface, and it is deliberately the
+ * intersection of the two platforms: Android's `addWebMessageListener` injects an object of this
+ * shape, and `MobileWebShellView.swift` installs one to match. Nothing else about the WebView is
+ * addressable from the page.
+ */
+export type OrcaBridgePageChannel = {
+  postMessage: (json: string) => void
+  onmessage: ((event: { data: string }) => void) | null
+}
+
+/**
+ * `null` for a page that is not inside the shell — a browser, or a WebView mounted with the bridge
+ * off. That is a supported way to open the bundle, so the caller substitutes rather than throws.
+ */
+export function readOrcaBridgePageChannel(): OrcaBridgePageChannel | null {
+  const scope: typeof globalThis & { orcaBridge?: OrcaBridgePageChannel } = globalThis
+  const channel = scope.orcaBridge
+  if (channel === undefined || typeof channel.postMessage !== 'function') {
+    return null
+  }
+  return channel
+}
+
+/**
+ * The channel as the page client's transport.
+ *
+ * One `onmessage` slot exists, so one client reads the channel; a second would silently take the
+ * first one's frames. The page holds exactly one client, which is what makes that safe.
+ */
+export function createOrcaBridgePageTransport(
+  channel: OrcaBridgePageChannel
+): Pick<BridgeRpcClientOptions, 'send' | 'onMessage'> {
+  return {
+    send: (json) => {
+      channel.postMessage(json)
+    },
+    onMessage: (handler) => {
+      channel.onmessage = (event) => {
+        handler(event.data)
+      }
+      return () => {
+        channel.onmessage = null
+      }
+    }
+  }
+}

mobile/src/transport/client-context.web.test.tsx

@@ -0,0 +1,161 @@
+import { createElement, type ReactElement } from 'react'
+import { act, create } from 'react-test-renderer'
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+import { BRIDGE_PROTOCOL_VERSION } from '../mobile-web-shell/bridge/bridge-envelope'
+import type { RpcClientContextValue } from './rpc-client-context-contract'
+
+// The web file re-exports the screen hooks, and reaching the real ones imports the Expo runtime
+// this test does not have. Nothing below calls one.
+vi.mock('./host-client-hooks', () => ({
+  useDisconnectHostClient: () => () => {},
+  useForceReconnect: () => () => Promise.resolve(),
+  useForgetHostClient: () => () => {},
+  useHostClient: () => ({ client: null, clientId: null, state: 'disconnected' }),
+  usePrimeHosts: () => () => {},
+  useRefreshHostClient: () => () => {}
+}))
+
+import { RpcClientProvider, useRpcClientContext } from './client-context.web'
+
+const INIT = {
+  v: BRIDGE_PROTOCOL_VERSION,
+  type: 'init',
+  sessionId: 'session-a',
+  buildId: 'build-a',
+  connection: {
+    state: 'connected',
+    reconnectAttempt: 2,
+    lastConnectedAt: 1700,
+    lastInboundAt: 1800,
+    generation: 5
+  },
+  grants: { rpc: { maxPendingRequests: 64, maxSubscriptions: 32 }, native: [] }
+}
+
+/** What the page mounted, and what it holds — the two things the provider decides. */
+const screen: { mounts: number; context: RpcClientContextValue | null } = {
+  mounts: 0,
+  context: null
+}
+
+function Screen(): null {
+  screen.context = useRpcClientContext()
+  screen.mounts += 1
+  return null
+}
+
+function render(): ReactElement {
+  return createElement(RpcClientProvider, null, createElement(Screen))
+}
+
+/** The channel the shell's document-start script installs, as a double. */
+function installChannel(): { posted: string[]; deliver: (frame: unknown) => void } {
+  const posted: string[] = []
+  const channel: {
+    postMessage: (json: string) => void
+    onmessage: ((e: { data: string }) => void) | null
+  } = {
+    postMessage: (json) => {
+      posted.push(json)
+    },
+    onmessage: null
+  }
+  Object.defineProperty(globalThis, 'orcaBridge', { value: channel, configurable: true })
+  return {
+    posted,
+    deliver: (frame) => {
+      channel.onmessage?.({ data: JSON.stringify(frame) })
+    }
+  }
+}
+
+function readContext(): RpcClientContextValue {
+  const context = screen.context
+  if (context === null) {
+    throw new Error('no screen mounted')
+  }
+  return context
+}
+
+beforeEach(() => {
+  vi.useFakeTimers()
+  screen.mounts = 0
+  screen.context = null
+})
+
+afterEach(() => {
+  vi.useRealTimers()
+  Reflect.deleteProperty(globalThis, 'orcaBridge')
+})
+
+describe('the page provider inside the shell', () => {
+  it('mounts nothing until the shell answers with a session', () => {
+    const channel = installChannel()
+    act(() => {
+      create(render())
+    })
+    expect(screen.mounts).toBe(0)
+    expect(channel.posted.map((json: string) => JSON.parse(json).type)).toEqual(['ready'])
+    act(() => {
+      channel.deliver(INIT)
+    })
+    expect(screen.mounts).toBe(1)
+  })
+
+  it('answers every screen with the one client the page has', () => {
+    const channel = installChannel()
+    act(() => {
+      create(render())
+    })
+    act(() => {
+      channel.deliver(INIT)
+    })
+    const context = readContext()
+    const client = context.acquire('host-a', {})
+    expect(client).not.toBeNull()
+    expect(context.getState('host-a')).toBe('connected')
+    expect(context.getReconnectAttempt('host-a')).toBe(2)
+    expect(context.getLastConnectedAt('host-a')).toBe(1700)
+    expect(context.getAllClients()).toEqual([{ hostId: 'host-a', client }])
+  })
+
+  it('carries a state change from the shell to the screens watching it', () => {
+    const channel = installChannel()
+    act(() => {
+      create(render())
+    })
+    act(() => {
+      channel.deliver(INIT)
+    })
+    const listener = vi.fn()
+    readContext().subscribeHostState('host-a', listener)
+    act(() => {
+      channel.deliver({
+        v: BRIDGE_PROTOCOL_VERSION,
+        type: 'state',
+        connection: { ...INIT.connection, state: 'reconnecting' }
+      })
+    })
+    expect(listener).toHaveBeenCalledWith('reconnecting')
+    expect(readContext().getState('host-a')).toBe('reconnecting')
+  })
+})
+
+describe('the page provider outside the shell', () => {
+  it('mounts the route tree at once, because no session is ever coming', () => {
+    act(() => {
+      create(render())
+    })
+    expect(screen.mounts).toBe(1)
+    expect(readContext().getState('host-a')).toBe('disconnected')
+  })
+
+  it('hands out a client that reaches nothing rather than none at all', async () => {
+    act(() => {
+      create(render())
+    })
+    const client = readContext().acquire('host-a', {})
+    expect(client).not.toBeNull()
+    await expect(client?.sendRequest('worktree.ps')).rejects.toThrow('bridge transport unavailable')
+  })
+})

mobile/src/transport/client-context.web.tsx

@@ -1,6 +1,24 @@
-// Web sibling: RN Web has no pairing keychain and no websocket transport of its own, so the page
+// Web sibling: RN Web has no pairing keychain and no websocket transport of its own, so the page's
-// gets a placeholder client until C0.4 lands BridgeRpcClient over the shell bridge.
+// client is the shell bridge. Nothing here dials, retries or pairs — the native client on the other
-import { createContext, useContext, useMemo, type ReactNode } from 'react'
+// side of the bridge already did, and this provider only carries what it holds across the boundary.
+import {
+  createContext,
+  useContext,
+  useEffect,
+  useMemo,
+  useRef,
+  useState,
+  type ReactNode
+} from 'react'
+import {
+  createBridgeRpcClient,
+  type BridgeRpcClient,
+  type BridgeRpcClientDiagnostic
+} from '../mobile-web-shell/bridge/bridge-rpc-client'
+import {
+  createOrcaBridgePageTransport,
+  readOrcaBridgePageChannel
+} from '../mobile-web-shell/bridge/orca-bridge-page-channel'
 import type { RpcClient } from './rpc-client'
 import type { ConnectionState, HostProfile } from './types'
 import type { RpcClientContextValue } from './rpc-client-context-contract'
@@ -22,6 +40,12 @@ export class BridgeTransportUnavailableError extends Error {
   }
 }
 
+/**
+ * For a page opened outside the shell: a browser, or a WebView mounted with the bridge off.
+ *
+ * It answers every member and reaches nothing, which is what lets the route tree mount and paint
+ * its empty states instead of crashing on a client that is not there.
+ */
 function createPlaceholderClient(): RpcClient {
   return {
     sendRequest: (method) => Promise.reject(new BridgeTransportUnavailableError(method)),
@@ -40,14 +64,64 @@ function createPlaceholderClient(): RpcClient {
   }
 }
 
+/** One line per kind for the life of one page: a page that is failing frames fails all of them. */
+function createPageDiagnosticReporter(): (diagnostic: BridgeRpcClientDiagnostic) => void {
+  const reported = new Set<BridgeRpcClientDiagnostic['kind']>()
+  return (diagnostic) => {
+    if (reported.has(diagnostic.kind)) {
+      return
+    }
+    reported.add(diagnostic.kind)
+    console.warn('[page-bridge]', diagnostic.kind, diagnostic)
+  }
+}
+
 const Ctx = createContext<RpcClientContextValue | null>(null)
 
 export function RpcClientProvider({ children }: { children: ReactNode }) {
+  // Held in a ref as well as in state: the context value is built once, because `useHostClient`
+  // re-acquires whenever the value's identity changes.
+  const clientRef = useRef<RpcClient | null>(null)
+  const acquiredRef = useRef<Set<string>>(new Set())
+  const [ready, setReady] = useState(false)
+
+  useEffect(() => {
+    const channel = readOrcaBridgePageChannel()
+    if (channel === null) {
+      // Nothing to wait for, so the tree mounts against the placeholder rather than never.
+      clientRef.current = createPlaceholderClient()
+      setReady(true)
+      return
+    }
+    const client: BridgeRpcClient = createBridgeRpcClient({
+      ...createOrcaBridgePageTransport(channel),
+      onDiagnostic: createPageDiagnosticReporter()
+    })
+    // Nothing mounts before `init`: every member of this client throws until the shell answers,
+    // and a screen that rendered first would record its first frame against a session-less client.
+    const release = client.onReady(() => {
+      clientRef.current = client
+      setReady(true)
+    })
+    return () => {
+      release()
+      clientRef.current = null
+      setReady(false)
+      client.close()
+    }
+  }, [])
+
   const value = useMemo<RpcClientContextValue>(() => {
-    const client = createPlaceholderClient()
+    const state = (): ConnectionState => clientRef.current?.getState() ?? 'connecting'
-    const disconnected: ConnectionState = 'disconnected'
     return {
-      acquire: () => client,
+      // One client for one page: the shell opened this document for one host, so whichever host
+      // the route names is the host on the other side of the bridge.
+      acquire: (hostId: string) => {
+        acquiredRef.current.add(hostId)
+        return clientRef.current
+      },
+      // The shell owns the connection, and a page client cannot be reopened once it says goodbye.
+      // Every member that would close, drop or re-dial one is inert here for that reason.
       release: () => {},
       releaseAndCloseIfUnused: () => {},
       closeIfUnused: () => {},
@@ -55,24 +129,33 @@ export function RpcClientProvider({ children }: { children: ReactNode }) {
       refreshHostClient: () => {},
       forgetHostClient: () => {},
       disconnectHostClient: () => {},
-      getState: () => disconnected,
+      getState: state,
-      getKnownState: () => disconnected,
+      getKnownState: () => (clientRef.current === null ? null : state()),
       getClientId: () => null,
-      getReconnectAttempt: () => 0,
+      getReconnectAttempt: () => clientRef.current?.getReconnectAttempt() ?? 0,
-      getLastConnectedAt: () => null,
+      getLastConnectedAt: () => clientRef.current?.getLastConnectedAt() ?? null,
       // The page reaches its host through the shell bridge, which rides whatever path the RN
       // client already negotiated. 'relay' is the honest default until init carries the real one.
       getActivePath: () => 'relay',
       getPendingPath: () => null,
+      // Both are pairing verdicts, and pairing happened natively before this document existed.
       isPairingRejected: () => false,
       isHostSignedOut: () => false,
-      subscribeHostState: () => () => {},
+      subscribeHostState: (_hostId: string, listener: (next: ConnectionState) => void) =>
-      getAllClients: () => [],
+        clientRef.current?.onStateChange(listener) ?? (() => {}),
-      subscribeAllHosts: () => () => {},
+      getAllClients: () => {
+        const client = clientRef.current
+        return client === null ? [] : [...acquiredRef.current].map((hostId) => ({ hostId, client }))
+      },
+      subscribeAllHosts: (listener: () => void) =>
+        clientRef.current?.onStateChange(() => {
+          listener()
+        }) ?? (() => {}),
       primeHosts: (_hosts: HostProfile[]) => {}
     }
   }, [])
-  return <Ctx.Provider value={value}>{children}</Ctx.Provider>
+
+  return <Ctx.Provider value={value}>{ready ? children : null}</Ctx.Provider>
 }
 
 export function useRpcClientContext(): RpcClientContextValue {

mobile/src/transport/unvalidated-rpc-request-port-inventory.ts

@@ -25,13 +25,17 @@ export type UnvalidatedRpcRequestPortEntry = {
 
 /** Modules whose job is the port. These do not shrink to zero. */
 export const UNVALIDATED_RPC_REQUEST_PORT_OWNERS: readonly UnvalidatedRpcRequestPortEntry[] = [
-  // Carries the port across the page boundary for the hybrid shell. Not a call site: it picks no
+  // Forwards raw requests as a transport, reads no reply. Not a call site: it picks no method and
-  // method, reads no reply and decides no acceptance — the page names the method and runs the
+  // decides no acceptance — the page names the method and runs the typed operation over it, exactly
-  // typed operation over it, exactly as a native screen does over a socket client.
+  // as a native screen does over a socket client.
   { file: 'src/mobile-web-shell/bridge-host.ts', references: 3 },
+  // The far end of that transport: it offers the port to the page and posts what it is handed,
+  // reading neither the method nor the reply.
+  { file: 'src/mobile-web-shell/bridge/bridge-rpc-client.ts', references: 1 },
   // Fakes the port for the bridge host suites; a non-test file only because tsconfig excludes tests.
   { file: 'src/mobile-web-shell/bridge-host-test-fakes.ts', references: 1 },
-  // Placeholder page transport until C0.4's BridgeRpcClient replaces it; rejects every call, reads no reply.
+  // The page's client is BridgeRpcClient over the shell bridge; this one reference is the
+  // placeholder it falls back to outside the shell, which rejects every call and reads no reply.
   { file: 'src/transport/client-context.web.tsx', references: 1 },
   // Implements the port over the device-to-host websocket.
   { file: 'src/transport/direct-rpc-client.ts', references: 3 },

mobile/web-entry/web-overrides.json

@@ -3,7 +3,7 @@
   "overrides": [
     {
       "file": "src/transport/client-context.web.tsx",
-      "reason": "The page has no websocket transport and no pairing keychain. This is the single transport substitution point: a placeholder RpcClient until C0.4 lands BridgeRpcClient over the shell bridge."
+      "reason": "The page has no websocket transport and no pairing keychain. This is the single transport substitution point: BridgeRpcClient over the shell bridge, and a placeholder RpcClient for a page opened outside it, which is what lets the route tree mount in a plain browser."
     },
     {
       "file": "packages/expo-two-way-audio/src/ExpoTwoWayAudioModule.web.ts",