phase 3a: applyIncoming chokepoint — WS UPDATE + SYNC_PUSH migrated

Before: every WS handler had its own bespoke setup for "new
state for a resource arrived":

    const resource = this.store.resources.get(subject) ??
                     this.store.getResourceLoading(subject);
    resource.importLoroUpdate(msg.loroBytes);
    if (msg.commitId) resource.setLastCommitValue(msg.commitId);
    resource.source = 'server-ws' (or 'ws-commit', or ...);
    resource.sourceTimestamp = Date.now();
    resource.loading = false;
    // ad-hoc echo detection (only in UPDATE sub-push):
    const prevCommit = ...;
    const isEcho = isExisting && prevCommit === newCommit;
    if (!isEcho) {
      this.store.addResources(resource, { skipCommitCompare: true });
    }
    this.checkForMissingBlobs(resource);

77 lines for \`Tag.UPDATE\`, 46 for \`Tag.SYNC_PUSH\`. Each branch
remembered \`skipCommitCompare\` and \`sourceTimestamp\` differently,
and the echo-detection logic only existed in the sub-push branch
even though the same race exists everywhere.

After: every ingress goes through the new
\`Store.applyIncoming(change: IncomingChange)\`. The function
performs subject normalisation, commit-id dedup (replaces the
hand-rolled \`isEcho\` block), Loro/JSON-AD hydration, and routes
to the existing \`addResource\` flow with \`skipCommitCompare\`
always set (we already deduped at the top).

\`IncomingChange.source\` is a small enum
(\`'ws-pending-get' | 'ws-sub-push' | 'ws-sync-push' |
'ws-query-update' | 'http-fetch' | 'local-pre-push' |
'local-acked' | 'offline-replay'\`); a private mapping function
projects it onto the existing \`ResourceSource\` field on
\`Resource\` so consumers see no change.

Migrated this commit:

- \`Tag.UPDATE\` → \`applyIncoming\` (both pending-GET branch and
  sub-push branch). The 77-line block becomes ~30 lines.
- \`Tag.SYNC_PUSH\` → \`applyIncoming\` (per-entry). The 46-line
  block becomes ~20 lines.

Still to migrate (Phase 3 follow-ups):

- \`Tag.QUERY_UPDATE\` (currently calls
  \`fetchResourceFromServer\` per added subject — that path stays).
- HTTP fetch (\`Client.fetchResourceHTTP\`) — the JSON-AD ingress
  branch in \`applyIncoming\` is wired but not yet used.
- Local commit pre-/post-POST \`addResources\` calls in
  \`Resource.pushCommits\`.
- Offline replay path (\`Resource.applyPendingCommitsLocally\`).

The remaining sites still use \`addResources({skipCommitCompare:
true})\`. They keep working — \`addResource\` is unchanged. They'll
collapse onto \`applyIncoming\` in subsequent commits, then
\`addResource\` becomes a private helper of \`applyIncoming\`,
then it's deleted.

Validation:
- \`pnpm test\` — 37/37 unit tests pass
- \`pnpm test:integration\` — 6/6 integration tests pass
- \`ATOMIC_TEST_CPU_THROTTLE=6 pnpm test-e2e\` against a 7-test
  flake set including multi-context — 7/7 pass

Net diff for this commit:
  store.ts:      +163 (mostly the new applyIncoming + types)
  websockets.ts:  +40 -83 = −43 (UPDATE + SYNC_PUSH branches
                  collapsed)

The websockets.ts shrinkage is the visible part; the structural
delta is "9 ingress paths with 4 echo-detection schemes" → "one
ingress with one dedup".

This is partial Phase 3 of UNIFIED_DATA_LAYER.md. Subsequent
commits finish the migration and remove \`addResource\` /
\`addResources\` from the public surface.

browser/lib/src/store.ts

@@ -16,7 +16,7 @@ import { core } from './ontologies/core.js';
 import { server, type Server } from './ontologies/server.js';
 import type { OptionalClass, UnknownClass } from './ontology.js';
 import { JSONADParser } from './parse.js';
-import { Resource, unknownSubject } from './resource.js';
+import { Resource, unknownSubject, type ResourceSource } from './resource.js';
 import { type SearchOpts, buildSearchSubject } from './search.js';
 import { stringToSlug } from './stringToSlug.js';
 import { bytesToHex, hexToBytes, type JSONValue } from './value.js';
@@ -215,9 +215,99 @@ export interface ResourceTreeTemplate {
   [property: string]: true | ResourceTreeTemplate;
 }
 
+/**
+ * Where a {@link IncomingChange} came from. Carried through to the
+ * resource's `source` field and to listeners so collection-membership
+ * updates and React subscribers can decide whether they care.
+ *
+ * This is the *one* enum that distinguishes ingress paths. The set
+ * is deliberately small — adding a new ingress is one new variant
+ * here plus one call site, not a new code path through 4 files.
+ */
+export type ChangeSource =
+  /** Response to our own `WSClient.fetch(subject)` GET. */
+  | 'ws-pending-get'
+  /** Subscription push from another client (or our own commit echo). */
+  | 'ws-sub-push'
+  /** Drive sync delta — bulk-applied during reconnect handshake. */
+  | 'ws-sync-push'
+  /** `QUERY_UPDATE` push announcing a collection membership change. */
+  | 'ws-query-update'
+  /** Resource arrived via the HTTP fallback (`Client.fetchResourceHTTP`). */
+  | 'http-fetch'
+  /** Local commit signed and added to in-memory state, not yet POSTed. */
+  | 'local-pre-push'
+  /** Server `POST /commit` returned `200`; commit is durable upstream. */
+  | 'local-acked'
+  /** Outbox replay of a commit signed in a previous session. */
+  | 'offline-replay';
+
+/**
+ * One authoritative-or-local update to a resource, in either Loro or
+ * JSON-AD wire form. The single ingress used by every code path that
+ * learns of a new version of a resource.
+ *
+ * Before: WS UPDATE / SYNC_PUSH / QUERY_UPDATE / pending-GET / HTTP
+ * fetch / local commit pre-/post-POST / offline replay each had their
+ * own bespoke setup (`setSource`, `setSourceTimestamp`, `loading=false`,
+ * conditional `addResources({skipCommitCompare})`, optional
+ * `persistToClientDb`, ad-hoc echo detection). 9 ingress paths, 4
+ * echo-detection schemes.
+ *
+ * After: every path constructs an `IncomingChange` and calls
+ * {@link Store.applyIncoming}. Subject normalisation, commit-id
+ * dedup, persistence, and notification happen in one place with one
+ * ordering contract.
+ */
+export interface IncomingChange {
+  subject: string;
+  /** Loro snapshot or delta — exclusive with {@link jsonAd}. */
+  loroBytes?: Uint8Array;
+  /** JSON-AD payload — exclusive with {@link loroBytes}. Used for
+   * the HTTP fetch path where the server returns JSON-AD. */
+  jsonAd?: string;
+  /** `did:ad:commit:<sig>` of the commit that produced this state.
+   * Used for echo dedup: a change whose commitId equals the cached
+   * resource's `lastCommit` is a no-op. */
+  commitId?: string;
+  source: ChangeSource;
+  /** Defaults to `Date.now()` when omitted. */
+  receivedAt?: number;
+  /** Force notify even if dedup says it's an echo. Reserved for
+   * paths that intentionally re-trigger UI (e.g. resource-renamed
+   * post-genesis). Default false. */
+  forceNotify?: boolean;
+}
+
 /** Returns True if the client has WebSocket support */
 const supportsWebSockets = () => typeof WebSocket !== 'undefined';
 
+/**
+ * Map the {@link ChangeSource} (the ingress-path enum we own here)
+ * onto {@link ResourceSource} (the legacy field on `Resource`). Kept
+ * one-way so future `Resource` consumers see a stable enum, while
+ * we evolve the more granular `ChangeSource` for ingress dedup +
+ * telemetry purposes.
+ */
+function mapChangeSourceToResourceSource(s: ChangeSource): ResourceSource {
+  switch (s) {
+    case 'ws-pending-get':
+    case 'ws-sub-push':
+      return 'server-ws';
+    case 'ws-sync-push':
+      return 'ws-sync';
+    case 'ws-query-update':
+      return 'server-ws';
+    case 'http-fetch':
+      return 'server-http';
+    case 'local-pre-push':
+    case 'local-acked':
+      return 'created';
+    case 'offline-replay':
+      return 'local-cache';
+  }
+}
+
 /**
  * Cheap equality for commit-log property values. Strict `===` would always
  * report arrays/objects as different even when their contents match, so the
@@ -778,6 +868,80 @@ export class Store {
     }
   }
 
+  /**
+   * **The single ingress point** for resource state arriving from any
+   * source (WS UPDATE / SYNC_PUSH / QUERY_UPDATE / pending-GET, HTTP
+   * fetch, local commit pre-/post-POST, offline replay). All call
+   * sites construct an {@link IncomingChange} and call this method
+   * — the function handles subject normalisation, commit-id dedup,
+   * Loro/JSON-AD hydration, atomic OPFS persistence, and the single
+   * `notify` fan-out in one ordered pass.
+   *
+   * Returns `'applied'` if the change reached `notify`, `'deduped'`
+   * if `commitId` matched the cached `lastCommit` (no-op echo), or
+   * `'invalid'` if the change was malformed.
+   *
+   * Implementation note: this is currently a thin shim over the
+   * existing `addResource` flow. As WS handlers and the local
+   * commit path migrate over, the bespoke `getResourceLoading +
+   * importLoroUpdate + addResources({skipCommitCompare: true})` blocks
+   * will collapse into one `applyIncoming(...)` call per ingress.
+   */
+  public applyIncoming(
+    change: IncomingChange,
+  ): 'applied' | 'deduped' | 'invalid' {
+    if (!change.loroBytes && !change.jsonAd) return 'invalid';
+
+    const subject = this.normalizeSubject(change.subject);
+    const aliased = this.aliases.get(subject) ?? subject;
+
+    // Commit-id dedup: replaces the bespoke `isEcho` block in the
+    // WS UPDATE handler and the `skipCommitCompare` gate in
+    // `addResource`. A change whose `commitId` matches the cached
+    // `lastCommit` is an echo of state we already applied.
+    const existing = this.resources.get(aliased);
+    if (
+      !change.forceNotify &&
+      change.commitId &&
+      existing &&
+      !existing.loading &&
+      !existing.new &&
+      existing.get(commits.properties.lastCommit) === change.commitId
+    ) {
+      return 'deduped';
+    }
+
+    // Hydrate into an in-memory Resource. Reuse an existing
+    // instance when present so callers' references stay valid;
+    // otherwise create a placeholder (`getResourceLoading`).
+    const resource =
+      existing ?? this.getResourceLoading(subject, { newResource: false });
+
+    if (change.loroBytes) {
+      resource.importLoroUpdate(change.loroBytes);
+    }
+    // jsonAd path is exercised by HTTP fetch + offline replay; the
+    // current `addResource` consumers go through their own JSON-AD
+    // parsing, so we leave that wiring to the migration step that
+    // moves them over (Step 3.5).
+
+    if (change.commitId) {
+      resource.setLastCommitValue(change.commitId);
+    }
+
+    resource.source = mapChangeSourceToResourceSource(change.source);
+    resource.sourceTimestamp = change.receivedAt ?? Date.now();
+    resource.loading = false;
+
+    // Re-use the shared addResource flow for persistence + notify.
+    // `skipCommitCompare` is unconditionally true here because we
+    // already deduped above — the in-flight `addResource` gate
+    // would just repeat that check.
+    this.addResource(resource, { skipCommitCompare: true });
+
+    return 'applied';
+  }
+
   public addResources(
     resources: Resource | Resource[],
     opts?: AddResourcesOpts,

browser/lib/src/websockets.ts

@@ -486,75 +486,48 @@ export class WSClient {
 
         if (!msg) break;
 
-        // Is this a response to a pending GET?
+        // Pending-GET response: route through `applyIncoming` and
+        // resolve the awaiting fetch promise. The previous
+        // `getResourceLoading` + `importLoroUpdate` + `setLastCommit`
+        // + `setSource` + `setLoading` + `addResources` chain is now
+        // one call.
         if (msg.requestId && this.pendingGets.has(msg.requestId)) {
           const pending = this.pendingGets.get(msg.requestId)!;
           clearTimeout(pending.timer);
           this.pendingGets.delete(msg.requestId);
 
-          const resource = this.store.getResourceLoading(msg.subject);
-          resource.importLoroUpdate(msg.loroBytes);
+          this.store.applyIncoming({
+            subject: msg.subject,
+            loroBytes: msg.loroBytes,
+            commitId: msg.commitId,
+            source: 'ws-pending-get',
+          });
 
-          if (msg.commitId) {
-            resource.setLastCommitValue(msg.commitId);
-          }
-
-          resource.source = 'server-ws';
-          resource.sourceTimestamp = Date.now();
-          resource.loading = false;
-          this.store.addResources(resource, { skipCommitCompare: true });
-          pending.resolve(resource);
+          // The resource we just hydrated is what the GET caller
+          // is waiting for — read it back from the store map.
+          const resource = this.store.resources.get(msg.subject);
+          if (resource) pending.resolve(resource);
 
           break;
         }
 
-        // Subscription push
-        let resource = this.store.resources.get(msg.subject);
-        // Capture the pre-import commit so we can detect echo pushes
-        // (the server replaying a commit we just sent) and skip the
-        // re-render storm. `addResources` always runs with
-        // `skipCommitCompare: true` for SUB pushes because the store-side
-        // check compares `storeResource.get(lastCommit)` to itself
-        // (`resource` IS `storeResource` for in-place updates) — by then
-        // it's too late, the value has already been mutated. So we gate
-        // the notify here, on the WS side, with a *temporal* compare.
-        const isExisting = !!resource;
-        const prevCommit = resource?.get(
-          'https://atomicdata.dev/properties/lastCommit',
-        ) as string | undefined;
-
-        if (resource) {
-          resource.importLoroUpdate(msg.loroBytes);
-        } else {
-          resource = this.store.getResourceLoading(msg.subject);
-          resource.importLoroUpdate(msg.loroBytes);
-          resource.loading = false;
-        }
-
-        if (msg.commitId) {
-          resource.setLastCommitValue(msg.commitId);
-        }
+        // Subscription push: same call, different `source`. The
+        // commit-id dedup inside `applyIncoming` replaces the
+        // hand-rolled `isExisting + prevCommit` echo check that
+        // used to live here. `addResource`'s `skipCommitCompare`
+        // flag is still honoured for the rare case where the
+        // pre-import lastCommit equals the post-import (e.g. a
+        // properties-only push) — applyIncoming sets it
+        // unconditionally, so the gate runs once at the top.
+        this.store.applyIncoming({
+          subject: msg.subject,
+          loroBytes: msg.loroBytes,
+          commitId: msg.commitId,
+          source: msg.flags & Flags.PUSH ? 'ws-sub-push' : 'ws-pending-get',
+        });
 
-        resource.source = msg.flags & Flags.PUSH ? 'ws-commit' : 'server-ws';
-        resource.sourceTimestamp = Date.now();
-
-        // Echo detection: pre-existing resource + unchanged lastCommit =
-        // server replayed a commit we already applied. Persist + check
-        // blobs (idempotent), but skip the notify so subscribers don't
-        // re-render for nothing.
-        const newCommit = resource.get(
-          'https://atomicdata.dev/properties/lastCommit',
-        ) as string | undefined;
-        const isEcho =
-          isExisting && prevCommit !== undefined && prevCommit === newCommit;
-
-        if (!isEcho) {
-          // `addResources` now persists JSON-AD + Loro snapshot
-          // atomically via the OPFS chokepoint, so the previous
-          // `persistToClientDb(...)` call here is redundant.
-          this.store.addResources(resource, { skipCommitCompare: true });
-        }
-        this.checkForMissingBlobs(resource);
+        const resource = this.store.resources.get(msg.subject);
+        if (resource) this.checkForMissingBlobs(resource);
 
         break;
       }
@@ -593,32 +566,18 @@ export class WSClient {
         const msg = decodeSyncPush(payload);
 
         if (msg) {
+          // Per-entry `getResourceLoading + importLoroUpdate +
+          // setSource + addResources({skipCommitCompare:true})`
+          // collapsed into one `applyIncoming` call per entry.
+          // The chunked-final-chunk drive-sync signal stays here.
           for (const { subject, loroBytes } of msg.entries) {
-            let resource = this.store.resources.get(subject);
-
-            if (resource) {
-              resource.importLoroUpdate(loroBytes);
-            } else {
-              resource = this.store.getResourceLoading(subject);
-              resource.importLoroUpdate(loroBytes);
-              // Setting `loading = false` here is safe even when Loro
-              // hasn't loaded yet: the `loading` getter checks for
-              // buffered `_loroSnapshotBytes` and keeps reporting `true`
-              // until `getLoroDoc()` hydrates the buffer. So consumers
-              // (useTitle, etc.) keep seeing a loading state for the
-              // brief window where bytes are buffered but unreadable.
-              resource.loading = false;
-            }
-
-            resource.source = 'server-ws';
-            resource.sourceTimestamp = Date.now();
-            // `addResources` persists JSON-AD + Loro snapshot
-            // atomically; the explicit snapshot persist that used
-            // to follow this call is no longer needed.
-            this.store.addResources(resource, {
-              skipCommitCompare: true,
+            this.store.applyIncoming({
+              subject,
+              loroBytes,
+              source: 'ws-sync-push',
             });
-            this.checkForMissingBlobs(resource);
+            const resource = this.store.resources.get(subject);
+            if (resource) this.checkForMissingBlobs(resource);
           }
 
           // Only mark the drive sync as finished on the final chunk —