feat: kv events value opt in (#3894)

src/backend/clients/database/SqliteDatabaseClient.ts

@@ -114,6 +114,7 @@ const AVAILABLE_MIGRATIONS: [number, string[]][] = [
     [77, ['0082_temp-password-expiry.sql']],
     [78, ['0083_team-directory.sql']],
     [79, ['0084_share-anyone-with-link.sql']],
+    [80, ['0085_event-subscriptions-include-value.sql']],
 ];
 
 export class SqliteDatabaseClient extends AbstractDatabaseClient {

src/backend/clients/database/migrations/mysql/mysql_mig_40.sql

@@ -0,0 +1,22 @@
+-- Copyright (C) 2024-present Puter Technologies Inc.
+--
+-- This file is part of Puter.
+--
+-- Puter is free software: you can redistribute it and/or modify
+-- it under the terms of the GNU Affero General Public License as published
+-- by the Free Software Foundation, either version 3 of the License, or
+-- (at your option) any later version.
+--
+-- This program is distributed in the hope that it will be useful,
+-- but WITHOUT ANY WARRANTY; without even the implied warranty of
+-- MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+-- GNU Affero General Public License for more details.
+--
+-- You should have received a copy of the GNU Affero General Public License
+-- along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+-- See sqlite/0085_event-subscriptions-include-value.sql for the column
+-- rationale. No per-file applied-state tracking, so it goes through
+-- _puter_add_col as mysql_mig_39 does.
+
+CALL _puter_add_col('event_subscriptions', 'include_value', '`include_value` tinyint(1) NOT NULL DEFAULT 0');

src/backend/clients/database/migrations/postgres/postgres_mig_29.sql

@@ -0,0 +1,22 @@
+-- Copyright (C) 2024-present Puter Technologies Inc.
+--
+-- This file is part of Puter.
+--
+-- Puter is free software: you can redistribute it and/or modify
+-- it under the terms of the GNU Affero General Public License as published
+-- by the Free Software Foundation, either version 3 of the License, or
+-- (at your option) any later version.
+--
+-- This program is distributed in the hope that it will be useful,
+-- but WITHOUT ANY WARRANTY; without even the implied warranty of
+-- MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+-- GNU Affero General Public License for more details.
+--
+-- You should have received a copy of the GNU Affero General Public License
+-- along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+-- See sqlite/0085_event-subscriptions-include-value.sql for the column
+-- rationale. Idempotent via IF NOT EXISTS; there is no per-file applied-state
+-- tracking.
+
+ALTER TABLE event_subscriptions ADD COLUMN IF NOT EXISTS include_value smallint NOT NULL DEFAULT 0;

src/backend/clients/database/migrations/sqlite/0085_event-subscriptions-include-value.sql

@@ -0,0 +1,23 @@
+-- Copyright (C) 2024-present Puter Technologies Inc.
+--
+-- This file is part of Puter.
+--
+-- Puter is free software: you can redistribute it and/or modify
+-- it under the terms of the GNU Affero General Public License as published
+-- by the Free Software Foundation, either version 3 of the License, or
+-- (at your option) any later version.
+--
+-- This program is distributed in the hope that it will be useful,
+-- but WITHOUT ANY WARRANTY; without even the implied warranty of
+-- MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+-- GNU Affero General Public License for more details.
+--
+-- You should have received a copy of the GNU Affero General Public License
+-- along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+-- A key-value subscription may ask for the key's new value to ride along with
+-- each delivery. Off by default: a delivery that carries nothing but the key is
+-- what every existing row was made under.
+
+ALTER TABLE `event_subscriptions`
+    ADD COLUMN `include_value` INTEGER NOT NULL DEFAULT 0;

src/backend/clients/event/types.ts

@@ -423,6 +423,12 @@ export type EventMap = {
         userId: number;
         keys: string[];
         op: KvOp;
+        /**
+         * What each key holds after the change, aligned with `keys`: the
+         * written value on a `set`, `null` on a `del`. Absent when the write
+         * did not have it in hand, as an `expire` does not.
+         */
+        values?: unknown[];
     };
     /**
      * A whole namespace was emptied. Namespace-level on purpose: `flush`'s own

src/backend/controllers/events/limits.ts

@@ -262,6 +262,14 @@ export const EVENTS_WORKER_LIST_LIMIT = userWindow('events:workers:list', 120);
  */
 export const EVENTS_MATCHED_SUBSCRIPTIONS_PER_EVENT = 50;
 
+/**
+ * Largest key-value value a delivery inlines, in serialized bytes. A value over
+ * this is left out and the subscriber re-reads the key: a delivery fans out to
+ * many rows, may cross regions and may sit in a backlog, none of which is sized
+ * for the store's own ceiling.
+ */
+export const EVENTS_KV_VALUE_MAX_BYTES = 16 * 1024;
+
 /**
  * Broadcast deliveries per minute, per subscription.
  *

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

@@ -22,6 +22,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
 import {
     EVENTS_BROADCAST_DELIVERY_LIMIT,
     EVENTS_COALESCE_WINDOW_MS,
+    EVENTS_KV_VALUE_MAX_BYTES,
     EVENTS_MATCHED_SUBSCRIPTIONS_PER_EVENT,
     EVENTS_SUBSCRIBE_LIMIT,
 } from '../../controllers/events/limits.js';
@@ -36,6 +37,7 @@ import type { FSEntry } from '../../stores/fs/FSEntry.js';
 import type { UsageInput } from '../metering/types.js';
 import type { IConfig } from '../../types.js';
 import { EVENTS_COSTS } from './costs.js';
+import type { ForwardEvent } from './forwardQueue.js';
 import {
     EventsService,
     EVENTS_ACK_VERB,
@@ -421,7 +423,11 @@ const dispatch = async (node: FSEntry, key = 'fs.write.file' as const) =>
 /** Dispatch as the KV store's bus announcement does. */
 const dispatchKv = async (
     keys: string[],
-    options: { appUid?: string; op?: 'set' | 'del' | 'expire' } = {},
+    options: {
+        appUid?: string;
+        op?: 'set' | 'del' | 'expire';
+        values?: unknown[];
+    } = {},
     on: EventsService = service,
 ) =>
     on.dispatchKv({
@@ -429,6 +435,7 @@ const dispatchKv = async (
         namespace: `v1:user-${userId}:${options.appUid ?? OWN_APP}`,
         keys,
         op: options.op ?? 'set',
+        ...(options.values ? { values: options.values } : {}),
     });
 
 /** The app the KV tests act as, so "own namespace" has something to be. */
@@ -1598,6 +1605,44 @@ it('names the delivery channel the clients listen on', () => {
 
 // -- KV subjects -----------------------------------------------------
 
+describe('asking a kv subscription for the value', () => {
+    it('is recorded on the row and reported in the view', async () => {
+        const { sub } = await service.subscribe(appActorFor(OWN_APP), socketId, {
+            subject: `kv:${OWN_APP}:cart`,
+            includeValue: true,
+        });
+        expect(sub.includeValue).toBe(true);
+
+        const plain = await subscribeKv(`kv:${OWN_APP}:cart`);
+        expect(plain.includeValue).toBe(false);
+    });
+
+    it('is refused on anything but a kv subject', async () => {
+        const { documents } = seedTree();
+        await expect(
+            service.subscribe(actorFor(), socketId, {
+                subject: `fs:${documents.path}`,
+                includeValue: true,
+            }),
+        ).rejects.toSatisfy(
+            (err: unknown) =>
+                isHttpError(err) && err.legacyCode === 'invalid_include_value',
+        );
+    });
+
+    it('is a flag, not a string', async () => {
+        await expect(
+            service.subscribe(appActorFor(OWN_APP), socketId, {
+                subject: `kv:${OWN_APP}:cart`,
+                includeValue: 'yes',
+            }),
+        ).rejects.toSatisfy(
+            (err: unknown) =>
+                isHttpError(err) && err.legacyCode === 'invalid_include_value',
+        );
+    });
+});
+
 describe('resolving a kv subject', () => {
     it('anchors an exact key on the key itself, with no filter', async () => {
         const sub = await subscribeKv(`kv:${OWN_APP}:cart`);
@@ -1756,6 +1801,110 @@ describe('delivering a kv change', () => {
         expect(sent[0].envelope.event).toMatchObject({ op: 'del' });
     });
 
+    const subscribeKvForValue = async (subject: string) =>
+        (
+            await service.subscribe(appActorFor(OWN_APP), socketId, {
+                subject,
+                includeValue: true,
+            })
+        ).sub;
+
+    const eventFor = (subId: string) =>
+        sent.find((one) => one.envelope.subId === subId)?.envelope.event as
+            | Record<string, unknown>
+            | undefined;
+
+    it('hands the value to the row that asked for it and to no other', async () => {
+        vi.useFakeTimers();
+        const asking = await subscribeKvForValue(`kv:${OWN_APP}:cart`);
+        const silent = await subscribeKv(`kv:${OWN_APP}:cart`);
+
+        await dispatchKv(['cart'], { values: [{ items: [1, 2] }] });
+        await vi.advanceTimersByTimeAsync(EVENTS_COALESCE_WINDOW_MS + 1);
+
+        expect(sent).toHaveLength(2);
+        expect(eventFor(asking.subId)).toMatchObject({
+            key: 'cart',
+            value: { items: [1, 2] },
+        });
+        expect(eventFor(silent.subId)).not.toHaveProperty('value');
+    });
+
+    it('aligns values with keys across a batch', async () => {
+        vi.useFakeTimers();
+        await subscribeKvForValue(`kv:${OWN_APP}:cart:*`);
+
+        await dispatchKv(['cart:a', 'cart:b'], { values: ['A', 'B'] });
+        await vi.advanceTimersByTimeAsync(EVENTS_COALESCE_WINDOW_MS + 1);
+
+        expect(
+            sent.map((one) => one.envelope.event as Record<string, unknown>),
+        ).toMatchObject([
+            { key: 'cart:a', value: 'A' },
+            { key: 'cart:b', value: 'B' },
+        ]);
+    });
+
+    it('carries null for a deletion and nothing for an expire', async () => {
+        vi.useFakeTimers();
+        const sub = await subscribeKvForValue(`kv:${OWN_APP}:cart`);
+
+        await dispatchKv(['cart'], { op: 'del', values: [null] });
+        await vi.advanceTimersByTimeAsync(EVENTS_COALESCE_WINDOW_MS + 1);
+        expect(eventFor(sub.subId)).toHaveProperty('value', null);
+
+        sent.length = 0;
+        await dispatchKv(['cart'], { op: 'expire' });
+        await vi.advanceTimersByTimeAsync(EVENTS_COALESCE_WINDOW_MS + 1);
+        expect(eventFor(sub.subId)).toMatchObject({ op: 'expire' });
+        expect(eventFor(sub.subId)).not.toHaveProperty('value');
+    });
+
+    it('leaves out a value too large to inline', async () => {
+        vi.useFakeTimers();
+        const sub = await subscribeKvForValue(`kv:${OWN_APP}:cart`);
+
+        await dispatchKv(['cart'], {
+            values: ['x'.repeat(EVENTS_KV_VALUE_MAX_BYTES + 1)],
+        });
+        await vi.advanceTimersByTimeAsync(EVENTS_COALESCE_WINDOW_MS + 1);
+
+        expect(eventFor(sub.subId)).toMatchObject({ op: 'set', key: 'cart' });
+        expect(eventFor(sub.subId)).not.toHaveProperty('value');
+    });
+
+    it('carries the value a peer region forwarded', async () => {
+        vi.useFakeTimers();
+        const sub = await subscribeKvForValue(`kv:${OWN_APP}:cart`);
+
+        const item: ForwardEvent = {
+            kind: 'event',
+            family: 'kv',
+            ownerUserId: userId,
+            actingUserId: userId,
+            id: 'ev-remote',
+            ts: 1_700_000_000,
+            sessionOnly: true,
+            hop: 1,
+            kv: {
+                userUuid: `user-${userId}`,
+                appUid: OWN_APP,
+                kvKey: 'cart',
+                op: 'set',
+                value: { from: 'afar' },
+            },
+        };
+        await expect(service.dispatchForwarded(item)).resolves.toMatchObject({
+            matched: true,
+        });
+        await vi.advanceTimersByTimeAsync(EVENTS_COALESCE_WINDOW_MS + 1);
+
+        expect(eventFor(sub.subId)).toMatchObject({
+            id: 'ev-remote',
+            value: { from: 'afar' },
+        });
+    });
+
     it('leaves another app`s namespace alone', async () => {
         vi.useFakeTimers();
         await subscribeKv(`kv:${OWN_APP}:cart`);
@@ -2061,6 +2210,28 @@ describe('cross-user kv handles', () => {
         expect(wire).not.toContain(`u${userId}`);
     });
 
+    it('never delivers values: a handle grants watching, not reading', async () => {
+        mintHandle();
+        await expect(
+            service.subscribe(actorFor(guestId), socketId, {
+                subject: `kv:${handle}:*`,
+                includeValue: true,
+            }),
+        ).rejects.toSatisfy(
+            (err: unknown) =>
+                isHttpError(err) &&
+                err.legacyCode === 'events_kv_handle_no_values',
+        );
+
+        // And a value on the wire never reaches a guest row either way.
+        vi.useFakeTimers();
+        const { sub } = await subscribeAsGuest(`kv:${handle}:*`);
+        await dispatchKv([`${PREFIX}title`], { values: ['secret'] });
+        await vi.advanceTimersByTimeAsync(EVENTS_COALESCE_WINDOW_MS + 1);
+        expect(sent[0].envelope.subId).toBe(sub.subId);
+        expect(sent[0].envelope.event).not.toHaveProperty('value');
+    });
+
     it('delivers every key under the granted region', async () => {
         mintHandle();
         vi.useFakeTimers();

src/backend/services/events/EventsService.ts

@@ -33,6 +33,7 @@ import {
     EVENTS_KV_HANDLE_LIMIT,
     EVENTS_KV_HANDLES_PER_APP,
     EVENTS_KV_HANDLES_PER_USER,
+    EVENTS_KV_VALUE_MAX_BYTES,
     EVENTS_MATCHED_SUBSCRIPTIONS_PER_EVENT,
     EVENTS_SINGLE_DELIVERY_LIMIT,
     EVENTS_SUBSCRIBE_LIMIT,
@@ -261,6 +262,8 @@ import {
 export interface SubscribeRequest {
     subject?: unknown;
     targets?: unknown;
+    /** KV subjects only: deliver the key's new value alongside the key. */
+    includeValue?: unknown;
 }
 
 export interface UnsubscribeRequest {
@@ -386,6 +389,8 @@ export interface SubscriptionView {
     match: string | null;
     op: FsOp | null;
     targets: SubscriptionTarget[];
+    /** Whether KV deliveries on this row carry the key's new value. */
+    includeValue: boolean;
 }
 
 /**
@@ -599,6 +604,8 @@ export interface KvDispatchInput {
     namespace: string;
     keys: readonly string[];
     op: KvOp;
+    /** What each key now holds, aligned with `keys`; absent when unknown. */
+    values?: readonly unknown[];
 }
 
 // -- Socket wire names ------------------------------------------------
@@ -901,6 +908,7 @@ const toView = (sub: DispatchSubscription): SubscriptionView => {
                 : sub.match,
         op: sub.op,
         targets: sub.targets ?? SESSION_TARGETS,
+        includeValue: sub.includeValue === true,
     };
 };
 
@@ -1112,6 +1120,59 @@ const isCrossAppKvRow = (
 
 // -- Durable request parsing ------------------------------------------
 
+/** A flag: `true` to opt in, anything falsy for the default. */
+const parseIncludeValue = (value: unknown): true | undefined => {
+    if (value === undefined || value === null || value === false)
+        return undefined;
+    if (value === true) return true;
+    throw badRequest('includeValue must be a boolean', 'invalid_include_value');
+};
+
+/**
+ * A value rides only on a key-value row, and never through a share handle: the
+ * grant behind a handle is to watch a region, not to read it, and a delivery
+ * carrying the value would be a read the grant never gave. Decided on the raw
+ * subject, before anything is resolved, so the refusal names this and not
+ * whatever resolution would have said.
+ */
+const assertValueDeliverable = (rawSubject: string): void => {
+    if (parseSubject(rawSubject).family !== 'kv')
+        throw badRequest(
+            'includeValue applies to kv: subjects only',
+            'invalid_include_value',
+        );
+    if (kvHandleFromSubject(rawSubject) !== null)
+        throw badRequest(
+            'A share handle does not deliver values',
+            'events_kv_handle_no_values',
+        );
+};
+
+/**
+ * The value as a delivery may carry it: the value itself under the cap, nothing
+ * over it. `undefined` never rides — it is what "no value in hand" looks like.
+ */
+const inlineKvValue = (value: unknown): { value: unknown } | undefined => {
+    if (value === undefined) return undefined;
+    const bytes = Buffer.byteLength(JSON.stringify(value) ?? 'null', 'utf8');
+    return bytes > EVENTS_KV_VALUE_MAX_BYTES ? undefined : { value };
+};
+
+/**
+ * The value rides only where the row asked for it and its holder owns the
+ * namespace — never across a share handle, whose grant is to watch, not read.
+ */
+const valueAsRowAskedFor = (
+    row: DispatchSubscription,
+    event: ProjectedKvEvent,
+): ProjectedKvEvent => {
+    if (event.value === undefined) return event;
+    if (row.includeValue === true && row.holderUserId === row.ownerUserId)
+        return event;
+    const { value: _value, ...withoutValue } = event;
+    return withoutValue;
+};
+
 const parseDelivery = (value: unknown): DeliveryClass => {
     if (value === undefined || value === null || value === 'broadcast')
         return 'broadcast';
@@ -1564,7 +1625,9 @@ export class EventsService extends PuterService {
         await this.#spendCallBudget(holderUserId);
 
         const targets = parseSessionTargets(request?.targets);
+        const includeValue = parseIncludeValue(request?.includeValue);
         const rawSubject = String(request?.subject ?? '');
+        if (includeValue) assertValueDeliverable(rawSubject);
         const anchor = await this.#resolveSubscribeAnchor(actor, rawSubject);
 
         const sub: SessionSubscription = {
@@ -1581,6 +1644,7 @@ export class EventsService extends PuterService {
             appUid: actor.effectiveApp?.uid ?? null,
             permission: anchor.permission,
             targets,
+            ...(includeValue ? { includeValue } : {}),
         };
 
         const bump = await this.stores.eventSubscription.add(sub);
@@ -1662,6 +1726,7 @@ export class EventsService extends PuterService {
         const handlerHash = parseHandlerHash(request?.handlerHash);
         const context = parseContext(request?.context);
         const expiresAt = parseExpiresAt(request?.expiresAt);
+        const includeValue = parseIncludeValue(request?.includeValue);
 
         // A `single` is owed to exactly one consumer, and the handler is the
         // only one that is always there to take it.
@@ -1679,6 +1744,7 @@ export class EventsService extends PuterService {
             await this.#assertHandlerBinding(appUid, handlerName, handlerHash);
 
         const rawSubject = String(request?.subject ?? '');
+        if (includeValue) assertValueDeliverable(rawSubject);
         const anchor = await this.#resolveSubscribeAnchor(actor, rawSubject);
 
         const { row, bump } = await this.stores.durableSubscription.create({
@@ -1698,6 +1764,7 @@ export class EventsService extends PuterService {
             permission: anchor.permission,
             expiresAt,
             limits,
+            ...(includeValue ? { includeValue } : {}),
         });
         this.#publishGeneration(bump, true);
 
@@ -3525,7 +3592,8 @@ export class EventsService extends PuterService {
      * reaching another app's namespace still reaches it under its own user.
      *
      * A batch is one bus event over many keys, so the watched-set check is one
-     * command for the whole batch rather than one per key.
+     * command for the whole batch rather than one per key. A value is measured
+     * once per key, and only once something is listening for it.
      */
     async dispatchKv(
         input: KvDispatchInput,
@@ -3557,6 +3625,11 @@ export class EventsService extends PuterService {
             id: options.forwarded && options.id ? options.id : randomUUID(),
             ts,
         }));
+        const carried: Array<{ value: unknown } | undefined> = [];
+        const valueAt = (i: number): { value: unknown } | undefined => {
+            if (!(i in carried)) carried[i] = inlineKvValue(input.values?.[i]);
+            return carried[i];
+        };
 
         const tokensPerKey = contexts.map((context) => subject.tokens(context));
         const { local, remote } =
@@ -3583,6 +3656,7 @@ export class EventsService extends PuterService {
                         appUid: namespace.appUid,
                         kvKey: context.kvKey,
                         op: context.op,
+                        ...valueAt(i),
                     },
                 });
             });
@@ -3597,6 +3671,11 @@ export class EventsService extends PuterService {
             rows = rows.filter((row) => row.socketId !== undefined);
         if (rows.length === 0) return false;
 
+        if (rows.some((row) => row.includeValue === true))
+            contexts.forEach((context, i) =>
+                Object.assign(context, valueAt(i)),
+            );
+
         // Indexed once: a row holds one token, so a key's candidates are the
         // rows under the tokens it enumerated.
         const byToken = new Map<string, DispatchSubscription[]>();
@@ -3688,6 +3767,9 @@ export class EventsService extends PuterService {
                     namespace: `v1:${item.kv.userUuid}:${item.kv.appUid}`,
                     keys: [item.kv.kvKey],
                     op: item.kv.op,
+                    ...(item.kv.value !== undefined
+                        ? { values: [item.kv.value] }
+                        : {}),
                 },
                 {
                     actingUserId: item.actingUserId,
@@ -3912,15 +3994,13 @@ export class EventsService extends PuterService {
             if (!isFsToken(row.token)) return event;
             return this.#asRecipientAddressesIt(row, event);
         }
+        const kv = valueAsRowAskedFor(row, event as ProjectedKvEvent);
         const handle = kvHandleFromSubject(row.subject);
-        if (handle === null) return event;
+        if (handle === null) return kv as P;
 
-        const key = relativeToKvShareRoot(
-            row.permission,
-            (event as ProjectedKvEvent).key,
-        );
+        const key = relativeToKvShareRoot(row.permission, kv.key);
         if (key === null) return null;
-        return { ...event, subject: `kv:${handle}:${key}`, key };
+        return { ...kv, subject: `kv:${handle}:${key}`, key } as P;
     }
 
     /**

src/backend/services/events/forwardQueue.ts

@@ -95,7 +95,14 @@ export interface ForwardEvent {
             ancestors: Array<{ uid: string; path: string }>;
         };
     };
-    kv?: { userUuid: string; appUid: string; kvKey: string; op: KvOp };
+    kv?: {
+        userUuid: string;
+        appUid: string;
+        kvKey: string;
+        op: KvOp;
+        /** The value after the change, when it is small enough to carry. */
+        value?: unknown;
+    };
 }
 
 /** A subscription-set or presence generation moved in another region. */
@@ -110,7 +117,11 @@ export interface ForwardBump {
 }
 
 export type ForwardItem =
-    ForwardDelivery | ForwardAck | ForwardWatch | ForwardEvent | ForwardBump;
+    | ForwardDelivery
+    | ForwardAck
+    | ForwardWatch
+    | ForwardEvent
+    | ForwardBump;
 
 /** One batch, as a peer receives it. */
 export interface ForwardBatch {
@@ -323,7 +334,7 @@ const isGapMarker = (item: ForwardItem): boolean =>
  */
 const shed = (queue: PeerQueue, count: number): ForwardItem[] => {
     const dropped: ForwardItem[] = [];
-    for (let i = 0; i < queue.items.length && dropped.length < count;) {
+    for (let i = 0; i < queue.items.length && dropped.length < count; ) {
         if (isGapMarker(queue.items[i])) {
             i++;
             continue;
@@ -346,7 +357,7 @@ const shed = (queue: PeerQueue, count: number): ForwardItem[] => {
 const shedBytes = (queue: PeerQueue, maxBytesHeld: number): ForwardItem[] => {
     const dropped: ForwardItem[] = [];
     let remaining = queue.bytes;
-    for (let i = 0; i < queue.items.length && remaining > maxBytesHeld;) {
+    for (let i = 0; i < queue.items.length && remaining > maxBytesHeld; ) {
         if (isGapMarker(queue.items[i])) {
             i++;
             continue;

src/backend/services/events/kv.integration.test.ts

@@ -104,14 +104,31 @@ const kvSet = async (
     );
 };
 
-const subscribe = async (subject: string, token: string) => {
+const subscribe = async (
+    subject: string,
+    token: string,
+    extra: { includeValue?: boolean } = {},
+) => {
     const actor = await actorFor(token);
-    return (await events().subscribe(actor, SOCKET_ID, { subject })).sub;
+    return (await events().subscribe(actor, SOCKET_ID, { subject, ...extra }))
+        .sub;
 };
 
-const subscribeDurable = async (subject: string, token: string) => {
+const subscribeDurable = async (
+    subject: string,
+    token: string,
+    extra: { includeValue?: boolean } = {},
+) => {
     const actor = await actorFor(token);
-    return (await events().subscribeDurable(actor, { subject })).sub;
+    return (await events().subscribeDurable(actor, { subject, ...extra })).sub;
+};
+
+/** `kv.del` as a caller makes it: through the driver, as this app. */
+const kvDel = async (token: string, key: string): Promise<void> => {
+    const actor = await actorFor(token);
+    await runWithContext({ actor }, () =>
+        env.server.drivers.kvStore.del({ key }),
+    );
 };
 
 const clearRows = async () => {
@@ -196,6 +213,83 @@ describe('a kv write reaches its subscribers', () => {
     });
 });
 
+describe('a subscription that asked for the value', () => {
+    // Session rows from earlier tests stay live on the shared socket, so
+    // every assertion here reads its own row's delivery rather than the first.
+    const eventFor = (subId: string) =>
+        vi.waitFor(
+            () => {
+                const found = delivered.find((one) => one.subId === subId);
+                expect(found).toBeDefined();
+                return found!.event as Record<string, unknown>;
+            },
+            { timeout: EVENTS_COALESCE_WINDOW_MS * 12, interval: 25 },
+        );
+
+    it('is handed what the key now holds, and a row that did not ask is not', async () => {
+        const asking = await subscribe(`kv:${ownAppUid}:value:*`, ownAppToken, {
+            includeValue: true,
+        });
+        const silent = await subscribe(`kv:${ownAppUid}:value:*`, ownAppToken);
+        expect(asking.includeValue).toBe(true);
+        expect(silent.includeValue).toBe(false);
+        delivered.length = 0;
+
+        await kvSet(ownAppToken, 'value:items', { qty: 2 });
+
+        expect(await eventFor(asking.subId)).toMatchObject({
+            op: 'set',
+            key: 'value:items',
+            value: { qty: 2 },
+        });
+        expect(await eventFor(silent.subId)).not.toHaveProperty('value');
+    });
+
+    it('is handed null once the key is gone', async () => {
+        await kvSet(ownAppToken, 'value:gone', 1);
+        const sub = await subscribe(`kv:${ownAppUid}:value:gone`, ownAppToken, {
+            includeValue: true,
+        });
+        delivered.length = 0;
+
+        await kvDel(ownAppToken, 'value:gone');
+
+        expect(await eventFor(sub.subId)).toMatchObject({
+            op: 'del',
+            value: null,
+        });
+    });
+
+    it('is refused on a subject that is not a key', async () => {
+        await expect(
+            subscribe(`fs:/${env.users.user.username}`, ownAppToken, {
+                includeValue: true,
+            }),
+        ).rejects.toMatchObject({ legacyCode: 'invalid_include_value' });
+    });
+
+    it('keeps the flag on a durable row across a region rebuild', async () => {
+        await clearRows();
+        const sub = await subscribeDurable(
+            `kv:${ownAppUid}:orders:*`,
+            ownAppToken,
+            { includeValue: true },
+        );
+        expect(sub.includeValue).toBe(true);
+
+        events().invalidateUser(userId);
+        await env.server.stores.eventSubscription.markRegionCold(userId);
+        await env.server.clients.redis.del(`ev:w:{${userId}}`);
+        delivered.length = 0;
+
+        await kvSet(ownAppToken, 'orders:2', { total: 5 });
+
+        expect(await eventFor(sub.subId)).toMatchObject({
+            value: { total: 5 },
+        });
+    });
+});
+
 describe('a durable kv subscription', () => {
     it('survives the region cache being rebuilt from the table', async () => {
         await clearRows();
@@ -270,6 +364,24 @@ describe('the cross-app gate against real grants', () => {
         await revokeRead(otherAppUid);
     });
 
+    it('hands the value across apps — the grant it takes is a read grant', async () => {
+        await clearRows();
+        await grantRead(otherAppUid);
+        const sub = await subscribe(`kv:${otherAppUid}:cart:*`, ownAppToken, {
+            includeValue: true,
+        });
+        delivered.length = 0;
+
+        await kvSet(env.users.user.token, 'cart:items', [7], {
+            appUuid: otherAppUid,
+        });
+        await settle();
+
+        const own = delivered.find((one) => one.subId === sub.subId);
+        expect(own?.event).toMatchObject({ value: [7] });
+        await revokeRead(otherAppUid);
+    });
+
     it('settles a durable row when the grant is withdrawn', async () => {
         await clearRows();
         await grantRead(otherAppUid);

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

@@ -229,6 +229,18 @@ describe('the kv subject', () => {
             expect(subject().project({ ...kvDelivery, op }).op).toBe(op);
     });
 
+    it('carries a value only once dispatch has put one on the context', () => {
+        expect(subject().project(kvDelivery)).not.toHaveProperty('value');
+        expect(
+            subject().project({ ...kvDelivery, value: { items: [1] } }).value,
+        ).toEqual({ items: [1] });
+        // `null` is a value — what a deleted key now holds.
+        expect(subject().project({ ...kvDelivery, value: null })).toHaveProperty(
+            'value',
+            null,
+        );
+    });
+
     it('anchors on the exact key and on its prefixes', () => {
         expect(subject().tokens(kvDelivery)).toEqual([
             'k#user-uuid#app-1234#cart:items',

src/backend/services/events/registry.ts

@@ -74,6 +74,12 @@ export interface ProjectedFsEvent extends ProjectedEventBase {
 export interface ProjectedKvEvent extends ProjectedEventBase {
     op: KvOp;
     key: string;
+    /**
+     * What the key holds after the change — the written value, or `null` after
+     * a `del`. Only on a subscription that asked for it, and only when the
+     * value is small enough to inline; an `expire` never carries one.
+     */
+    value?: unknown;
 }
 
 /**
@@ -166,6 +172,8 @@ export interface KvEventContext extends EventContextBase {
     appUid: string;
     kvKey: string;
     op: KvOp;
+    /** The value after the change, once dispatch has decided it may ride. */
+    value?: unknown;
 }
 
 /**
@@ -319,6 +327,7 @@ const kvProject = (delivery: KvDeliveryContext): ProjectedKvEvent => ({
     subject: `kv:${delivery.appUid}:${delivery.kvKey}`,
     op: delivery.op,
     key: delivery.kvKey,
+    ...(delivery.value !== undefined ? { value: delivery.value } : {}),
     self: delivery.self,
     ts: delivery.ts,
     seq: delivery.seq,

src/backend/stores/events/DurableSubscriptionStore.integration.test.ts

@@ -136,6 +136,20 @@ describe('creating a subscription', () => {
         });
     });
 
+    it('remembers whether deliveries carry the value, off by default', async () => {
+        const { row: asking } = await durable().create(
+            input({ includeValue: true }),
+        );
+        const { row: silent } = await durable().create(input());
+
+        await expect(durable().getBySubId(asking.subId)).resolves.toMatchObject(
+            { includeValue: true },
+        );
+        await expect(
+            durable().getBySubId(silent.subId),
+        ).resolves.not.toHaveProperty('includeValue');
+    });
+
     it('names a session`s row for the account, not an app', async () => {
         const { row } = await durable().create(input());
         expect(row.subId.startsWith('user#')).toBe(true);

src/backend/stores/events/DurableSubscriptionStore.ts

@@ -99,6 +99,7 @@ export interface DurableSubscriptionInput {
     context: string | null;
     permission: SubscriptionPermission;
     expiresAt: number | null;
+    includeValue?: true;
     /**
      * Plan-resolved caps this subscribe is held to. Omitted falls back to the
      * structural maximum, so a writer that never resolved a plan still cannot
@@ -248,6 +249,7 @@ const toRow = (row: Record<string, unknown>): DurableSubscription => ({
         row.context === null || row.context === undefined
             ? null
             : String(row.context),
+    ...(Number(row.include_value) === 1 ? { includeValue: true } : {}),
     expiresAt: asNumber(row.expires_at),
     suspendedAt: asNumber(row.suspended_at),
     suspendedReason:
@@ -260,8 +262,8 @@ const toRow = (row: Record<string, unknown>): DurableSubscription => ({
 const SELECT_COLUMNS =
     '`id`, `sub_id`, `token`, `owner_user_id`, `holder_user_id`, `app_uid`, ' +
     '`subject`, `anchor_uid`, `anchor_path`, `match`, `delivery`, `ops`, ' +
-    '`handler_name`, `targets`, `context`, `permission`, `suspended_at`, ' +
-    '`suspended_reason`, `expires_at`, `created_at`';
+    '`handler_name`, `targets`, `context`, `permission`, `include_value`, ' +
+    '`suspended_at`, `suspended_reason`, `expires_at`, `created_at`';
 
 export class DurableSubscriptionStore extends PuterStore {
     // -- Writes ------------------------------------------------------
@@ -321,6 +323,7 @@ export class DurableSubscriptionStore extends PuterStore {
             targets,
             handlerName: input.handlerName,
             context: input.context,
+            ...(input.includeValue ? { includeValue: true } : {}),
             expiresAt: input.expiresAt,
             suspendedAt: null,
             suspendedReason: null,
@@ -344,6 +347,7 @@ export class DurableSubscriptionStore extends PuterStore {
             targets: JSON.stringify(row.targets),
             context: row.context,
             permission: row.permission,
+            include_value: row.includeValue ? 1 : 0,
             expires_at: row.expiresAt,
             created_at: row.createdAt,
         });

src/backend/stores/events/types.ts

@@ -106,6 +106,8 @@ export interface DispatchSubscription {
     permission: SubscriptionPermission;
     /** Transports this row's deliveries may take. */
     targets?: SubscriptionTarget[];
+    /** KV rows only: deliveries carry the key's new value where they can. */
+    includeValue?: true;
     /** Session rows only: the connection a delivery is addressed at. */
     socketId?: string;
     /** Durable rows only: set on every row that outlives its connection. */

src/backend/stores/systemKv/SystemKVStore.ts

@@ -771,9 +771,10 @@ export class SystemKVStore extends PuterStore {
         namespace: string,
         keys: string[],
         op: KvMutation,
+        values?: unknown[],
     ): Promise<void> {
         await this.#invalidate(namespace, keys);
-        this.#emitMutation(actor, namespace, keys, op);
+        this.#emitMutation(actor, namespace, keys, op, values);
     }
 
     /**
@@ -786,6 +787,7 @@ export class SystemKVStore extends PuterStore {
         namespace: string,
         keys: string[],
         op: KvMutation,
+        values?: unknown[],
     ): void {
         // Internal system data has no subscribable subject, and it is written
         // often enough that the emit itself would be the cost.
@@ -803,9 +805,20 @@ export class SystemKVStore extends PuterStore {
                 return;
             }
             if (keys.length === 0) return;
+            const unique = [...new Set(keys)];
             this.clients.event.emit(
                 'kv.mutated',
-                { namespace, userId, keys: [...new Set(keys)], op },
+                {
+                    namespace,
+                    userId,
+                    keys: unique,
+                    op,
+                    // Callers hand values aligned with `keys`; a deduped set
+                    // would misalign them, so only a dedupe-free batch carries.
+                    ...(values && unique.length === keys.length
+                        ? { values }
+                        : {}),
+                },
                 {},
             );
         } catch {
@@ -1215,7 +1228,7 @@ export class SystemKVStore extends PuterStore {
             ttl: expireAt,
             ...(disableSharing ? { [KV_PRIVATE_ATTR]: true } : {}),
         });
-        await this.#committed(actor, namespace, [key], 'set');
+        await this.#committed(actor, namespace, [key], 'set', [value]);
 
         return {
             res: true,
@@ -1286,7 +1299,13 @@ export class SystemKVStore extends PuterStore {
         }));
 
         const response = await this.clients.dynamo.batchPut(putParams);
-        await this.#committed(actor, namespace, [...byKey.keys()], 'set');
+        await this.#committed(
+            actor,
+            namespace,
+            [...byKey.keys()],
+            'set',
+            [...byKey.values()].map((item) => item.value),
+        );
         const units =
             response.ConsumedCapacity?.reduce(
                 (acc, curr) => acc + Number(curr.CapacityUnits ?? 0),
@@ -1311,7 +1330,7 @@ export class SystemKVStore extends PuterStore {
             namespace,
             key,
         });
-        await this.#committed(actor, namespace, [key], 'del');
+        await this.#committed(actor, namespace, [key], 'del', [null]);
         return {
             res: true,
             usage: addUsage(
@@ -1344,7 +1363,7 @@ export class SystemKVStore extends PuterStore {
             { namespace, key },
             { returnOld: true },
         );
-        await this.#committed(actor, namespace, [key], 'del');
+        await this.#committed(actor, namespace, [key], 'del', [null]);
 
         const old = response.Attributes as
             | { value?: unknown; ttl?: number }
@@ -1401,7 +1420,13 @@ export class SystemKVStore extends PuterStore {
                 key: { namespace, key },
             })),
         );
-        await this.#committed(actor, namespace, uniqueKeys, 'del');
+        await this.#committed(
+            actor,
+            namespace,
+            uniqueKeys,
+            'del',
+            uniqueKeys.map((): unknown => null),
+        );
         const units =
             response.ConsumedCapacity?.reduce(
                 (acc, curr) => acc + Number(curr.CapacityUnits ?? 0),
@@ -1837,7 +1862,9 @@ export class SystemKVStore extends PuterStore {
             );
             response = await runUpdate();
         }
-        await this.#committed(actor, namespace, [key], 'set');
+        await this.#committed(actor, namespace, [key], 'set', [
+            response.Attributes?.value,
+        ]);
 
         const usage = addUsage(
             probeUsage,
@@ -1911,7 +1938,9 @@ export class SystemKVStore extends PuterStore {
             valueAttributeValues,
             renderer.names,
         );
-        await this.#committed(actor, namespace, [key], 'set');
+        await this.#committed(actor, namespace, [key], 'set', [
+            response.Attributes?.value,
+        ]);
 
         const usage = addUsage(
             probeUsage,
@@ -1954,7 +1983,9 @@ export class SystemKVStore extends PuterStore {
                 undefined,
                 renderer.names,
             );
-            await this.#committed(actor, namespace, [key], 'set');
+            await this.#committed(actor, namespace, [key], 'set', [
+                response.Attributes?.value,
+            ]);
             return {
                 res: response.Attributes?.value,
                 usage: addUsage(
@@ -2051,7 +2082,9 @@ export class SystemKVStore extends PuterStore {
             renderer.names,
         );
 
-        await this.#committed(actor, namespace, [key], 'set');
+        await this.#committed(actor, namespace, [key], 'set', [
+            response.Attributes?.value,
+        ]);
 
         const usage = addUsage(
             probeUsage,

src/backend/stores/systemKv/kvMutationEvents.test.ts

@@ -100,6 +100,7 @@ describe('what a mutation announces', () => {
             userId: 42,
             keys: [KEY],
             op: 'set',
+            values: [1],
         });
         expect(mutations()[0].key).toBe('kv.mutated');
     });
@@ -116,7 +117,12 @@ describe('what a mutation announces', () => {
             opts,
         );
 
-        expect(onlyMutation()).toMatchObject({ keys: ['a', 'b'], op: 'set' });
+        // The last write to `a` is the one that landed, so it is the one told.
+        expect(onlyMutation()).toMatchObject({
+            keys: ['a', 'b'],
+            op: 'set',
+            values: [3, 2],
+        });
     });
 
     it('says nothing for a batch with nothing in it', async () => {
@@ -216,6 +222,48 @@ describe('every mutating method emits once', () => {
     });
 });
 
+describe('what a mutation carries', () => {
+    beforeEach(async () => {
+        await store.set({ key: KEY, value: { count: 1, list: [] } }, opts);
+        emitted = [];
+    });
+
+    it('carries the written value on a set', async () => {
+        await store.set({ key: KEY, value: { a: 1 } }, opts);
+        expect(onlyMutation().values).toEqual([{ a: 1 }]);
+    });
+
+    it('carries null for a deletion, whichever way it is made', async () => {
+        await store.del({ key: KEY }, opts);
+        expect(onlyMutation().values).toEqual([null]);
+
+        emitted = [];
+        await store.take({ key: KEY }, opts);
+        expect(onlyMutation().values).toEqual([null]);
+
+        emitted = [];
+        await store.batchDel({ keys: ['a', 'b'] }, opts);
+        expect(onlyMutation().values).toEqual([null, null]);
+    });
+
+    it('carries the whole item after a path mutation', async () => {
+        await store.incr({ key: KEY, pathAndAmountMap: { count: 2 } }, opts);
+        expect(onlyMutation().values).toEqual([{ count: 3, list: [] }]);
+
+        emitted = [];
+        await store.update(
+            { key: KEY, pathAndValueMap: { count: 7 } },
+            opts,
+        );
+        expect(onlyMutation().values).toEqual([{ count: 7, list: [] }]);
+    });
+
+    it('carries nothing on an expire, where the value did not change', async () => {
+        await store.expire({ key: KEY, ttl: 60 }, opts);
+        expect(onlyMutation()).not.toHaveProperty('values');
+    });
+});
+
 describe('flush', () => {
     it('marks the namespace rather than fanning out per key', async () => {
         await store.batchPut(

src/docs/src/Events.md

@@ -96,6 +96,14 @@ await puter.events.onLocal(`kv:${puter.appID}:orders:pending`, handler);
 
 The `subject` and [anchor](#anchor) on the subscription you get back are always fully qualified, whichever form you subscribed with.
 
+A delivery names the key and not what it now holds, so a handler that needs the value reads it back. Ask for it instead with `includeValue`, and every `set` carries the new value while every `del` carries `null`:
+
+```js
+await puter.events.onLocal('kv:cart', ({ event }) => render(event.value), { includeValue: true });
+```
+
+A value over **16 KB** serialized is not inlined: the event arrives without `value`, and you read the key as you would have anyway. An `expire` never carries one, since the value did not change. `includeValue` is accepted on `kv:` subjects only — anything else is refused with `invalid_include_value` — and not through a [share handle](#share-handle), whose grant is to watch a region rather than read it.
+
 Watching **another app's** key-value data takes the same consent as reading it: that app must not have opted out of data sharing, and the user must have granted your app `app-data:<appId>:kv:read`. It is checked when you subscribe and again on every delivery, so deliveries stop the moment either goes away. Where the feature is not enabled, a cross-app subject is refused with `events_cross_app_disabled`.
 
 ### Sharing a region with another user
@@ -127,7 +135,7 @@ const { handle } = await res.json();
 await puter.events.onLocal(`kv:${handle}:*`, ({ event }) => render(event.key));
 ```
 
-The handle is the whole of what the holder learns: not whose data it is, not where in the namespace it sits, and not anything above the prefix it was granted on. Events name it too: `subject` and `key` on every delivery are relative to the handle, in the same grammar the subscription was written in. `kv:<handle>:messages:*` narrows to part of the shared region, and one handle per channel gives one subscription covering every key written in that channel.
+The handle is the whole of what the holder learns: not whose data it is, not where in the namespace it sits, and not anything above the prefix it was granted on. Events name it too: `subject` and `key` on every delivery are relative to the handle, in the same grammar the subscription was written in. `kv:<handle>:messages:*` narrows to part of the shared region, and one handle per channel gives one subscription covering every key written in that channel. Values never cross a handle: `includeValue` on a handle subject is refused with `events_kv_handle_no_values`, because the grant behind it is to watch the region, not to read it.
 
 **Key layout is the access boundary.** A handle pins the prefix it was granted on, and nothing rewrites it afterwards: rename `workspace:<uuid>:` to `project:<uuid>:` and every handle already given out points at keys nothing writes any more. Grant on a **stable synthetic segment** — `workspace:<uuid>:`, `thread:<uuid>:` — rather than a semantic one like `acme-corp:` or `q3-planning:`, which is more likely to get renamed later.
 
@@ -182,11 +190,12 @@ A key-value change carries `key` where a filesystem change carries `uid` and `pa
 | `subject` | String | `kv:<appId>:<key>`, naming the key that changed. |
 | `op` | String | `set` for a write, `del` for a removal, `expire` when only the key's lifetime changed. |
 | `key` | String | The key that changed. |
+| `value` | Any | What the key holds after the change — the written value on a `set`, `null` on a `del`. Only on a subscription made with `includeValue`, only up to 16 KB serialized, and never on an `expire`. |
 | `self` | Boolean | As above. |
 | `ts` | Number | As above. |
 | `seq` | Number | As above. |
 
-Nothing else is included — in particular there is no field naming *who* made the change, because on a shared folder that would tell every subscriber who else is in there, and no field carrying the new **value**, so a subscription never becomes a way to read data the delivery check has not just re-authorized.
+Nothing else is included — in particular there is no field naming *who* made the change, because on a shared folder that would tell every subscriber who else is in there. The new **value** rides only where the subscription asked for it with `includeValue`, and only to the account whose namespace it is — never through a share handle — so a subscription never becomes a way to read data its holder could not read anyway.
 
 Emptying a whole store with [`puter.kv.flush()`](/KV/flush/) delivers nothing: no subject names "everything in this namespace went", and the keys a flush can enumerate are not reliably the keys it removed.
 

src/docs/src/Events/onLocal.md

@@ -32,6 +32,7 @@ Called with a single `{ event }` object per delivery. `event.op === 'gap'` means
 
 - `onError` (Function): Called with `{ message, code }` if the subscription lapses — the connection was lost and re-subscribing failed. The subscription is over at that point; call `onLocal()` again to resume. Without it, a lapse is reported on the console.
 - `timeout` (Number): How long to wait for the server to confirm the subscription, in milliseconds. Defaults to `30000`.
+- `includeValue` (Boolean): For a `kv:` subject, deliver the key's new value on every event as `event.value` — the written value on a `set`, `null` on a `del`, nothing on an `expire`. A value over 16 KB serialized is left out. Refused on a non-`kv:` subject and on a share handle.
 
 ## Return value
 
@@ -42,6 +43,7 @@ A `Promise` that resolves, once the server has confirmed the subscription, to a
 - `anchor` (Object): The subscription's [anchor](/Events/#anchor), as `{ uid, path }`. For a `kv:` subject, `uid` is the app whose store is being watched and `path` is the key prefix it is anchored at; for one made through a share handle, `uid` is the handle and `path` is empty. The path is the one the anchor had when you subscribed — a later rename or move does not update it.
 - `match` (String | null): The pattern events under the anchor are matched against, if the subject had one.
 - `op` (String | null): The single operation this subscription is limited to, or `null` for all of them.
+- `includeValue` (Boolean): Whether `kv:` deliveries on this subscription carry the key's new value.
 - `off` (Function): Ends the subscription — see [`subscription.off()`](/Events/off/).
 
 The promise rejects with `{ message, code }`:
@@ -54,6 +56,8 @@ The promise rejects with `{ message, code }`:
 | `invalid_subject_pattern` | The match pattern is past its bounds: 256 characters, 16 segments, one `*` per segment, one `**` in total. |
 | `invalid_kv_pattern` | A `kv:` subject has a `*` somewhere other than the end, or a `?`. |
 | `invalid_kv_handle_key` | A `kv:<handle>:…` subject names no key, or one that tries to leave the handle's granted region. |
+| `invalid_include_value` | `includeValue` is not a boolean, or was asked for on a subject that is not `kv:`. |
+| `events_kv_handle_no_values` | `includeValue` on a share-handle subject: a handle grants watching a region, not reading it. |
 | `events_cross_app_disabled` | The subject names another app's key-value data and that is not enabled here. |
 | `forbidden` | The target app does not share its data, or this app has not been granted `app-data:<appId>:kv:read` on it. |
 | `subject_does_not_exist` | The subject is not there, or this account cannot read it. |

src/docs/src/Events/onPersistent.md

@@ -28,6 +28,7 @@ puter.events.onPersistent(options)
 - `handler` (Function | String | Object): The handler source this subscription was written against. Sent as a **hash**, never as source: the subscription binds only if that hash matches what is published under `handlerName`, which is why `handlerName` is required alongside it. Accepts a function, a source string, or `{ file: '~/AppData/…/handler.js' }`.
 - `context` (Object): Values the handler needs, delivered to it as a frozen `ctx`. **Capped at 4 KB serialized** — see below.
 - `expiresAt` (Number | String): When the subscription ends by itself — unix seconds or an ISO-8601 string, and it has to be in the future.
+- `includeValue` (Boolean): For a `kv:` subject, deliver the key's new value on every event as `event.value` — the written value on a `set`, `null` on a `del`, nothing on an `expire`. A value over 16 KB serialized is left out. Refused on a non-`kv:` subject and on a share handle.
 
 ## Background delivery takes the user's consent
 
@@ -89,7 +90,7 @@ A `Promise` that resolves to the subscription:
 
 - `subId` (String): Its id, and what [`puter.events.unsubscribe()`](/Events/unsubscribe/) names. Stable for the life of the subscription.
 - `subject`, `anchor`, `match`, `op`: as `onLocal()` returns them.
-- `delivery` (String), `targets` (Array), `handlerName` (String | null).
+- `delivery` (String), `targets` (Array), `handlerName` (String | null), `includeValue` (Boolean).
 - `appUid` (String | null): The app that created it, or `null` for one an account session made.
 - `contextKeys` (Array | null), `contextHash` (String | null): the shape of the stored context, never its values.
 - `createdAt`, `expiresAt` (Number | null): unix seconds.
@@ -113,6 +114,8 @@ The promise rejects with `{ message, code }`:
 | `events_context_invalid` | `context` is not JSON-serializable. |
 | `invalid_targets` | A target outside `socket`/`worker`/`push`, `push` on a `single` subscription (which may not target it), or `worker` on a subscription with no app. |
 | `invalid_expires_at` | `expiresAt` is not a future time. |
+| `invalid_include_value` | `includeValue` is not a boolean, or was asked for on a subject that is not `kv:`. |
+| `events_kv_handle_no_values` | `includeValue` on a share-handle subject: a handle grants watching a region, not reading it. |
 | `subject_does_not_exist` | The subject is not there, or this account cannot read it. |
 | `events_subscription_limit` | This account already holds the maximum number of persistent subscriptions. |
 | `events_durable_requires_account` | Called from a temporary (anonymous) account, which gets session subscriptions only. |

src/docs/src/rate-limits-and-quotas.md

@@ -245,6 +245,7 @@ A temporary (anonymous) account cannot create durable subscriptions at all — `
 | Events per fetch page                        | 200          |
 | Matched subscriptions per event              | 50           |
 | Filter evaluations per event                 | 200          |
+| Key-value value inlined in a delivery        | 16 KB        |
 | Broadcast deliveries per minute, per subscription | 600     |
 | `single` deliveries per minute, per subscription | 120      |
 | Handler invocations per minute, per (account, app) | 60     |
@@ -291,7 +292,7 @@ Deleting the node a subscription is anchored on ends it too, unless the subject
 
 Match patterns are compiled once when you subscribe and are capped at **256 characters** and **16 segments**, with **one `*` per segment** and **one `**` per pattern**; anything past that is rejected with `invalid_subject_pattern`. `**` crosses directories and costs no more than `*`.
 
-A `kv:` subject is indexed on the first **6** `:`-segments, or **160 bytes**, of its key — whichever comes first; past that the remainder becomes a match pattern, which is subject to the caps above. A key-value subject matches its key exactly unless it ends in `*`, and a `*` anywhere else — or a `?` — is rejected with `invalid_kv_pattern`. Watching another app's key-value data is refused with `events_cross_app_disabled` where that is not enabled, and otherwise takes the same consent as reading it. The app slot names an app uid and is capped at **40 characters**; past that the subscription is refused with `events_value_too_large`.
+A `kv:` subject is indexed on the first **6** `:`-segments, or **160 bytes**, of its key — whichever comes first; past that the remainder becomes a match pattern, which is subject to the caps above. A key-value subject matches its key exactly unless it ends in `*`, and a `*` anywhere else — or a `?` — is rejected with `invalid_kv_pattern`. Watching another app's key-value data is refused with `events_cross_app_disabled` where that is not enabled, and otherwise takes the same consent as reading it. The app slot names an app uid and is capped at **40 characters**; past that the subscription is refused with `events_value_too_large`. A subscription made with `includeValue` is handed the key's new value on each delivery, up to **16 KB** serialized; a larger value is left out of the event and the subscriber reads the key back. Values never ride through a share handle, which grants watching a region and not reading it — `includeValue` on one is refused with `events_kv_handle_no_values`.
 
 **Deliveries are coalesced over 250 ms per subject.** A multipart upload, a save loop, or a recursive delete is one thing the user did, and it arrives as one event carrying the newest state rather than as one event per write. Two different files in the same window are two deliveries.
 

src/puter-js/src/modules/events/events.test.js

@@ -202,6 +202,36 @@ describe('delivery routing', () => {
     });
 });
 
+describe('includeValue', () => {
+    it('is sent with the subscribe and again on every re-subscribe', async () => {
+        const events = makeModule();
+        const sub = await subscribed(events, 'kv:cart', () => {}, { includeValue: true });
+        expect(sub.includeValue).toBe(true);
+        expect(sockets[0].sent.find(s => s.verb === 'events.subscribe').payload).toEqual({
+            subject: 'kv:cart',
+            includeValue: true,
+        });
+
+        sockets[0].fire('disconnect');
+        sockets[0].fire('connect');
+        sockets[0].answer('events.subscribe', okSub('sub-2', 'kv:app-1:cart'));
+        await Promise.resolve();
+
+        const resent = sockets[0].sent.filter(s => s.verb === 'events.subscribe');
+        expect(resent).toHaveLength(2);
+        expect(resent[1].payload).toEqual({ subject: 'kv:cart', includeValue: true });
+    });
+
+    it('is left off the wire unless asked for', async () => {
+        const events = makeModule();
+        const sub = await subscribed(events, 'kv:cart', () => {});
+        expect(sub.includeValue).toBe(false);
+        expect(sockets[0].sent.find(s => s.verb === 'events.subscribe').payload).toEqual({
+            subject: 'kv:cart',
+        });
+    });
+});
+
 describe('reconnect', () => {
     it('re-subscribes on reconnect and keeps the same handle', async () => {
         const events = makeModule();

src/puter-js/src/modules/events/lib/channel.js

@@ -22,6 +22,18 @@ import { EventSubscription } from './subscription.js';
 
 /** @typedef {{ ok: true, sub?: SubscriptionView }} VerbAck */
 
+/**
+ * The `subscribe` body for one handle, the same on first subscribe and on
+ * every re-subscribe.
+ *
+ * @param {EventSubscription} sub
+ * @returns {{ subject: string, includeValue?: true }}
+ */
+const subscribePayload = (sub) => ({
+    subject: sub.subject,
+    ...(sub.includeValue ? { includeValue: true } : {}),
+});
+
 // The wire, fixed by the server: three verbs answered with an ack, one channel
 // events arrive on.
 const SUBSCRIBE_VERB = 'events.subscribe';
@@ -131,7 +143,7 @@ export class EventChannel {
         const sub = new EventSubscription(this, subject, handler, options);
         this.inflight++;
         try {
-            const response = await this.request(SUBSCRIBE_VERB, { subject }, timeoutFor(sub));
+            const response = await this.request(SUBSCRIBE_VERB, subscribePayload(sub), timeoutFor(sub));
             sub.apply(viewOf(response));
             this.subscriptions.add(sub);
             this.byId.set(/** @type {string} */ (sub.subId), sub);
@@ -313,7 +325,7 @@ export class EventChannel {
         for ( const sub of [...this.subscriptions] ) {
             if ( sub.subId !== null || sub.pending ) continue;
             sub.pending = true;
-            this.request(SUBSCRIBE_VERB, { subject: sub.subject }, timeoutFor(sub))
+            this.request(SUBSCRIBE_VERB, subscribePayload(sub), timeoutFor(sub))
                 .then(response => {
                     sub.pending = false;
                     const view = viewOf(response);

src/puter-js/src/modules/events/lib/subscription.js

@@ -48,17 +48,25 @@ export class EventSubscription {
      */
     op = null;
 
+    /**
+     * Whether `kv:` deliveries on this subscription carry the key's new value.
+     *
+     * @type {boolean}
+     */
+    includeValue = false;
+
     /**
      * @internal
      * @param {import('./channel.js').EventChannel} channel
      * @param {string} subject
      * @param {EventHandler} handler
-     * @param {{ onError?: (error: Error & { code?: string }) => void, timeout?: number }} options
+     * @param {{ onError?: (error: Error & { code?: string }) => void, timeout?: number, includeValue?: boolean }} options
      */
     constructor (channel, subject, handler, options = {}) {
         /** @internal @type {import('./channel.js').EventChannel} */
         this.channel = channel;
         this.subject = subject;
+        this.includeValue = options.includeValue === true;
         /** @internal @type {EventHandler} */
         this.handler = handler;
         /** @internal @type {((error: Error & { code?: string }) => void) | undefined} */

src/puter-js/src/modules/events/onPersistent.js

@@ -62,6 +62,7 @@ export async function onPersistent (options = {}) {
         ...(options.expiresAt !== undefined && options.expiresAt !== null
             ? { expiresAt: options.expiresAt }
             : {}),
+        ...(options.includeValue ? { includeValue: true } : {}),
     };
 
     // Serialized only to check it against the cap before the round trip; the

src/puter-js/src/modules/events/persistent.test.js

@@ -74,13 +74,14 @@ describe('onPersistent', () => {
         expect(sub).toBe(view);
     });
 
-    it('carries delivery, targets, handlerName and expiry when given', async () => {
+    it('carries delivery, targets, handlerName, expiry and includeValue when given', async () => {
         await makeModule().onPersistent({
             subject: SUBJECT,
             delivery: 'single',
             targets: ['worker'],
             handlerName: 'ingestUpload',
             expiresAt: 4102444800,
+            includeValue: true,
         });
 
         expect(bodyOf()).toEqual({
@@ -89,6 +90,7 @@ describe('onPersistent', () => {
             targets: ['worker'],
             handlerName: 'ingestUpload',
             expiresAt: 4102444800,
+            includeValue: true,
         });
     });
 

src/puter-js/src/modules/events/types.js

@@ -39,7 +39,7 @@
 /**
  * One key-value change. It carries `key` where a filesystem event carries `uid`
  * and `path` — a KV change happens to a key in a store, and there is no node to
- * name — and never the new value, so a delivery cannot become a read.
+ * name — and the new value only where the subscription asked for it.
  *
  * @typedef {Object} PuterKvEvent
  * @property {string} id Unique id for this event.
@@ -47,6 +47,10 @@
  * @property {'set' | 'del' | 'expire'} op `set` for a write, `del` for a
  *   removal, `expire` when only the key's lifetime changed.
  * @property {string} key The key the event is about.
+ * @property {unknown} [value] What the key holds after the change — the
+ *   written value on a `set`, `null` on a `del`. Present only on a
+ *   subscription made with `includeValue`, and left out when the value is
+ *   over 16 KB serialized; an `expire` never carries one.
  * @property {boolean} self `true` when the change was made by the account
  *   holding the subscription.
  * @property {number} ts Milliseconds since the epoch.
@@ -162,6 +166,8 @@
  *   on the console.
  * @property {number} [timeout] How long to wait for the server to answer
  *   `subscribe`, in milliseconds. Default `30000`.
+ * @property {boolean} [includeValue] `kv:` subjects only: deliver the key's new
+ *   value on each event as `event.value`. Refused on a share handle.
  */
 
 /**
@@ -188,6 +194,8 @@
  * @property {number | string} [expiresAt] When the subscription ends by
  *   itself — unix seconds or an ISO-8601 string, and it has to be in the
  *   future.
+ * @property {boolean} [includeValue] `kv:` subjects only: deliver the key's new
+ *   value on each event as `event.value`. Refused on a share handle.
  */
 
 /**
@@ -210,6 +218,8 @@
  * @property {Array<'socket' | 'worker' | 'push'>} targets Transports its
  *   deliveries may take.
  * @property {'broadcast' | 'single'} delivery Its delivery class.
+ * @property {boolean} includeValue Whether its `kv:` deliveries carry the
+ *   key's new value.
  * @property {string | null} handlerName The handler it is bound to.
  * @property {string | null} appUid The app that created it, or `null` for one
  *   an account session made.