fix: make Browser RPC and Realtime own the physical work they report on

A server stream's registration was pruned on any settled close receipt.
`waitClosed()` rejecting, throwing synchronously, or not returning a
promise at all was absorbed into a fulfilled `undefined`, so the runtime
opened a second physical stream for the same operation while the first was
still running against the server. Only a fulfilled, contract-shaped
receipt confirms closure now; every negative receipt keeps the operation
DRAINING.

Cleanup also read foreign state outside the result boundary. A throwing
iterator `return` accessor replaced the already selected timeout with a
native `TypeError` and skipped the rest of the teardown, and the exported
lease decoder threw on a hostile `Symbol.asyncIterator`. Both reads move
inside their own boundaries, and the positive-close subscription is
installed before any fallible cleanup.

Composition validated the caller's registries before snapshotting them, so
a hostile accessor ran twice during validation, and rows hiding fields
behind a prototype or a non-enumerable key installed. Transport results
were checked for allowed own keys only, so own `{ok,message,encodedBytes}`
plus a prototype `injected` was a success and a missing `message` reached
a permissive schema as `undefined`.

In Realtime the tracked task was registered after the collaborator
returned. An authority that re-entered `close()` from inside its own
invocation saw an empty registry and got `{ok:true}` while its effect was
pending. The task is now registered first and the collaborator is invoked
a microtask later. A `scheduleTimeout` that threw was worse: the caller's
own catch treated it as an apply failure and started a recovery beside the
still-running effect, and `close()` rejected with a native `TypeError`. An
uninstallable deadline now fails closed as an expired one.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

tests/unit/realtime/stream-coordinator.test.ts

@@ -1068,6 +1068,8 @@ type HarnessOptions = Readonly<{
     recoveryTimeoutMs?: number;
     drainTimeoutMs?: number;
   }>;
+  scheduleTimeout?: (callback: () => void, delayMs: number) => unknown;
+  clearScheduledTimeout?: (handle: unknown) => void;
 }>;
 
 function createHarness(options: HarnessOptions = {}) {
@@ -1114,6 +1116,12 @@ function createHarness(options: HarnessOptions = {}) {
     now: () => 10_000,
     observe: options.observe,
     ...(options.taskLimits ? { taskLimits: options.taskLimits } : {}),
+    ...(options.scheduleTimeout
+      ? { scheduleTimeout: options.scheduleTimeout }
+      : {}),
+    ...(options.clearScheduledTimeout
+      ? { clearScheduledTimeout: options.clearScheduledTimeout }
+      : {}),
   });
 
   return {
@@ -1176,3 +1184,168 @@ function deferred<Value>() {
   });
   return { promise, resolve };
 }
