fix!: keep authorized snapshot live queries subscribed

README.md

@@ -4,6 +4,10 @@
 
 ## How it works
 
+Live queries stream authorized results even when row-level permissions prevent
+safe resume cursors. [Snapshot and resumable delivery](docs/live-query-delivery.md)
+describe the two modes and their consistency guarantees.
+
 Write the domain once, compose it into one `Service` or several, then
 generate the client. Each stage below uses real code from
 [`tests/e2e-ui`](tests/e2e-ui).

docs/gateway/live-sharing.md

@@ -19,12 +19,16 @@ upstream subscription and producer. HTTP control requests do not execute result
 Groups bind the exact document, variables, origin subject/cache scope, policy,
 schema and protocol. Each consumer keeps its own transport ID, expiry, queue and
 freshness requirements. Different resume cursors start independent replay. Handoff
-requires the same operation plus an exact supported, comparable cursor vector and
+requires the same operation plus an exact resumable, comparable cursor vector and
 matching data; the consumer's replay frames remain queued before future shared
 frames. Unknown cursors keep independent streams. Equal data alone never proves
 cursor equality. Duplicate suppression hashes the whole data and protocol envelope,
 so new confirmation evidence is delivered even when values are unchanged.
 
+Cursorless `live.mode = "snapshot"` responses cannot prove shared replay handoff;
+they retain independent authorized subscriptions and reconnect with fresh queries.
+See [live query delivery](../live-query-delivery.md) for the wire contract.
+
 Defaults bound a coordinator to 256 groups, 1,024 consumers per group, 16 pending
 frames per consumer, 1 MiB per full frame, eight retained history frames and a
 one-hour group lifetime. Native ingress bounds socket/request counts and wire

docs/live-query-delivery.md

@@ -0,0 +1,28 @@
+# Live queries and authorization
+
+`@live` keeps an authorized query current. Resuming missed projection changes is
+a separate capability: a partition-wide cursor can reveal activity outside a
+row-filtered result, even when the rows themselves are protected.
+
+Every live response declares `extensions.distributed.live.mode`:
+
+- `snapshot`: a fresh authorized replacement result, with `reset: true` and
+  `cursors: []`. It carries no comparable index vector or projection observations.
+  The client keeps listening and reconnects with a fresh query, not a resume token.
+- `resumable`: a result with matching, nonempty index and cursor vectors. Existing
+  resume validation, reset, replay and causal reconciliation rules apply.
+
+Snapshot delivery does not relax read permissions or invent causal evidence.
+Changes affecting only denied rows must not produce activity frames. A row
+leaving the authorized result disappears from that operation's membership;
+absence is not a globally authoritative deletion or tombstone.
+
+Within one current subscription, snapshot frames are ordered. The client fences
+older HTTP requests when a live result takes ownership, and fences callbacks
+from disposed subscriptions and previous authorization generations. Local cache
+membership revisions are not server projection positions and cannot confirm
+optimistic commands.
+
+This is a breaking v5 protocol change: `mode` replaces the ambiguous `supported`
+boolean. Upgrade server and generated-client runtime together. Old or unknown
+wire forms fail closed; applications do not need a polling or reload workaround.

js/README.md

@@ -51,6 +51,12 @@ to validate that committed artifacts are current without rewriting them.
 
 A co-located route document can opt into SSR and live continuation:
 
