feat: payload signals on live references

The session machinery already delivered payload envelopes; the live
adapter dropped them. live.payloads.<name> now carries personalized
payload projections: a newly watched payload name re-sends the
subscription with the grown name list, each payload keeps the
independent per-name revision fence the wire protocol defines, and
payloads evaluate under the live session's authorization context.
payloads joins snapshot as a reserved name on live.

The test watches one payload, sees it flow, then watches a second
payload mid-session and proves the re-subscription delivers both
with per-name fencing and stable signal identity. This removes the
payload-signals item from the deferred list in #24; the thin-page
client, the optimistic layer, and the Lit demo remain follow-ups on
their own merits.

Author: cardmagic

CHANGELOG.md

@@ -10,7 +10,9 @@
   `runtime.realtime` session when first watched and unsubscribe after a
   linger when the last watcher leaves. Value-broadcast observables feed
   named signals; `live.snapshot` re-fetches the authorized snapshot on
-  each accepted envelope; stale revisions are fenced. Works in Node and
+  each accepted envelope; `live.payloads.<name>` carries personalized
+  payload projections under their independent revision fences; stale
+  revisions are fenced. Works in Node and
   in the browser runtime.
 
 ## 0.14.0 - 2026-08-23

docs/api.md

@@ -542,10 +542,15 @@ Behavior:
   envelope. Invalidation-only observables stay `undefined` by design;
   `live.snapshot` re-fetches the authorized snapshot (coalesced) on
   every accepted envelope, so private-value flows read from there.
+- Personalized payload projections arrive as
+  `live.payloads.<name>` signals. A newly watched payload name re-sends
+  the subscription with the grown name list, and each payload keeps the
+  independent per-name revision fence the wire protocol gives it.
+  Payloads evaluate under the live session's authorization context.
 - Envelopes apply only on a monotonic revision advance for the same
   instance, the same fence the browser client uses.
-- `snapshot` is a reserved name on `live`; an observable named
-  `snapshot` is shadowed.
+- `snapshot` and `payloads` are reserved names on `live`; observables
+  with those names are shadowed.
 - `signal-polyfill` is an optional peer dependency. Nothing loads it
   until the `solid-objects/signals` entry is imported; `reference.live`
   throws a pointer to that import otherwise.

src/reference.ts

