fix(events): forward kv values only to regions with rows that ask

- A region counts its session rows asking for kv values per token and
  announces the token's value variant (`v#<token>`) over the existing
  `watch` item on the first such row, withdrawing it with the last. Add,
  remove, reap, reanchor and refresh all keep it, reading `includeValue` off
  the row as it is dropped.
- A kv write reads the value variants alongside the watch index and attaches
  the value only to forwards bound for a region that announced one.

src/backend/services/events/EventsService.test.ts

@@ -33,6 +33,7 @@ import type { Actor } from '../../core/actor.js';
 import { isHttpError } from '../../core/http/HttpError.js';
 import {
     EventSubscriptionStore,
+    valueWatchToken,
     type DurableSubscription,
 } from '../../stores/events/EventSubscriptionStore.js';
 import type { KvShareHandle } from '../../stores/events/KvShareHandleStore.js';
@@ -2809,6 +2810,42 @@ describe('the cross-app kv gate', () => {
         expect(sent).toEqual([]);
     });
 
+    it('sends no value to a peer whose rows did not ask for one', async () => {
+        await store.noteRemoteWatch(
+            userId,
+            kvAnchorToken(`user-${userId}`, OWN_APP, 'cart'),
+            'east',
+            'add',
+        );
+
+        await dispatchKv(['cart'], { values: [{ items: 3 }] });
+
+        expect(forwarded).toHaveLength(1);
+        expect(forwarded[0].regions).toEqual(['east']);
+        expect(forwarded[0].item.kv).not.toHaveProperty('value');
+    });
+
+    it('sends the value only to the peers whose rows asked for it', async () => {
+        const token = kvAnchorToken(`user-${userId}`, OWN_APP, 'cart');
+        for (const region of ['east', 'south'])
+            await store.noteRemoteWatch(userId, token, region, 'add');
+        await store.noteRemoteWatch(
+            userId,
+            valueWatchToken(token),
+            'east',
+            'add',
+        );
+
+        await dispatchKv(['cart'], { values: [{ items: 3 }] });
+
+        const to = (region: string) =>
+            forwarded.find((call) => call.regions.includes(region));
+        expect(forwarded).toHaveLength(2);
+        expect(to('east')?.regions).toEqual(['east']);
+        expect(to('east')?.item.kv).toMatchObject({ value: { items: 3 } });
+        expect(to('south')?.item.kv).not.toHaveProperty('value');
+    });
+
     it('marks a private key on the way to a peer region', async () => {
         vi.useFakeTimers();
         // A remote watcher is what makes `dispatchKv` forward at all.

src/backend/services/events/EventsService.ts

@@ -3817,34 +3817,62 @@ export class EventsService extends PuterService {
         };
 
         const tokensPerKey = contexts.map((context) => subject.tokens(context));
-        const { local, remote } =
-            await this.stores.eventSubscription.watchedFor(ownerUserId, [
-                ...new Set(tokensPerKey.flat()),
-            ]);
+        const { local, remote, remoteValues } =
+            await this.stores.eventSubscription.watchedFor(
+                ownerUserId,
+                [...new Set(tokensPerKey.flat())],
+                { values: true },
+            );
         if (local.length === 0 && remote.size === 0) return false;
 
         if (!options.forwarded && remote.size > 0)
             contexts.forEach((context, i) => {
                 const regions = new Set<string>();
-                for (const token of tokensPerKey[i])
+                const asking = new Set<string>();
+                for (const token of tokensPerKey[i]) {
                     for (const region of remote.get(token) ?? [])
                         regions.add(region);
+                    for (const region of remoteValues.get(token) ?? [])
+                        asking.add(region);
+                }
                 if (regions.size === 0) return;
-                this.services.eventForward.forwardEvent([...regions], {
-                    family: 'kv',
-                    ownerUserId,
-                    actingUserId: options.actingUserId,
-                    id: context.id,
-                    ts: context.ts,
-                    kv: {
-                        userUuid: namespace.userUuid,
-                        appUid: namespace.appUid,
-                        kvKey: context.kvKey,
-                        op: context.op,
-                        ...valueAt(i),
-                        ...(context.noShare ? { noShare: true as const } : {}),
-                    },
-                });
+                // The value goes only where a row asked for it; everywhere
+                // else it would cross regions to be discarded on arrival.
+                const value = [...regions].some((region) => asking.has(region))
+                    ? valueAt(i)
+                    : undefined;
+                const forward = (
+                    to: string[],
+                    carried?: { value: unknown },
+                ) => {
+                    if (to.length === 0) return;
+                    this.services.eventForward.forwardEvent(to, {
+                        family: 'kv',
+                        ownerUserId,
+                        actingUserId: options.actingUserId,
+                        id: context.id,
+                        ts: context.ts,
+                        kv: {
+                            userUuid: namespace.userUuid,
+                            appUid: namespace.appUid,
+                            kvKey: context.kvKey,
+                            op: context.op,
+                            ...carried,
+                            ...(context.noShare
+                                ? { noShare: true as const }
+                                : {}),
+                        },
+                    });
+                };
+                if (!value) {
+                    forward([...regions]);
+                    return;
+                }
+                forward([...regions].filter((region) => !asking.has(region)));
+                forward(
+                    [...regions].filter((region) => asking.has(region)),
+                    value,
+                );
             });
 
         if (local.length === 0) return false;

src/backend/services/events/forwarding.test.ts

@@ -48,7 +48,11 @@ import type { FSEntry } from '../../stores/fs/FSEntry.js';
 import type { IConfig } from '../../types.js';
 import { EventForwardService } from './EventForwardService.js';
 import { EventsService, type DeliveryEnvelope } from './EventsService.js';
-import type { ForwardBatch, ForwardDelivery } from './forwardQueue.js';
+import type {
+    ForwardBatch,
+    ForwardDelivery,
+    ForwardEvent,
+} from './forwardQueue.js';
 import { fsAnchorToken } from './subjects.js';
 import type {
     WorkerInvocation,
@@ -1922,6 +1926,126 @@ describe('forward-path metrics', () => {
 /** Session forwarding is on by default, so this is the ordinary config. */
 const forwardCfg = { events: { enabled: true } } as Partial<IConfig>;
 
+describe('a kv value crossing to another region', () => {
+    const APP = 'app-cart';
+
+    const subscribeKv = async (
+        region: Region,
+        socketId: string,
+        includeValue?: true,
+    ) =>
+        (
+            await region.events.subscribe(actorFor(), socketId, {
+                subject: `kv:${APP}:cart`,
+                ...(includeValue ? { includeValue } : {}),
+            })
+        ).sub.subId;
+
+    const write = (region: Region) =>
+        region.events.dispatchKv({
+            userId,
+            namespace: `v1:user-${userId}:${APP}`,
+            keys: ['cart'],
+            op: 'set',
+            values: [{ items: 3 }],
+        });
+
+    /** The raw events a region forwarded, in order. */
+    const eventsIn = (region: Region): ForwardEvent[] =>
+        region.posts.flatMap((post) =>
+            post.batch.items.filter(
+                (item): item is ForwardEvent => item.kind === 'event',
+            ),
+        );
+
+    /** Let `region` post past `before`, then let what it queued land. */
+    const announced = async (region: Region, before = 0): Promise<void> => {
+        await posted(region, before + 1);
+        await quiet(100);
+    };
+
+    it('goes nowhere a row did not ask for it', async () => {
+        const west = makeRegion('west', ['east'], forwardCfg);
+        const east = makeRegion('east', ['west'], forwardCfg);
+        await subscribeKv(east, 'socket-plain');
+        await announced(east);
+
+        await write(west);
+        await posted(west);
+        await arrived(east);
+
+        expect(eventsIn(west)).toHaveLength(1);
+        expect(eventsIn(west)[0].kv).not.toHaveProperty('value');
+        expect(east.sent[0].event).not.toHaveProperty('value');
+    });
+
+    it('goes to a region holding a row that asked for it', async () => {
+        const west = makeRegion('west', ['east'], forwardCfg);
+        const east = makeRegion('east', ['west'], forwardCfg);
+        await subscribeKv(east, 'socket-valued', true);
+        await announced(east);
+
+        await write(west);
+        await posted(west);
+        await arrived(east);
+
+        expect(eventsIn(west)[0].kv).toMatchObject({ value: { items: 3 } });
+        expect(east.sent[0].event).toMatchObject({ value: { items: 3 } });
+    });
+
+    it('stops once the last row asking for it is unsubscribed', async () => {
+        const west = makeRegion('west', ['east'], forwardCfg);
+        const east = makeRegion('east', ['west'], forwardCfg);
+        const valued = await subscribeKv(east, 'socket-valued', true);
+        await subscribeKv(east, 'socket-plain');
+        await announced(east);
+        const before = east.posts.length;
+
+        await east.events.unsubscribe(actorFor(), 'socket-valued', {
+            subId: valued,
+        });
+        await announced(east, before);
+        await write(west);
+        await posted(west);
+
+        expect(eventsIn(west)).toHaveLength(1);
+        expect(eventsIn(west)[0].kv).not.toHaveProperty('value');
+    });
+
+    it('stops once the connection holding it is reaped', async () => {
+        const west = makeRegion('west', ['east'], forwardCfg);
+        const east = makeRegion('east', ['west'], forwardCfg);
+        await subscribeKv(east, 'socket-valued', true);
+        await subscribeKv(east, 'socket-plain');
+        await announced(east);
+        const before = east.posts.length;
+
+        await east.events.reapSocket(userId, 'socket-valued');
+        await announced(east, before);
+        await write(west);
+        await posted(west);
+
+        expect(eventsIn(west)[0].kv).not.toHaveProperty('value');
+    });
+
+    it('keeps going while another row there still asks for it', async () => {
+        const west = makeRegion('west', ['east'], forwardCfg);
+        const east = makeRegion('east', ['west'], forwardCfg);
+        const valued = await subscribeKv(east, 'socket-valued', true);
+        await subscribeKv(east, 'socket-also-valued', true);
+        await announced(east);
+
+        await east.events.unsubscribe(actorFor(), 'socket-valued', {
+            subId: valued,
+        });
+        await quiet(100);
+        await write(west);
+        await posted(west);
+
+        expect(eventsIn(west)[0].kv).toMatchObject({ value: { items: 3 } });
+    });
+});
+
 describe('a session subscription in another region', () => {
     it('delivers a write committed in the region that has no rows', async () => {
         const west = makeRegion('west', ['east'], forwardCfg);

src/backend/stores/events/EventSubscriptionStore.test.ts

@@ -25,6 +25,7 @@ import type { IConfig } from '../../types.js';
 import {
     EventSubscriptionStore,
     SESSION_SUBSCRIPTION_TTL_SECONDS,
+    valueWatchToken,
     type DurableSubscription,
     type SessionSubscription,
 } from './EventSubscriptionStore.js';
@@ -762,6 +763,109 @@ describe('the watched-set race', () => {
     });
 });
 
+describe('rows asking for kv values', () => {
+    const VALUE = valueWatchToken('f#anchor');
+    const asking = (over: Partial<SessionSubscription> = {}) =>
+        makeSub({ includeValue: true, ...over });
+
+    it('announces the value watch with the first row asking, and not again', async () => {
+        const first = await store.add(asking());
+        const second = await store.add(asking({ socketId: 'socket-b' }));
+
+        expect(first.announce).toEqual([
+            { token: 'f#anchor', op: 'add' },
+            { token: VALUE, op: 'add' },
+        ]);
+        expect(second.announce).toBeUndefined();
+    });
+
+    it('never announces one for a row that did not ask', async () => {
+        const bump = await store.add(makeSub());
+
+        expect(bump.announce).toEqual([{ token: 'f#anchor', op: 'add' }]);
+    });
+
+    it('withdraws it with the last row asking, and keeps the plain watch', async () => {
+        const valued = asking();
+        await store.add(valued);
+        await store.add(makeSub({ socketId: 'socket-b' }));
+
+        const bump = await store.remove(valued);
+
+        expect(bump?.announce).toEqual([{ token: VALUE, op: 'drop' }]);
+    });
+
+    it('keeps it while another row still asks', async () => {
+        const valued = asking();
+        await store.add(valued);
+        await store.add(asking({ socketId: 'socket-b' }));
+
+        const bump = await store.remove(valued);
+
+        expect(bump?.announce).toBeUndefined();
+    });
+
+    it('withdraws it when the socket holding the row is reaped', async () => {
+        await store.add(asking());
+        await store.add(makeSub({ socketId: 'socket-b' }));
+
+        const [bump] = await store.reapSocket(USER, 'socket-a');
+
+        expect(bump.announce).toEqual([{ token: VALUE, op: 'drop' }]);
+    });
+
+    it('moves it with a row carried to another anchor', async () => {
+        const valued = asking();
+        await store.add(valued);
+
+        const { moved, bumps } = await store.reanchorSession(valued, {
+            ...valued,
+            token: 'f#parent',
+            anchorUid: 'parent',
+            anchorPath: '/testuser',
+        });
+
+        expect(moved).toBe(true);
+        expect(bumps[0].announce).toEqual(
+            expect.arrayContaining([
+                { token: 'f#anchor', op: 'drop' },
+                { token: VALUE, op: 'drop' },
+                { token: 'f#parent', op: 'add' },
+                { token: valueWatchToken('f#parent'), op: 'add' },
+            ]),
+        );
+    });
+
+    it('reasserts it on refresh only for a socket whose rows ask', async () => {
+        await store.add(asking());
+        await store.add(makeSub({ socketId: 'socket-b' }));
+
+        expect(await store.refresh(USER, 'socket-a')).toEqual([
+            { ownerUserId: USER, token: 'f#anchor' },
+            { ownerUserId: USER, token: VALUE },
+        ]);
+        expect(await store.refresh(USER, 'socket-b')).toEqual([
+            { ownerUserId: USER, token: 'f#anchor' },
+        ]);
+    });
+
+    it('answers which peers asked for values alongside which watch', async () => {
+        await store.noteRemoteWatch(USER, 'f#anchor', 'east', 'add');
+        await store.noteRemoteWatch(USER, 'f#anchor', 'south', 'add');
+        await store.noteRemoteWatch(USER, VALUE, 'east', 'add');
+
+        const watched = await store.watchedFor(USER, ['f#anchor'], {
+            values: true,
+        });
+
+        expect(watched.remote.get('f#anchor')?.sort()).toEqual([
+            'east',
+            'south',
+        ]);
+        expect(watched.remoteValues.get('f#anchor')).toEqual(['east']);
+    });
+});
+
 describe('reading rows for dispatch', () => {
     const durableRows = (count: number): DurableSubscription[] =>
         Array.from(

src/backend/stores/events/EventSubscriptionStore.ts

@@ -82,6 +82,10 @@ import type {
  * into _its own_ `ev:rw`: which regions currently have a session watcher on one
  * of this owner's tokens. A write by this owner reads `ev:rw` alongside `ev:w`
  * to decide which peers, if any, get the raw event forwarded to them.
+ *
+ * A token that session rows here ask KV values on is counted and announced a
+ * second time, as its value variant, so a write sends the value only to the
+ * regions that asked for one.
  */
 
 export type {
@@ -105,6 +109,12 @@ const sessionCountKey = (userId: number | string): string =>
     `ev:sc:{${userId}}`;
 const remoteWatchKey = (userId: number | string): string => `ev:rw:{${userId}}`;
 
+/**
+ * The `ev:sc`/`ev:rw` field for session rows on `token` that ask for KV values.
+ * No anchor token starts with `v#`, so it can never name one.
+ */
+export const valueWatchToken = (token: string): string => `v#${token}`;
+
 const safeParseRegions = (raw: string): Record<string, number> | null => {
     try {
         return JSON.parse(raw) as Record<string, number>;
@@ -113,6 +123,30 @@ const safeParseRegions = (raw: string): Record<string, number> | null => {
     }
 };
 
+/** Peers whose announcement of each token is still inside its TTL. */
+const liveRegionsByToken = (
+    tokens: readonly string[],
+    raw: unknown,
+    cutoffMs: number,
+): Map<string, string[]> => {
+    const announced = (raw as Array<string | null> | undefined) ?? [];
+    const live = new Map<string, string[]>();
+    tokens.forEach((token, i) => {
+        const entry = announced[i];
+        if (!entry) return;
+        const regions = safeParseRegions(entry);
+        if (!regions) return;
+        const fresh = Object.entries(regions)
+            .filter(
+                ([, announcedAt]) =>
+                    typeof announcedAt === 'number' && announcedAt >= cutoffMs,
+            )
+            .map(([region]) => region);
+        if (fresh.length > 0) live.set(token, fresh);
+    });
+    return live;
+};
+
 /** A row we cannot read is a row we cannot deliver against. */
 const parseRow = (raw: string): DispatchSubscription | null => {
     try {
@@ -303,23 +337,29 @@ export class EventSubscriptionStore extends PuterStore {
             throw counted?.[0] ?? subscriptionLimitReached();
         }
 
-        const sessionCount = await this.#writeRow(sub);
+        const counts = await this.#writeRow(sub);
         await this.#keepDurableWindow(ownerUserId, [token]);
 
+        const announce: RemoteWatchAnnounce[] = [];
+        if (counts.rows === 1) announce.push({ token, op: 'add' });
+        if (counts.values === 1)
+            announce.push({ token: valueWatchToken(token), op: 'add' });
         return {
             userId: ownerUserId,
             generation: await this.bumpGeneration(ownerUserId),
-            announce: sessionCount === 1 ? [{ token, op: 'add' }] : undefined,
+            announce: announce.length > 0 ? announce : undefined,
         };
     }
 
     /**
-     * Write one row's hash entry, watched-set membership and session count.
-     * Returns the resulting count for that token. Shared by `add` and
-     * `reanchorSession`, which both need the row in place before any ref names
-     * it.
+     * Write one row's hash entry, watched-set membership and session counts.
+     * Returns the resulting counts for that token — `values` only for a row
+     * asking for KV values. Shared by `add` and `reanchorSession`, which both
+     * need the row in place before any ref names it.
      */
-    async #writeRow(sub: SessionSubscription): Promise<number> {
+    async #writeRow(
+        sub: SessionSubscription,
+    ): Promise<{ rows: number; values: number | null }> {
         const { ownerUserId, token, subId } = sub;
         const rows = this.clients.redis.pipeline();
         rows.hset(tokenKey(ownerUserId, token), subId, JSON.stringify(sub));
@@ -330,12 +370,21 @@ export class EventSubscriptionStore extends PuterStore {
         rows.sadd(watchedKey(ownerUserId), token);
         rows.expire(watchedKey(ownerUserId), SESSION_SUBSCRIPTION_TTL_SECONDS);
         rows.hincrby(sessionCountKey(ownerUserId), token, 1);
+        if (sub.includeValue)
+            rows.hincrby(
+                sessionCountKey(ownerUserId),
+                valueWatchToken(token),
+                1,
+            );
         rows.expire(
             sessionCountKey(ownerUserId),
             SESSION_SUBSCRIPTION_TTL_SECONDS,
         );
         const results = (await rows.exec()) ?? [];
-        return Number(results[4]?.[1]);
+        return {
+            rows: Number(results[4]?.[1]),
+            values: sub.includeValue ? Number(results[5]?.[1]) : null,
+        };
     }
 
     /**
@@ -409,7 +458,7 @@ export class EventSubscriptionStore extends PuterStore {
         previous: SessionSubscription,
         next: SessionSubscription,
     ): Promise<ReanchorResult> {
-        const nextCount = await this.#writeRow(next);
+        const nextCounts = await this.#writeRow(next);
 
         const swapped = await this.#scripts().eventsSwapSocketRef(
             socketKey(previous.holderUserId, previous.socketId),
@@ -431,38 +480,50 @@ export class EventSubscriptionStore extends PuterStore {
             // first — undo the row this call wrote speculatively. 2: another
             // settle racing the same move already landed this exact ref;
             // its row is the one in place, so only this call's own count
-            // increment needs undoing.
+            // increments need undoing.
             if (swapped === 0)
                 await this.#dropRows(next.ownerUserId, [
                     { token: next.token, subId: next.subId },
                 ]);
-            const undone =
-                (await this.#dropSessionCounts(next.ownerUserId, [next.token]))
-                    .length > 0;
-            const announced = nextCount === 1;
-            if (announced === undone) return { moved: false, bumps: [] };
+            const counted: Array<[string, number]> = [
+                [next.token, nextCounts.rows],
+            ];
+            if (nextCounts.values !== null)
+                counted.push([valueWatchToken(next.token), nextCounts.values]);
+            const undone = new Set(
+                (
+                    await this.#dropSessionCounts(
+                        next.ownerUserId,
+                        counted.map(([token]) => token),
+                    )
+                ).map((entry) => entry.token),
+            );
+            const announce = counted.flatMap(
+                ([token, count]): RemoteWatchAnnounce[] => {
+                    const announced = count === 1;
+                    if (announced === undone.has(token)) return [];
+                    return [{ token, op: announced ? 'add' : 'drop' }];
+                },
+            );
+            if (announce.length === 0) return { moved: false, bumps: [] };
             return {
                 moved: false,
                 bumps: [
                     {
                         userId: next.ownerUserId,
                         generation: await this.bumpGeneration(next.ownerUserId),
-                        announce: [
-                            {
-                                token: next.token,
-                                op: announced ? 'add' : 'drop',
-                            },
-                        ],
+                        announce,
                     },
                 ],
             };
         }
 
-        await this.#dropRows(previous.ownerUserId, [
+        const valued = await this.#dropRows(previous.ownerUserId, [
             { token: previous.token, subId: previous.subId },
         ]);
         const dropped = await this.#dropSessionCounts(previous.ownerUserId, [
             previous.token,
+            ...valued,
         ]);
         await this.#keepDurableWindow(next.ownerUserId, [next.token]);
 
@@ -472,10 +533,14 @@ export class EventSubscriptionStore extends PuterStore {
                 ...(announceByOwner.get(ownerUserId) ?? []),
                 { token, op: 'drop' },
             ]);
-        if (nextCount === 1)
+        const added: RemoteWatchAnnounce[] = [];
+        if (nextCounts.rows === 1) added.push({ token: next.token, op: 'add' });
+        if (nextCounts.values === 1)
+            added.push({ token: valueWatchToken(next.token), op: 'add' });
+        if (added.length > 0)
             announceByOwner.set(next.ownerUserId, [
                 ...(announceByOwner.get(next.ownerUserId) ?? []),
-                { token: next.token, op: 'add' },
+                ...added,
             ]);
 
         const owners = new Set([previous.ownerUserId, next.ownerUserId]);
@@ -553,12 +618,12 @@ export class EventSubscriptionStore extends PuterStore {
 
         const dropped: DroppedSessionToken[] = [];
         for (const [ownerUserId, owned] of byOwner(removed)) {
-            await this.#dropRows(ownerUserId, owned);
+            const valued = await this.#dropRows(ownerUserId, owned);
             dropped.push(
-                ...(await this.#dropSessionCounts(
-                    ownerUserId,
-                    owned.map((ref) => ref.token),
-                )),
+                ...(await this.#dropSessionCounts(ownerUserId, [
+                    ...owned.map((ref) => ref.token),
+                    ...valued,
+                ])),
             );
         }
         return { dropped, removed };
@@ -605,17 +670,30 @@ export class EventSubscriptionStore extends PuterStore {
      * gone. Emptiness is what un-watches a token, which is what keeps one
      * socket's unsubscribe — or a durable row's removal — from silencing
      * another subscription on the same anchor.
+     *
+     * Returns the value variant of each token a dropped session row asked KV
+     * values on, read off the row as it goes.
      */
     async #dropRows(
         ownerUserId: number,
         rows: ReadonlyArray<{ token: string; subId: string }>,
-    ): Promise<void> {
-        if (rows.length === 0) return;
+    ): Promise<string[]> {
+        if (rows.length === 0) return [];
 
         const drop = this.clients.redis.pipeline();
-        for (const { token, subId } of rows)
+        for (const { token, subId } of rows) {
+            drop.hget(tokenKey(ownerUserId, token), subId);
             drop.hdel(tokenKey(ownerUserId, token), subId);
-        await drop.exec();
+        }
+        const dropResults = (await drop.exec()) ?? [];
+        const valued = rows.flatMap(({ token }, i) => {
+            const raw = dropResults[2 * i]?.[1];
+            if (Number(dropResults[2 * i + 1]?.[1]) !== 1) return [];
+            const row = typeof raw === 'string' ? parseRow(raw) : null;
+            return row?.socketId !== undefined && row.includeValue === true
+                ? [valueWatchToken(token)]
+                : [];
+        });
 
         const tokens = [...new Set(rows.map((row) => row.token))];
         const counts = this.clients.redis.pipeline();
@@ -625,7 +703,7 @@ export class EventSubscriptionStore extends PuterStore {
         const orphaned = tokens.filter(
             (_token, i) => Number(results[i]?.[1] ?? 0) === 0,
         );
-        if (orphaned.length === 0) return;
+        if (orphaned.length === 0) return valued;
 
         await this.clients.redis.srem(watchedKey(ownerUserId), ...orphaned);
 
@@ -642,6 +720,7 @@ export class EventSubscriptionStore extends PuterStore {
         );
         if (revived.length > 0)
             await this.clients.redis.sadd(watchedKey(ownerUserId), ...revived);
+        return valued;
     }
 
     /**
@@ -653,10 +732,10 @@ export class EventSubscriptionStore extends PuterStore {
      * dropped it (see `#dropRows`) heals itself on the next refresh even if
      * nothing catches it sooner.
      *
-     * Returns every (owner, token) pair the socket still holds, so the caller
-     * can re-announce them to peers — the whole of how a remote-watch
-     * announcement survives longer than one refresh window without a second
-     * timer.
+     * Returns every (owner, token) pair the socket still holds, and the value
+     * variant of each one a row asks KV values on, so the caller can
+     * re-announce them to peers — the whole of how a remote-watch announcement
+     * survives longer than one refresh window without a second timer.
      */
     async refresh(
         holderUserId: number,
@@ -689,10 +768,23 @@ export class EventSubscriptionStore extends PuterStore {
                     tokenKey(ownerUserId, token),
                     SESSION_SUBSCRIPTION_TTL_SECONDS,
                 );
-            await pipeline.exec();
+            for (const ref of owned)
+                pipeline.hget(tokenKey(ownerUserId, ref.token), ref.subId);
+            const results = (await pipeline.exec()) ?? [];
+
+            const rowsAt = 3 + tokens.length;
+            const valued = new Set(
+                owned.flatMap((ref, i) => {
+                    const raw = results[rowsAt + i]?.[1];
+                    const row = typeof raw === 'string' ? parseRow(raw) : null;
+                    return row?.includeValue === true ? [ref.token] : [];
+                }),
+            );
 
             await this.#keepDurableWindow(ownerUserId, tokens);
             for (const token of tokens) reasserted.push({ ownerUserId, token });
+            for (const token of valued)
+                reasserted.push({ ownerUserId, token: valueWatchToken(token) });
         }
         return reasserted;
     }
@@ -827,37 +919,40 @@ export class EventSubscriptionStore extends PuterStore {
     async watchedFor(
         ownerUserId: number,
         tokens: readonly string[],
-    ): Promise<{ local: string[]; remote: Map<string, string[]> }> {
-        if (tokens.length === 0) return { local: [], remote: new Map() };
+        options: { values?: boolean } = {},
+    ): Promise<{
+        local: string[];
+        remote: Map<string, string[]>;
+        /**
+         * With `values`: the regions whose rows on each token ask for KV
+         * values.
+         */
+        remoteValues: Map<string, string[]>;
+    }> {
+        if (tokens.length === 0)
+            return { local: [], remote: new Map(), remoteValues: new Map() };
 
         const pipeline = this.clients.redis.pipeline();
         pipeline.smismember(watchedKey(ownerUserId), ...tokens);
         pipeline.hmget(remoteWatchKey(ownerUserId), ...tokens);
+        if (options.values)
+            pipeline.hmget(
+                remoteWatchKey(ownerUserId),
+                ...tokens.map(valueWatchToken),
+            );
         const results = (await pipeline.exec()) ?? [];
 
         const flags = (results[0]?.[1] as number[] | undefined) ?? [];
         const local = tokens.filter((_token, i) => Number(flags[i]) === 1);
 
-        const rawRemote =
-            (results[1]?.[1] as Array<string | null> | undefined) ?? [];
         const cutoffMs = Date.now() - REMOTE_WATCH_TTL_SECONDS * 1000;
-        const remote = new Map<string, string[]>();
-        tokens.forEach((token, i) => {
-            const raw = rawRemote[i];
-            if (!raw) return;
-            const regions = safeParseRegions(raw);
-            if (!regions) return;
-            const live = Object.entries(regions)
-                .filter(
-                    ([, announcedAt]) =>
-                        typeof announcedAt === 'number' &&
-                        announcedAt >= cutoffMs,
-                )
-                .map(([region]) => region);
-            if (live.length > 0) remote.set(token, live);
-        });
-
-        return { local, remote };
+        return {
+            local,
+            remote: liveRegionsByToken(tokens, results[1]?.[1], cutoffMs),
+            remoteValues: options.values
+                ? liveRegionsByToken(tokens, results[2]?.[1], cutoffMs)
+                : new Map(),
+        };
     }
 
     /**