+
+/**
+ * RT-01. The registry entry has to exist before the collaborator is called.
+ * Registering after the invocation returned left a window in which an authority
+ * that re-entered `close()` from inside its own callback saw an empty set, so
+ * `close()` reported quiescence while its effect was still running.
+ */
+describe("RT-01 reentrant close cannot pass the registration window", () => {
+  it("refuses quiescence when apply re-enters close during its invocation", async () => {
+    const applyGate = deferred<RealtimeResult<void>>();
+    let closeResult: RealtimeResult<void> | undefined;
+    let harness: ReturnType<typeof createHarness> | undefined;
+    harness = createHarness({
+      apply: () => {
+        // Re-entered from inside the invocation itself.
+        void harness!.coordinator.close().then((result) => {
+          closeResult = result;
+        });
+        return applyGate.promise;
+      },
+      taskLimits: { effectTimeoutMs: 10_000, drainTimeoutMs: 20 },
+    });
+    await harness.initialize();
+
+    const applying = harness.coordinator.accept(harness.event());
+    await new Promise((resolve) => setTimeout(resolve, 50));
+
+    expect(closeResult?.ok).toBe(false);
+    expect(closeResult?.ok ? null : closeResult?.error.kind).toBe(
+      "IDLE_TIMEOUT",
+    );
+
+    applyGate.resolve(realtimeSuccess(undefined));
+    await applying;
+    // Once the raw task settled, a second close does converge.
+    await expect(harness.coordinator.close()).resolves.toMatchObject({
+      ok: true,
+    });
+  });
+
+  it("refuses quiescence when recovery re-enters close during its invocation", async () => {
+    const recoveryGate = deferred<RealtimeResult<RealtimeRecoveryCommit>>();
+    let closeResult: RealtimeResult<void> | undefined;
+    let harness: ReturnType<typeof createHarness> | undefined;
+    let recoveries = 0;
+    harness = createHarness({
+      recover: () => {
+        recoveries += 1;
+        if (recoveries === 1) {
+          return Promise.resolve(realtimeSuccess(snapshotCommit("0")));
+        }
+        void harness!.coordinator.close().then((result) => {
+          closeResult = result;
+        });
+        return recoveryGate.promise;
+      },
+      taskLimits: { recoveryTimeoutMs: 10_000, drainTimeoutMs: 20 },
+    });
+    await harness.initialize();
+
+    const recovering = harness.coordinator.recover(STREAM_ID, "SEQUENCE_GAP");
+    await new Promise((resolve) => setTimeout(resolve, 50));
+
+    expect(closeResult?.ok).toBe(false);
+    expect(closeResult?.ok ? null : closeResult?.error.kind).toBe(
+      "IDLE_TIMEOUT",
+    );
+
+    recoveryGate.resolve(realtimeSuccess(snapshotCommit("0")));
+    await recovering;
+  });
+});
+
+/**
+ * RT-02. A scheduler that cannot install a deadline leaves the wait unbounded.
+ * Letting the exception escape turned a typed realtime result into a native
+ * rejection and, through the caller's own catch, started a recovery that
+ * overlapped the effect still running.
+ */
+describe("RT-02 a throwing scheduler stays inside the result contract", () => {
+  /** Installs deadlines normally until the stream is initialized. */
+  function breakableScheduler() {
+    const state = { broken: false };
+    return {
+      state,
+      scheduleTimeout: (callback: () => void, delayMs: number) => {
+        if (state.broken) throw new TypeError("scheduleTimeout exploded");
+        return setTimeout(callback, delayMs);
+      },
+      clearScheduledTimeout: (handle: unknown) => {
+        clearTimeout(handle as ReturnType<typeof setTimeout>);
+      },
+    };
+  }
+
+  it("does not start a recovery that overlaps a pending apply", async () => {
+    const applyGate = deferred<RealtimeResult<void>>();
+    const scheduler = breakableScheduler();
+    let recoveriesAfterApply = 0;
+    const harness = createHarness({
+      apply: () => applyGate.promise,
+      recover: async () => {
+        recoveriesAfterApply += 1;
+        return realtimeSuccess(snapshotCommit("0"));
+      },
+      taskLimits: { effectTimeoutMs: 10_000, drainTimeoutMs: 20 },
+      scheduleTimeout: scheduler.scheduleTimeout,
+      clearScheduledTimeout: scheduler.clearScheduledTimeout,
+    });
+    await harness.initialize();
+    recoveriesAfterApply = 0;
+    scheduler.state.broken = true;
+
+    const accepted = await harness.coordinator.accept(harness.event());
+
+    // The apply is still pending, so no recovery may have run beside it.
+    expect(recoveriesAfterApply).toBe(0);
+    expect(accepted.ok).toBe(false);
+    expect(harness.coordinator.lifecycle(STREAM_ID)).toBe("DRAINING");
+
+    applyGate.resolve(realtimeSuccess(undefined));
+  });
+
+  it("returns a typed close failure instead of rejecting natively", async () => {
+    const applyGate = deferred<RealtimeResult<void>>();
+    const scheduler = breakableScheduler();
+    const harness = createHarness({
+      apply: () => applyGate.promise,
+      taskLimits: { effectTimeoutMs: 10_000, drainTimeoutMs: 20 },
+      scheduleTimeout: scheduler.scheduleTimeout,
+      clearScheduledTimeout: scheduler.clearScheduledTimeout,
+    });
+    await harness.initialize();
+    scheduler.state.broken = true;
+
+    const applying = harness.coordinator.accept(harness.event());
+    // Let the apply reach the authority before the fence arrives.
+    await Promise.resolve();
+    await Promise.resolve();
+    const closed = await harness.coordinator.close();
+
+    // A drain that could not be bounded is not proof of quiescence.
+    expect(closed.ok).toBe(false);
+    expect(closed.ok ? null : closed.error.kind).toBe("IDLE_TIMEOUT");
+
+    applyGate.resolve(realtimeSuccess(undefined));
+    await applying;
+  });
+
+  it("keeps the classified outcome when clearing a timer throws", async () => {
+    const harness = createHarness({
+      clearScheduledTimeout: () => {
+        throw new TypeError("clearScheduledTimeout exploded");
+      },
+    });
+    await harness.initialize();
+
+    await expect(
+      harness.coordinator.accept(harness.event()),
+    ).resolves.toMatchObject({ ok: true });
+    await expect(harness.coordinator.close()).resolves.toMatchObject({
+      ok: true,
+    });
+  });
+});