@@ -171,6 +171,9 @@ export interface LiveSignal<Value> {
 
 export type ActorLiveSignals<ActorType extends Actor> = {
   readonly snapshot: LiveSignal<ActorSnapshot<ActorType> | undefined>
+  readonly payloads: {
+    readonly [name: string]: LiveSignal<DeepReadonly<JsonValue> | undefined>
+  }
 } & {
   readonly [name: string]: LiveSignal<DeepReadonly<JsonValue> | undefined>
 }

src/signals.ts

@@ -17,6 +17,7 @@ export interface LiveSignalsConfiguration {
 }
 
 const SNAPSHOT_KEY = "snapshot"
+const PAYLOADS_KEY = "payloads"
 
 let lingerMilliseconds = 1_000
 let retryMilliseconds = 1_000
@@ -65,6 +66,11 @@ class LiveActorEntry {
   readonly #mirrors = new Map<string, LiveSignal<DeepReadonly<JsonValue> | undefined>>()
   readonly #snapshotState = new Signal.State<ActorSnapshot<Actor> | undefined>(undefined)
   #snapshotMirror: LiveSignal<ActorSnapshot<Actor> | undefined> | undefined
+  readonly #payloadStates = new Map<string, Signal.State<DeepReadonly<JsonValue> | undefined>>()
+  readonly #payloadMirrors = new Map<string, LiveSignal<DeepReadonly<JsonValue> | undefined>>()
+  readonly #payloadWatcherCounts = new Map<string, number>()
+  readonly #payloadFences = new Map<string, { instanceId: string; revision: bigint }>()
+  #payloadsProxy: object | undefined
   #watcherCount = 0
   #snapshotWatcherCount = 0
   #session: LiveSession | undefined
@@ -91,6 +97,49 @@ class LiveActorEntry {
     return mirror
   }
 
+  payloadsProxy(): object {
+    this.#payloadsProxy ??= new Proxy(
+      {},
+      {
+        get: (_target, property) => {
+          if (typeof property !== "string") return undefined
+          return this.payloadSignalFor(property)
+        },
+        has: (_target, property) => typeof property === "string",
+      },
+    )
+    return this.#payloadsProxy
+  }
+
+  payloadSignalFor(name: string): LiveSignal<DeepReadonly<JsonValue> | undefined> {
+    const existing = this.#payloadMirrors.get(name)
+    if (existing) return existing
+    const state = this.payloadStateFor(name)
+    const mirror = new Signal.Computed(() => state.get(), {
+      [Signal.subtle.watched]: () => {
+        this.#payloadWatcherCounts.set(name, (this.#payloadWatcherCounts.get(name) ?? 0) + 1)
+        this.retain({ snapshot: false, payloadName: name })
+      },
+      [Signal.subtle.unwatched]: () => {
+        const count = (this.#payloadWatcherCounts.get(name) ?? 1) - 1
+        if (count <= 0) this.#payloadWatcherCounts.delete(name)
+        else this.#payloadWatcherCounts.set(name, count)
+        this.release({ snapshot: false })
+      },
+    })
+    this.#payloadMirrors.set(name, mirror)
+    return mirror
+  }
+
+  private payloadStateFor(name: string) {
+    let state = this.#payloadStates.get(name)
+    if (!state) {
+      state = new Signal.State<DeepReadonly<JsonValue> | undefined>(undefined)
+      this.#payloadStates.set(name, state)
+    }
+    return state
+  }
+
   snapshotSignal(): LiveSignal<ActorSnapshot<Actor> | undefined> {
     this.#snapshotMirror ??= new Signal.Computed(() => this.#snapshotState.get(), {
       [Signal.subtle.watched]: () => this.retain({ snapshot: true }),
@@ -108,14 +157,15 @@ class LiveActorEntry {
     return state
   }
 
-  private retain(options: { snapshot: boolean }): void {
+  private retain(options: { snapshot: boolean; payloadName?: string }): void {
     this.#watcherCount += 1
     if (options.snapshot) this.#snapshotWatcherCount += 1
     if (this.#linger !== undefined) {
       clearTimeout(this.#linger)
       this.#linger = undefined
     }
     if (!this.#session) this.open()
+    else if (options.payloadName !== undefined) this.sendSubscribe(this.#session)
     else if (options.snapshot) void this.refreshSnapshot()
   }
 
@@ -141,13 +191,7 @@ class LiveActorEntry {
     }) as LiveSession
     this.#session = session
     openSessions.add(this)
-    void session
-      .receive({
-        version: 1,
-        action: "subscribe",
-        actorType: this.#reference.actorType,
-        actorId: this.#reference.actorId,
-      })
+    void this.sendSubscribe(session)
       .then(() => this.refreshSnapshot())
       .catch((error: unknown) => {
         this.#reference.runtime.settings.logger.warn({
@@ -166,6 +210,17 @@ class LiveActorEntry {
       })
   }
 
+  private sendSubscribe(session: LiveSession): Promise<void> {
+    const payloadNames = [...this.#payloadWatcherCounts.keys()]
+    return session.receive({
+      version: 1,
+      action: "subscribe",
+      actorType: this.#reference.actorType,
+      actorId: this.#reference.actorId,
+      ...(payloadNames.length > 0 ? { payloads: payloadNames } : {}),
+    })
+  }
+
   private close(): void {
     if (this.#retry !== undefined) {
       clearTimeout(this.#retry)
@@ -179,6 +234,10 @@ class LiveActorEntry {
   }
 
   private receive(envelope: RealtimeEnvelope): void {
+    if (envelope.kind === "payload") {
+      this.receivePayload(envelope)
+      return
+    }
     if (envelope.kind !== "invalidation") return
     const revision = BigInt(envelope.revision)
     if (this.#instanceId === envelope.instanceId && revision <= this.#revision) return
@@ -190,6 +249,14 @@ class LiveActorEntry {
     void this.refreshSnapshot()
   }
 
+  private receivePayload(envelope: RealtimeEnvelope & { kind: "payload" }): void {
+    const revision = BigInt(envelope.revision)
+    const fence = this.#payloadFences.get(envelope.name)
+    if (fence && fence.instanceId === envelope.instanceId && revision <= fence.revision) return
+    this.#payloadFences.set(envelope.name, { instanceId: envelope.instanceId, revision })
+    this.payloadStateFor(envelope.name).set(envelope.payload as DeepReadonly<JsonValue>)
+  }
+
   private async refreshSnapshot(): Promise<void> {
     if (this.#snapshotWatcherCount === 0) return
     if (this.#refreshing) {
@@ -245,6 +312,7 @@ installLiveSignals((reference) => {
       get(_target, property) {
         if (typeof property !== "string") return undefined
         if (property === SNAPSHOT_KEY) return resolved.snapshotSignal()
+        if (property === PAYLOADS_KEY) return resolved.payloadsProxy()
         return resolved.signalFor(property)
       },
       has(_target, property) {

test/live-signals.test.ts

@@ -34,6 +34,25 @@ class LiveCounter extends Actor {
   }
 }
 
+class LivePayloadCounter extends Actor {
+  static override readonly actorType = "LivePayloadCounter"
+  static override readonly payloads = {
+    personalized: (actor: LivePayloadCounter) => ({ count: actor.count }),
+    doubled: (actor: LivePayloadCounter) => ({ value: actor.count * 2 }),
+  }
+
+  count = 0
+
+  increment(): number {
+    this.count += 1
+    return this.count
+  }
+
+  override observables(): Record<string, unknown> {
+    return { count: broadcastValue(this.count) }
+  }
+}
+
 const runtimes: SolidObjectsRuntime[] = []
 const watchers: Array<{
   watcher: InstanceType<typeof Signal.subtle.Watcher>
@@ -62,6 +81,7 @@ async function liveRuntime(): Promise<SolidObjectsRuntime> {
     syncPollingIntervalMilliseconds: 1,
   })
   runtime.register(LiveCounter)
+  runtime.register(LivePayloadCounter)
   runtimes.push(runtime)
   await runtime.install()
   return runtime
@@ -228,6 +248,30 @@ describe("live signals", () => {
     watcher.unwatch(second.live.count! as never)
   })
 
+  it("delivers personalized payload signals with independent fences", async () => {
+    const runtime = await liveRuntime()
+    const counter = runtime.ref(LivePayloadCounter, "payloaded")
+    watch(counter.live.payloads.personalized!)
+    await eventually(() => activeLiveSubscriptionCount() === 1)
+
+    await counter.increment()
+    await runtime.testing.drain({ roles: ["actors", "broadcasts"] })
+    await eventually(() => {
+      const payload = counter.live.payloads.personalized!.get() as { count?: number } | undefined
+      return payload?.count === 1
+    })
+
+    watch(counter.live.payloads.doubled!)
+    await counter.increment()
+    await runtime.testing.drain({ roles: ["actors", "broadcasts"] })
+    await eventually(() => {
+      const doubled = counter.live.payloads.doubled!.get() as { value?: number } | undefined
+      return doubled?.value === 4
+    })
+    expect((counter.live.payloads.personalized!.get() as { count?: number }).count).toBe(2)
+    expect(counter.live.payloads.personalized).toBe(counter.live.payloads.personalized)
+  })
+
   it("exposes read-only signals and a stable proxy", () => {
     const runtime = createRuntime({
       database: sqlite({ path: ":memory:" }),