feat(client): per-document reactive subscription (doc_subscribe), drop polling

Expose subscribeDoc(nuri, onChange) / subscribeDocs(nuris, onChange) wrapping the
real ng.doc_subscribe — per-document, event-driven (initial State + a Patch per
commit, local or broker-synced from a remote peer), returning a sync unsubscribe.
subscribeDocs isolates per doc (a failing/unsynced doc never aborts the others),
so it sidesteps the ORM fan-out hang (never orm_start_graph(graphs:[…])).

Replace the setInterval polling in inbox.watch() and discovery.watchIndex() with
doc_subscribe — same public contract, now push not poll. NextGraph is subscription-
first; no polling remains.

Verified against the real broker (@data harness spike): the doc_subscribe callback
marshals across the iframe RPC (@ng-org/web strips the callback and drives it via a
MessagePort — no DataCloneError) and fires on the initial push and on a real write.
Lib bun test 91 pass, tsc clean.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

packages/client/src/discovery.ts

@@ -38,6 +38,7 @@
  */
 
 import * as inbox from "./inbox";
+import { subscribeDoc } from "./subscribe";
 import { ensureAccount, reservedAccount } from "./store-registry";
 import { getCaps } from "./polyfill";
 import type { Nuri, PrincipalId } from "./types";
@@ -156,18 +157,25 @@ export async function readIndex(): Promise<IndexEntry[]> {
 }
 
 /**
- * Watch the discovery index. Polls {@link readIndex} and fires `onEntries`
- * whenever the index changes. Returns an unsubscribe. Fires once immediately.
- * (Deduplication is applied on each read.)
+ * Watch the discovery index — **event-driven, not polled**. Subscribes to the
+ * index document via {@link subscribeDoc} (the platform's `doc_subscribe` push):
+ * `onEntries` fires once on the initial state push and again on every subsequent
+ * change to the index document — a local submission OR a broker-synced remote one.
+ * Returns an unsubscribe. (Deduplication is applied on each read.)
+ *
+ * The `intervalMs` option is accepted for signature compatibility but IGNORED:
+ * there is no polling. The index is a single document, so this is immune to the
+ * ORM fan-out hang (see {@link subscribeDoc}).
  */
 export function watchIndex(
   onEntries: (entries: IndexEntry[]) => void,
-  opts?: { intervalMs?: number },
+  _opts?: { intervalMs?: number },
 ): () => void {
   let stopped = false;
   let lastCount = -1;
-  const intervalMs = opts?.intervalMs ?? 1000;
-  const tick = async (): Promise<void> => {
+  let unsubscribe: (() => void) | null = null;
+
+  const refresh = async (): Promise<void> => {
     if (stopped) return;
     try {
       const entries = await readIndex();
@@ -179,10 +187,25 @@ export function watchIndex(
       console.error("[discovery] watchIndex read failed:", error);
     }
   };
-  void tick();
-  const handle = setInterval(() => void tick(), intervalMs);
+
+  // The index document NURI is resolved async (ensureAccount); subscribe once it
+  // is known. The initial State push fires the first read (onEntries fires once),
+  // each later Patch a re-read.
+  void (async () => {
+    try {
+      const anchor = await indexInboxNuri();
+      if (stopped) return;
+      unsubscribe = subscribeDoc(anchor, () => void refresh());
+    } catch (error) {
+      console.error("[discovery] watchIndex subscribe failed:", error);
+    }
+  })();
+
   return () => {
     stopped = true;
-    clearInterval(handle);
+    if (unsubscribe) {
+      unsubscribe();
+      unsubscribe = null;
+    }
   };
 }

packages/client/src/inbox.ts

@@ -25,6 +25,7 @@
  */
 
 import { sparqlUpdate, sparqlQuery } from "./docs";
+import { subscribeDoc } from "./subscribe";
 import { getCurrentUser, getStoreRegistryDeps } from "./polyfill";
 import { escapeLiteral } from "./sparql";
 import type { Nuri, PrincipalId } from "./types";
@@ -186,22 +187,31 @@ export async function read(targetInbox: Nuri): Promise<Deposit[]> {
 export const materialize = read;
 
 /**
- * Subscription over an inbox. Polls {@link read} on an interval and invokes
- * `onDeposits` with the full current deposit list whenever it changes (grows).
- * Returns an unsubscribe function. The polyfill has no reactive inbox
- * subscription, so this polls; against real NextGraph it follows the recipient's
- * own inbox processing. `onDeposits` fires once immediately.
+ * Subscription over an inbox — **event-driven, not polled**. Subscribes to the
+ * inbox document via {@link subscribeDoc} (the platform's `doc_subscribe` push):
+ * `onDeposits` fires once on the initial state push and again on every subsequent
+ * change to the inbox document — a local deposit OR a broker-synced remote one.
+ * Returns an unsubscribe function.
+ *
+ * On each push it re-reads the full deposit list ({@link read}) and invokes
+ * `onDeposits` only when the deposit count changed (grew), keeping the same
+ * "fires on change" contract the polling watcher had — same callback signature
+ * and same behaviour, just event-driven instead of `setInterval`.
+ *
+ * The `intervalMs` option is accepted for signature compatibility but IGNORED:
+ * there is no polling. (The inbox document is a single doc, so this is immune to
+ * the ORM fan-out hang — see {@link subscribeDoc}.)
  */
 export function watch(
   targetInbox: Nuri,
   onDeposits: (deposits: Deposit[]) => void,
-  opts?: { intervalMs?: number },
+  _opts?: { intervalMs?: number },
 ): () => void {
   let stopped = false;
   let lastCount = -1;
-  const intervalMs = opts?.intervalMs ?? 1000;
 
-  const tick = async (): Promise<void> => {
+  // Re-read on every push; fire onDeposits only when the set changed (grew).
+  const refresh = async (): Promise<void> => {
     if (stopped) return;
     try {
       const deposits = await read(targetInbox);
@@ -214,10 +224,11 @@ export function watch(
     }
   };
 
-  void tick(); // fire once immediately
-  const handle = setInterval(() => void tick(), intervalMs);
+  // Subscribe to the inbox document: the initial State push fires the first read
+  // (so onDeposits fires once immediately, as before), each later Patch a re-read.
+  const unsubscribe = subscribeDoc(targetInbox, () => void refresh());
   return () => {
     stopped = true;
-    clearInterval(handle);
+    unsubscribe();
   };
 }

packages/client/src/index.ts

@@ -18,6 +18,8 @@ export * as inbox from "./inbox";
 export * as discovery from "./discovery";
 export type { IndexEntry, SubmitOptions } from "./discovery";
 export * as docs from "./docs";
+export { subscribeDoc, subscribeDocs } from "./subscribe";
+export type { DocChange, Unsubscribe } from "./subscribe";
 export * as readModel from "./read-model";
 export type { UnionSubject } from "./read-model";
 export * as storeRegistry from "./store-registry";

packages/client/src/subscribe.ts

@@ -0,0 +1,149 @@
+/**
+ * Reactive single-document subscription — the polyfill's typed wrapper over the
+ * platform's `doc_subscribe` primitive. This is the canonical NextGraph reactive
+ * read at the document granularity: subscribe once, get the initial state pushed,
+ * then a push on every subsequent commit to that document — whether the write was
+ * local (this session) or a broker-synced remote change. NO POLLING.
+ *
+ * ── Why call the REAL injected `ng` directly (never `makeNg`) ──────────────
+ * Same hard constraint as `docs.ts`: the public `ng` is a JS `Proxy` over
+ * `@ng-org/web`'s iframe-RPC proxy. `doc_subscribe` is a STREAMED method — the
+ * `@ng-org/web` RPC strips the callback (by positional index) BEFORE it posts to
+ * the iframe and drives it locally via a `MessageChannel` port (the function is
+ * never structured-cloned, so no `DataCloneError`). Layering our own Proxy on top
+ * risks re-wrapping that surface; reaching the real `ng` held in the config avoids
+ * the double-proxy exactly as the raw `docs` primitives do. Do not import from
+ * `./ng-proxy`.
+ *
+ * ── The primitive shape (verified against nextgraph-rs) ────────────────────
+ * `ng.doc_subscribe(repo_o: string, session_id, callback)`
+ *   (`sdk/js/lib-wasm/src/lib.rs:1907`) is **per-document** — one repo NURI, one
+ *   callback. It is `async`, resolving to a JS **unsubscribe function**. The
+ *   callback is invoked `callback(appResponse)` with a serialized `AppResponse`:
+ *   `{ V0: { State | Patch | TabInfo | ... } }`. It pushes an initial `State`
+ *   (plus a `TabInfo`) on subscribe, then a `Patch` per verified commit on the
+ *   branch. Returning `true` from the callback also cancels; we cancel by calling
+ *   the returned unsubscribe fn.
+ *
+ * ── Why per-document, never `orm_start_graph(graphs:[…])` ──────────────────
+ * A single not-yet-synced repo in an ORM graph fan-out makes `RepoNotFound` abort
+ * the WHOLE subscription (`initialize.rs:125-128`), so the readyPromise never
+ * resolves → the ~75s hang. `doc_subscribe` is per-branch/per-doc and has no
+ * fan-out: an absent doc breaks only its own subscription. {@link subscribeDocs}
+ * builds a set of these with per-doc error isolation to preserve that property.
+ */
+
+import { getConfig, getStoreRegistryDeps } from "./polyfill";
+import type { Nuri } from "./types";
+
+/**
+ * A push from the platform to a document subscriber. Loosely typed: the raw
+ * serialized `AppResponse` (`{ V0: { State | Patch | TabInfo | ... } }`). The
+ * consumer typically ignores the payload and uses the push purely as a
+ * change SIGNAL (re-query on change — the read-model pattern), so this stays
+ * permissive rather than modelling every AppResponse variant.
+ */
+export type DocChange = unknown;
+
+/** An unsubscribe function — idempotent (calling it twice is a no-op). */
+export type Unsubscribe = () => void;
+
+async function sessionId(): Promise<string> {
+  return (await getStoreRegistryDeps().getSession()).sessionId;
+}
+
+/**
+ * Subscribe to ONE document. `onChange` fires on the initial state push and on
+ * every subsequent change to that doc (local write OR broker-synced remote
+ * change). Returns an unsubscribe function.
+ *
+ * The wrapper is synchronous-returning (an unsubscribe fn) even though the
+ * underlying `ng.doc_subscribe` is async: the real unsubscribe is captured when
+ * the promise resolves; if the caller unsubscribes before setup completes, the
+ * cancellation is honoured as soon as the real unsubscribe is available (and no
+ * further `onChange` fires after unsubscribe).
+ *
+ * Calls the REAL injected `ng.doc_subscribe` directly (never `makeNg`).
+ */
+export function subscribeDoc(nuri: Nuri, onChange: (r: DocChange) => void): Unsubscribe {
+  const { ng } = getConfig();
+  let stopped = false;
+  let realUnsub: (() => void) | null = null;
+
+  const cb = (resp: DocChange): void => {
+    if (stopped) return;
+    try {
+      onChange(resp);
+    } catch (error) {
+      console.error("[subscribe] onChange handler threw for", nuri, error);
+    }
+  };
+
+  // Kick off the async subscription. Errors are isolated to this doc (they never
+  // reject a shared batch — see subscribeDocs). If setup fails, this doc simply
+  // never fires; the caller's unsubscribe stays a safe no-op.
+  void (async () => {
+    try {
+      const sid = await sessionId();
+      const unsub = (await ng.doc_subscribe(nuri, sid, cb)) as (() => void) | undefined;
+      if (stopped) {
+        // Unsubscribed before setup resolved — cancel immediately.
+        if (typeof unsub === "function") unsub();
+        return;
+      }
+      realUnsub = typeof unsub === "function" ? unsub : null;
+    } catch (error) {
+      console.error("[subscribe] doc_subscribe failed for", nuri, error);
+    }
+  })();
+
+  return () => {
+    if (stopped) return;
+    stopped = true;
+    if (realUnsub) {
+      try {
+        realUnsub();
+      } catch (error) {
+        console.error("[subscribe] unsubscribe failed for", nuri, error);
+      }
+      realUnsub = null;
+    }
+  };
+}
+
+/**
+ * Subscribe to a SET of documents, one {@link subscribeDoc} per NURI, with
+ * PER-DOC error isolation. `onChange(nuri, r)` fires for whichever doc changed.
+ * Returns a single unsubscribe that tears down all of them.
+ *
+ * The per-doc isolation is the point: a bad / not-yet-synced doc breaks only its
+ * own subscription and NEVER aborts the others (this is precisely what avoids the
+ * ORM fan-out hang — do NOT replace this with `orm_start_graph(graphs:[…])`). The
+ * set is deduplicated; an empty set returns a no-op unsubscribe.
+ */
+export function subscribeDocs(
+  nuris: Nuri[],
+  onChange: (nuri: Nuri, r: DocChange) => void,
+): Unsubscribe {
+  const unique = [...new Set(nuris.filter(Boolean))];
+  const unsubs = unique.map((nuri) => {
+    // Each subscription is independent: subscribeDoc already isolates its own
+    // async setup failure (logged, never thrown), so one bad doc cannot abort the
+    // construction of the others here.
+    try {
+      return subscribeDoc(nuri, (r) => onChange(nuri, r));
+    } catch (error) {
+      console.error("[subscribe] subscribeDocs: failed to subscribe", nuri, error);
+      return () => {};
+    }
+  });
+  return () => {
+    for (const u of unsubs) {
+      try {
+        u();
+      } catch (error) {
+        console.error("[subscribe] subscribeDocs: unsubscribe failed", error);
+      }
+    }
+  };
+}

packages/client/test/discovery.test.ts

@@ -57,6 +57,25 @@ function makeFakeNg() {
   const quads: Quad[] = [];
   let docCounter = 0;
 
+  // Reactive subscriptions (see inbox.test.ts): doc_subscribe registers a
+  // callback per anchor + fires an initial push; sparql_update pushes a Patch to
+  // that anchor's subscribers, so discovery.watchIndex (now event-driven) works
+  // without a timer.
+  const subs = new Map<string, Set<(r: unknown) => void>>();
+  const doc_subscribe = mock(async (nuri: string, _sid: unknown, cb: (r: unknown) => void) => {
+    let set = subs.get(nuri);
+    if (!set) {
+      set = new Set();
+      subs.set(nuri, set);
+    }
+    set.add(cb);
+    queueMicrotask(() => cb({ V0: { State: { doc: nuri } } }));
+    return () => set!.delete(cb);
+  });
+  const pushTo = (anchor: string): void => {
+    for (const cb of subs.get(anchor) ?? []) cb({ V0: { Patch: { doc: anchor } } });
+  };
+
   const doc_create = mock(async (..._a: unknown[]) => `did:ng:o:doc${++docCounter}`);
 
   const sparql_update = mock(async (...a: unknown[]) => {
@@ -93,6 +112,7 @@ function makeFakeNg() {
       const o = m[2] !== undefined ? unescapeLiteral(m[2]) : (m[3] ?? "");
       quads.push({ g, s, p, o });
     }
+    pushTo(g); // local-push to the written graph's subscribers
     return undefined;
   });
 
@@ -157,7 +177,7 @@ function makeFakeNg() {
     return { results: { bindings: [] } };
   });
 
-  return { doc_create, sparql_update, sparql_query, _quads: quads };
+  return { doc_create, doc_subscribe, sparql_update, sparql_query, _quads: quads };
 }
 
 const SESSION: RegistrySession = { sessionId: "sid-1", privateStoreId: "PRIV" };

packages/client/test/inbox.test.ts

@@ -57,6 +57,25 @@ function unescapeLiteral(s: string): string {
 function makeFakeNg() {
   const quads: Quad[] = [];
 
+  // Reactive subscriptions: doc_subscribe registers a callback per anchor and
+  // fires an initial State push; a matching sparql_update pushes a Patch to that
+  // anchor's subscribers. This mirrors the real broker's local-push behaviour so
+  // inbox.watch (now event-driven, no polling) can be tested without a timer.
+  const subs = new Map<string, Set<(r: unknown) => void>>();
+  const doc_subscribe = mock(async (nuri: string, _sid: unknown, cb: (r: unknown) => void) => {
+    let set = subs.get(nuri);
+    if (!set) {
+      set = new Set();
+      subs.set(nuri, set);
+    }
+    set.add(cb);
+    queueMicrotask(() => cb({ V0: { State: { doc: nuri } } })); // initial push
+    return () => set!.delete(cb);
+  });
+  const pushTo = (anchor: string): void => {
+    for (const cb of subs.get(anchor) ?? []) cb({ V0: { Patch: { doc: anchor } } });
+  };
+
   const doc_create = mock(async (..._a: unknown[]) => "did:ng:o:new");
 
   // Parses one deposit: `<subj> a <Deposit> ; <payload> "..." ; <ts> "..." [; <from> "..."] .`
@@ -89,6 +108,9 @@ function makeFakeNg() {
       const o = rawLit !== undefined ? unescapeLiteral(rawLit) : (m[3] ?? "");
       quads.push({ g, s, p, o });
     }
+    // A write to `g` (the anchored default graph) pushes a Patch to that doc's
+    // subscribers — the local-push the real broker performs on a verified commit.
+    pushTo(g);
     return undefined;
   });
 
@@ -121,7 +143,7 @@ function makeFakeNg() {
     return { results: { bindings } };
   });
 
-  return { doc_create, sparql_update, sparql_query, _quads: quads };
+  return { doc_create, doc_subscribe, sparql_update, sparql_query, _quads: quads };
 }
 
 const SESSION: RegistrySession = { sessionId: "sid-1", privateStoreId: "PRIV" };

packages/client/test/subscribe.test.ts

@@ -0,0 +1,147 @@
+import { test, expect, mock, afterAll } from "bun:test";
+import { subscribeDoc, subscribeDocs } from "../src/subscribe";
+import {
+  configure,
+  configureStoreRegistry,
+  resetConfig,
+  resetStoreRegistry,
+} from "../src/polyfill";
+import type { RegistrySession } from "../src/store-registry";
+
+// subscribeDoc/subscribeDocs wrap the REAL injected `ng.doc_subscribe`. This
+// suite injects a fake `ng` whose `doc_subscribe` records the callback per doc
+// and hands back an unsubscribe, so we can assert routing + isolation without a
+// broker. Restore the un-configured state at the end.
+afterAll(() => {
+  resetConfig();
+  resetStoreRegistry();
+});
+
+const SESSION: RegistrySession = { sessionId: "sid-1", privateStoreId: "PRIV" };
+
+/**
+ * A fake reactive `ng`: `doc_subscribe(nuri, sid, cb)` registers `cb` for `nuri`,
+ * fires it once (initial State push), and returns an unsubscribe. `push(nuri)`
+ * drives a later change to that doc's subscribers. A per-doc `failFor` set makes
+ * `doc_subscribe` reject for chosen NURIs (a not-yet-synced doc).
+ */
+function makeFakeNg(failFor: Set<string> = new Set()) {
+  const subs = new Map<string, Set<(r: unknown) => void>>();
+  const doc_subscribe = mock(async (nuri: string, _sid: unknown, cb: (r: unknown) => void) => {
+    if (failFor.has(nuri)) throw new Error(`RepoNotFound: ${nuri}`);
+    let set = subs.get(nuri);
+    if (!set) {
+      set = new Set();
+      subs.set(nuri, set);
+    }
+    set.add(cb);
+    // Initial State push, delivered async (as the real RPC does).
+    queueMicrotask(() => cb({ V0: { State: { doc: nuri } } }));
+    return () => set!.delete(cb);
+  });
+  const push = (nuri: string): void => {
+    for (const cb of subs.get(nuri) ?? []) cb({ V0: { Patch: { doc: nuri } } });
+  };
+  const isSubscribed = (nuri: string): boolean => (subs.get(nuri)?.size ?? 0) > 0;
+  return { doc_subscribe, push, isSubscribed, _subs: subs };
+}
+
+function inject(failFor?: Set<string>) {
+  const ng = makeFakeNg(failFor);
+  configure({ ng: ng as any, useShape: (() => {}) as any });
+  configureStoreRegistry({ getSession: async () => SESSION });
+  return ng;
+}
+
+const A = "did:ng:o:docA";
+const B = "did:ng:o:docB";
+
+const tick = () => new Promise((r) => setTimeout(r, 5));
+
+test("subscribeDoc calls ng.doc_subscribe with (nuri, sessionId, callback)", async () => {
+  const ng = inject();
+  const onChange = mock(() => {});
+  subscribeDoc(A, onChange);
+  await tick();
+  expect(ng.doc_subscribe).toHaveBeenCalledTimes(1);
+  const call = ng.doc_subscribe.mock.calls[0]!;
+  expect(call[0]).toBe(A);
+  expect(call[1]).toBe("sid-1"); // sessionId from the injected session
+  expect(typeof call[2]).toBe("function"); // the callback
+});
+
+test("subscribeDoc routes the initial State push and every later change", async () => {
+  const ng = inject();
+  const seen: unknown[] = [];
+  subscribeDoc(A, (r) => seen.push(r));
+  await tick();
+  expect(seen).toHaveLength(1); // initial State push
+  ng.push(A);
+  ng.push(A);
+  expect(seen).toHaveLength(3); // + 2 patches
+});
+
+test("subscribeDoc unsubscribe stops further callbacks", async () => {
+  const ng = inject();
+  const seen: unknown[] = [];
+  const stop = subscribeDoc(A, (r) => seen.push(r));
+  await tick();
+  expect(seen).toHaveLength(1);
+  stop();
+  expect(ng.isSubscribed(A)).toBe(false); // real unsubscribe was invoked
+  ng.push(A); // ignored — no subscriber
+  expect(seen).toHaveLength(1);
+});
+
+test("subscribeDoc unsubscribe BEFORE async setup resolves cancels cleanly", async () => {
+  const ng = inject();
+  const seen: unknown[] = [];
+  const stop = subscribeDoc(A, (r) => seen.push(r));
+  stop(); // before the microtask/promise setup resolved
+  await tick();
+  // The subscription was cancelled the moment setup resolved: no callbacks, and
+  // no lingering subscriber.
+  expect(seen).toHaveLength(0);
+  expect(ng.isSubscribed(A)).toBe(false);
+});
+
+test("subscribeDocs fans out one subscription per doc and reports the source nuri", async () => {
+  const ng = inject();
+  const seen: Array<[string, unknown]> = [];
+  subscribeDocs([A, B], (nuri, r) => seen.push([nuri, r]));
+  await tick();
+  // Two initial pushes, one per doc.
+  expect(seen.map((s) => s[0]).sort()).toEqual([A, B]);
+  ng.push(B);
+  expect(seen.filter((s) => s[0] === B)).toHaveLength(2); // initial + patch
+  expect(seen.filter((s) => s[0] === A)).toHaveLength(1); // isolated: A didn't fire
+});
+
+test("subscribeDocs isolates a failing doc — the others still fire", async () => {
+  const ng = inject(new Set([A])); // A's subscription throws (RepoNotFound)
+  const seen: Array<[string, unknown]> = [];
+  subscribeDocs([A, B], (nuri, r) => seen.push([nuri, r]));
+  await tick();
+  // A failed to subscribe (logged, not thrown); B is unaffected and fired.
+  expect(seen.map((s) => s[0])).toEqual([B]);
+  ng.push(B);
+  expect(seen.filter((s) => s[0] === B)).toHaveLength(2);
+});
+
+test("subscribeDocs unsubscribe tears down all subscriptions", async () => {
+  const ng = inject();
+  const stop = subscribeDocs([A, B], () => {});
+  await tick();
+  expect(ng.isSubscribed(A)).toBe(true);
+  expect(ng.isSubscribed(B)).toBe(true);
+  stop();
+  expect(ng.isSubscribed(A)).toBe(false);
+  expect(ng.isSubscribed(B)).toBe(false);
+});
+
+test("subscribeDocs deduplicates repeated NURIs", async () => {
+  const ng = inject();
+  subscribeDocs([A, A, A], () => {});
+  await tick();
+  expect(ng.doc_subscribe).toHaveBeenCalledTimes(1);
+});