+Row-filtered `@live` queries remain subscribed using authorized replacement
+snapshots; cursor-capable queries retain resumable delivery. Neither requires
+application polling. The v5 wire protocol uses `live.mode` (`snapshot` or
+`resumable`), replacing `live.supported`; upgrade client and server together.
+See [live query delivery](../docs/live-query-delivery.md).
+
 ```graphql
 query Todos @load @live {
   todos(order_by: [{ status: asc }, { todo_id: asc }]) {

js/src/protocol.ts

@@ -149,10 +149,10 @@ export type DistributedQuerySnapshot = Readonly<
 	}
 >;
 
-/** Per-frame decision about live support, reset, and resumable cursors. */
+/** Per-frame delivery mode, reset decision, and resumable cursors. */
 export type DistributedLiveMetadata = Readonly<
 	Record<string, unknown> & {
-		supported: boolean;
+		mode: 'snapshot' | 'resumable';
 		reset: boolean;
 		cursors: readonly DistributedLiveCursor[];
 	}
@@ -596,7 +596,8 @@ function parseSnapshot(value: unknown): DistributedQuerySnapshot {
 function parseLive(value: unknown): DistributedLiveMetadata {
 	const path = 'extensions.distributed.live';
 	const live = record(value, path);
-	if (typeof live.supported !== 'boolean') invalid(`${path}.supported`);
+	if (live.mode !== 'snapshot' && live.mode !== 'resumable') invalid(`${path}.mode`);
+	if ('supported' in live) invalid(`${path}.supported`);
 	if (typeof live.reset !== 'boolean') invalid(`${path}.reset`);
 	if (
 		!Array.isArray(live.cursors) ||
@@ -613,12 +614,12 @@ function parseLive(value: unknown): DistributedLiveMetadata {
 		cursors.map((cursor) => cursor.projection),
 		`${path}.cursors`
 	);
-	if (!live.supported && (!live.reset || cursors.length !== 0)) {
+	if (live.mode === 'snapshot' && (!live.reset || cursors.length !== 0)) {
 		invalid(path);
 	}
 	return Object.freeze({
 		...live,
-		supported: live.supported,
+		mode: live.mode,
 		reset: live.reset,
 		cursors
 	}) as DistributedLiveMetadata;
@@ -630,7 +631,10 @@ function validateLiveSnapshot(
 ): void {
 	if (live === undefined) return;
 	if (snapshot === undefined) invalid('extensions.distributed.snapshot');
-	if (!live.supported) return;
+	if (live.mode === 'snapshot') {
+		if (snapshot.indexesComparable) invalid('extensions.distributed.snapshot.indexesComparable');
+		return;
+	}
 	if (!snapshot.indexesComparable) {
 		invalid('extensions.distributed.snapshot.indexesComparable');
 	}

js/src/replica/distributed-replica/clocks.ts

@@ -122,7 +122,7 @@ export function latestCursors(
 	live: DistributedProtocolEnvelope['live']
 ): readonly DistributedLiveCursor[] {
 	if (live !== undefined) {
-		return live.supported ? live.cursors : Object.freeze([]);
+		return live.mode === 'resumable' ? live.cursors : Object.freeze([]);
 	}
 	return Object.freeze(
 		snapshot.indexes.flatMap((index) =>

js/src/replica/distributed-replica/impl-fetch-live.ts

@@ -1,7 +1,5 @@
-import { CacheRevisionConflictError } from '../../internal/cache-engine.js';
 import type { GraphqlVariables } from '../../types.js';
 import {
-	parseGraphqlResponseExtensions,
 	type DistributedLiveCursor,
 	type DistributedProtocolEnvelope
 } from '../../protocol.js';
@@ -309,40 +307,19 @@ export function retainLive<TData, TVariables extends GraphqlVariables>(
 						}
 						return;
 					}
-					let unsupportedLive = false;
 					try {
-						unsupportedLive =
-							parseGraphqlResponseExtensions(result.extensions)
-								?.distributed?.live?.supported === false;
-						if (unsupportedLive) state.live = 'off';
-						const distributed = host.writeCanonicalResult(
+						host.writeCanonicalResult(
 							watch.artifact,
 							watch.variables,
 							result,
 							'live',
 							undefined,
 							projectionGeneration
 						);
-						if (distributed.live?.supported === false) {
-							fallbackFromLive(host, watch, entry);
-							return;
-						}
 						state.live = 'active';
 						entry.operationGeneration =
 							host.operationGeneration(watch.key);
 					} catch (error) {
-						if (
-							unsupportedLive &&
-							error instanceof CacheRevisionConflictError
-						) {
-							/*
-							 * Revision zero is shared by provisional fallbacks.
-							 * Another operation may already have filled the same
-							 * semantic index differently; HTTP remains authoritative.
-							 */
-							fallbackFromLive(host, watch, entry);
-							return;
-						}
 						state.live = 'error';
 						state.errors = stableErrors(state.errors, [graphqlError(error)]);
 						host.emitState(watch.key, false);
@@ -429,14 +406,13 @@ export function fallbackFromLive<TData, TVariables extends GraphqlVariables>(
 	/*
 	 * Keep the inactive entry as an authorization-generation-scoped
 	 * sentinel. Query ingestion calls resumeLiveWatches(); deleting this
-	 * entry would otherwise reopen an unsupported or completed stream
+	 * entry would otherwise reopen a completed stream
 	 * immediately. Authorization invalidation clears it and may retry.
 	 *
-	 * A supported live frame advances the operation generation. Any HTTP
+	 * An accepted live frame advances the operation generation. Any HTTP
 	 * request that was already running is therefore doomed by its response
-	 * fence; drain it before starting the authoritative fallback. A first
-	 * unsupported frame never advances the generation, so its overlapping
-	 * HTTP request remains valid and can be reused directly.
+	 * fence; drain it before starting the authoritative fallback. A stream
+	 * that completed before its first frame can reuse the in-flight query.
 	 */
 	if (supersededFlight === undefined) {
 		refresh();

js/src/replica/distributed-replica/impl.ts

@@ -1184,21 +1184,9 @@ export class DistributedReplicaImpl implements DistributedReplicaApi {
 		}
 
 		this.#validateLiveSnapshot(snapshot, live);
-		const unsupportedLive =
-			source === 'live' && live?.supported === false;
+		const snapshotLive = source === 'live' && live?.mode === 'snapshot';
 		const reset = live?.reset === true;
 		const group = this.#operationProtocols.get(key)!;
-		/*
-		 * An unsupported subscription response is an authorized fallback
-		 * snapshot, not a live source. In particular, a row-filtered snapshot
-		 * has no comparable index vector, so retaining live ownership here
-		 * would reject every later query handoff. Relinquish any prior live
-		 * ownership without advancing the generation; the forced HTTP fallback
-		 * starts against the generation that remains after this frame.
-		 */
-		if (unsupportedLive && group.active === 'live') {
-			group.active = undefined;
-		}
 		const previousActiveSource = group.active;
 		const handoff =
 			previousActiveSource !== undefined &&
@@ -1226,8 +1214,22 @@ export class DistributedReplicaImpl implements DistributedReplicaApi {
 			requestRevision,
 			source
 		);
+		// A non-resumable stream has no causal vector to compare. An HTTP
+		// refresh may replace it only if the request began after its last
+		// accepted membership. fetchWatch also fences intervening live frames.
+		// Taking query ownership restarts the stream, fencing queued callbacks.
+		const snapshotRefresh =
+			source === 'network' &&
+			previousActiveSource === 'live' &&
+			!snapshot.indexesComparable &&
+			activeState?.snapshotScope === undefined &&
+			activeState?.indexRevision !== undefined &&
+			requestRevision !== undefined &&
+			compareCanonicalDecimalStrings(requestRevision, activeState.indexRevision) > 0;
 		const handoffBlocked =
 			handoff &&
+			!snapshotLive &&
+			!snapshotRefresh &&
 			(
 				!snapshot.indexesComparable ||
 				!isComparableHandoffDisposition(ownDisposition) ||
@@ -1251,7 +1253,6 @@ export class DistributedReplicaImpl implements DistributedReplicaApi {
 					: sharedDisposition.disposition ?? disposition;
 		}
 		const sourceSwitched =
-			!unsupportedLive &&
 			!handoffBlocked &&
 			isComparableHandoffDisposition(disposition) &&
 			this.#activateOperationSource(
@@ -1316,24 +1317,16 @@ export class DistributedReplicaImpl implements DistributedReplicaApi {
 			!rejectedHandoff &&
 			disposition !== 'lower' &&
 			disposition !== 'incomparable';
-		/*
-		 * Revision zero is the cache engine's lowest legal checkpoint. It lets
-		 * an unsupported live response fill an empty cache immediately while
-		 * guaranteeing that any HTTP request revision can replace it.
-		 */
+		// Snapshot frames receive local membership revisions, not fabricated
+		// projection clocks. Live ownership fences older overlapping HTTP work.
 		const indexRevision =
-			unsupportedLive
-				? '0'
+			writeIndexes &&
+			(sourceSwitched || sharedDisposition.disposition === 'higher')
+				? this.#allocateIndexRevision()
 				: writeIndexes &&
-					  (
-							sourceSwitched ||
-							sharedDisposition.disposition === 'higher'
-						)
-					? this.#allocateIndexRevision()
-					: writeIndexes &&
-						  sharedDisposition.disposition === 'equal' &&
-						  sharedDisposition.indexRevision !== undefined
-						? sharedDisposition.indexRevision
+					  sharedDisposition.disposition === 'equal' &&
+					  sharedDisposition.indexRevision !== undefined
+					? sharedDisposition.indexRevision
 				: snapshot.indexesComparable &&
 					  disposition === 'equal' &&
 					  operationState.indexRevision !== undefined
@@ -1615,7 +1608,7 @@ export class DistributedReplicaImpl implements DistributedReplicaApi {
 		} else if (live?.reset === true || !snapshot.indexesComparable) {
 			operationState.cursors = Object.freeze([]);
 		}
-		if (source === 'live' && !unsupportedLive) {
+		if (source === 'live') {
 			this.#advanceOperationGeneration(key);
 		}
 			if (source !== 'live' && sourceSwitched) {
@@ -2838,7 +2831,7 @@ export class DistributedReplicaImpl implements DistributedReplicaApi {
 		snapshot: DistributedQuerySnapshot,
 		live: DistributedProtocolEnvelope['live']
 	): void {
-		if (live === undefined || !live.supported) return;
+		if (live === undefined || live.mode === 'snapshot') return;
 		if (!snapshot.indexesComparable) {
 			protocolInvalid(
 				'extensions.distributed.snapshot.indexesComparable'

js/tests/fixtures/adapter-conformance.mjs

@@ -302,6 +302,7 @@ export function todoFrame(
 		authorizationGeneration = 'auth-1',
 		position = '1',
 		source = 'query',
+		mode = 'resumable',
 		reset = false,
 		errors
 	} = {}
@@ -343,9 +344,9 @@ export function todoFrame(
 				snapshot: {
 					scopeToken: `snapshot:${root.field}`,
 					recordsComplete: true,
-					indexesComparable: true,
+					indexesComparable: mode === 'resumable',
 					records,
-					indexes: [
+					indexes: mode === 'snapshot' ? [] : [
 						{
 							projection: 'todos-projector',
 							scopeToken: `index:${root.field}`,
@@ -358,9 +359,9 @@ export function todoFrame(
 				...(source === 'live'
 					? {
 							live: {
-								supported: true,
-								reset,
-								cursors: [resume]
+								mode,
+								reset: mode === 'snapshot' || reset,
+								cursors: mode === 'snapshot' ? [] : [resume]
 							}
 						}
 					: {})

js/tests/fixtures/unique-key-artifact.mjs

@@ -26,7 +26,7 @@ export function frame(position, targetId, targetTitle) {
 		protocolVersion: artifact.protocol.version, schemaHash: artifact.protocol.schemaHash,
 		authorizationGeneration: 'auth-1', cacheScope: 'unique-key-cache',
 		operation: position === '1' ? artifact.id : artifact.live.id,
-		...(position === '1' ? {} : { live: { supported: true, reset: false, cursors: [
+		...(position === '1' ? {} : { live: { mode: "resumable", reset: false, cursors: [
 			{ projection: 'unique-key-projector', position, token: `resume:${position}` }
 		] } }),
 		snapshot: { scopeToken: 'unique-key-snapshot', recordsComplete: true, indexesComparable: true,