diff --git a/CHANGELOG.md b/CHANGELOG.md
index 9b1ab641..cbcf5c81 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -31,6 +31,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Security
 
+- **Live SSE streams now apply a role's row-`filter` per subscriber, closing a query/stream row-level-security drift** (`internal/stream/hub.go`, `internal/stream/subscriber.go`, `internal/policy/policy.go`, `internal/policy/rowfilter.go` (new), `internal/discovery/validation.go`, `internal/api/stream.go`, `cmd/wavehouse/main.go`, `docs/src/content/docs/access-control.mdx`, `docs/src/content/docs/architecture.md`, plus tests in `internal/policy/rowfilter_test.go` (new), `internal/discovery/validation_test.go`, `internal/stream/hub_test.go`): closes #319. The SSE delivery path stripped denied columns but never applied a role's row-level `filter` predicate, so a subscriber received rows the structured-query path would have filtered out for that same role — a data-exposure on the streaming surface for any table that combines a row-policy with a shared or role-scoped stream (harmless on the public Stats table today, which carries no restrictive row-policy, but real for any private/PII table fronted by a stream). The row-filter is now resolved once into predicates that feed **both** read surfaces — the query path renders them to SQL, the stream evaluates them in memory (`ResolvedPermissions.RowVisible`, evaluated per subscriber against that subscriber's claims via the same `Evaluate` call the query path uses) — so the two can't drift (the row-level analogue of the shared `IsColumnAllowed` decision from #223). Because a row-filter resolves against each subscriber's JWT claims, the #294/#353 once-per-role projection is now claims-aware: a role **without** a filter keeps the pure once-per-role fast path unchanged (zero regression on the public stream), while a role **with** a filter keeps the shared column projection but delivers it only to the subscribers whose claims admit each row (evaluated against the full event, so a filter may key on a column the role can't select). Equality and set predicates (`_eq`/`_neq`/`_in` — the tenant/user scoping row-security actually uses) are enforced exactly — numeric ties beyond `float64` precision resolve at arbitrary precision (`math/big`), so the string-encoded 64-bit IDs ingest accepts to survive JS precision loss never falsely collide, and a `NaN` operand fails closed rather than comparing equal to everything — with one representation caveat: a `Bool`/`DateTime` filter value compares against the event's canonical text form, without ClickHouse's cross-representation coercion (`true`, not `1`); ordering predicates (`_gt`/`_lt`) are schema-informed — numeric columns compare numerically (matching ClickHouse), other types lexicographically — which is best-effort for exotic types stored as numbers (a documented boundary in access-control.mdx), with every ambiguous or uncomparable value failing closed (the row is withheld). Replay (gap-fill) applies the same per-connection row check. The resource limits (`max_rows`, `max_execution_time`, …) remain a query-path property and are still **not** applied to the stream (a separate, documented boundary). Supersedes the "stream path applies no row-level filter" invariant noted in the #294/#353 Changed entry below.
 - **Policy `_in` is now enforced on both the row-`filter` and insert-`check` paths, closing a fail-open row-security gap** (`internal/policy/policy.go`, `internal/api/ingest.go`, `docs/src/content/docs/access-control.mdx`, plus tests in `internal/policy/policy_test.go`, `internal/api/ingest_test.go`): closes #224. The `Filter` schema accepted `_in` but the engine never read it: on the row-`filter`/SELECT path `resolveFilters` had no `_in` branch, so a row-security filter like `tenant_id: { _in: … }` produced **no `WHERE` predicate** and the role saw every row instead of its tenant subset (a fail-open, same family as #223); on the `check`/INSERT path only `_eq` was honored, silently dropping any other operator. `_in` now takes a single claim that resolves to a JSON **array** (the multi-tenant case — a token's `tenant_ids` list) and emits `col IN (?, …)` with one bound param per element; a scalar claim is a one-element set, and an empty/absent claim matches **no rows** (fail-closed) rather than widening to all of them. On the insert path an `_in` check requires the column be present and one of the set — there is no single value to auto-inject as `_eq` does, so an omitted column is rejected (`403 check failed`). The comparison operators are enforced on `filter` (`_eq`/`_neq`/`_gt`/`_lt`/`_in` all produce predicates now, so nothing is rejected there) and, on `check`, `_neq`/`_gt`/`_lt` become a loud config-load rejection (no insert-time semantics; `check` honors `_eq` + `_in`). The `_in` value stays a single templated string in the wire schema (Go `Filter.In`, SDK `PolicyFilter._in`), matching the established "set = array" shape of the caller-query `in` operator.
 - **`denied_aggregations` is now enforced case-insensitively against the caller-supplied function name, closing a policy bypass** (`internal/policy/policy.go`, plus tests in `internal/policy/policy_test.go`): closes #318. `IsAggregationAllowed` lower-cased the aggregation name only *after* the deny-list loop and the empty-allow-list early return, so the deny check compared a lower-cased deny entry against the raw caller input — a denied aggregation slipped past simply by changing case (`SUM` bypassed a `sum` deny entry, and with an empty allow list the call was then permitted). `isValidAggFn` already accepts any casing and the SQL builder emits the function name verbatim, so the denied aggregation actually executed. The case fold now happens once, before the deny check, so deny wins regardless of caller casing — matching the case-insensitive contract the access-control docs already specified.
 - **A role's structured-query resource caps are now enforced server-side by ClickHouse, so a public read can't outrun its budget during a server-side scan/merge/aggregation phase** (`internal/api/ch_settings.go` (new), `internal/api/structured_query.go`, `internal/policy/policy.go`, `internal/policy/scalars.go` (new), `internal/config/config.go`, `internal/query/builder.go`, `cmd/wavehouse/main.go`, `config.yaml`, `clients/ts/src/types.ts`, `docs/src/content/docs/{access-control.mdx,configuration.mdx}`, plus tests in `internal/api/ch_settings_test.go`, `internal/policy/{policy,scalars}_test.go`, `internal/config/config_test.go`, `internal/query/builder_test.go`, `tests/integration/query_limits_test.go` (all new/expanded), `tests/e2e/sdk/query.test.ts`): closes #316. The native ClickHouse connection passed **no per-query `Settings`**, so a read's policy caps bound it only client-side — a Go `context` deadline (which cancels only while the client is reading result blocks) plus a SQL `LIMIT`. Memory and rows scanned were **never bounded server-side**, so a heavy aggregation could allocate gigabytes of state or scan an entire table well within the time budget; and because `clickhouse-go` derives a `max_execution_time` setting from the context deadline only for deadlines `> 1s`, a sub-second time cap reached ClickHouse with no server-side time bound at all. The structured-query path now attaches per-query `Settings` derived from the role's resolved permissions: `max_execution_time` (fractional seconds, emitted explicitly so the sub-second case is enforced), `max_result_rows` + `result_overflow_mode=throw` (defense-in-depth behind the SQL `LIMIT`), `max_rows_to_read` + `read_overflow_mode=throw`, and `max_memory_usage` — so a query that exceeds its budget is rejected by the server (ClickHouse codes 158 / 241) rather than running to completion. **Boundary:** WaveHouse owns the *dynamic, per-role* caps (sent as per-query settings); the *global, static* backstop — which applies to every query including named pipes and raw admin SQL — is configured in **ClickHouse's own settings profiles and quotas** (documented in `configuration.mdx`), composes with the per-role caps, and holds even against a WaveHouse bug. **Schema (pre-launch):** the per-role policy fields are human-readable in / numeric out — `max_execution_time_ms` (int) → **`max_execution_time`** (set as a duration string `"5s"` or a bare ms number; read back as ms), the new **`max_memory_usage`** (set as a size string `"4GiB"` — IEC/SI respected via `dustin/go-humanize`, so `4GB` ≠ `4GiB` — or a bare byte number; read back as bytes), and the new **`max_rows_to_read`** (int), backed by two small `Millis`/`ByteSize` types in the `policy` package. The formerly hard-coded `query.DefaultMaxRows = 10000` result-LIMIT becomes the documented, tunable `query.default_max_rows` config knob (`Build` takes it as a parameter). Raw admin SQL remains unbounded by WaveHouse (governed by ClickHouse). Verified RED before the fix: the handler-level integration test confirmed a capped read returned the full result set when the settings weren't sent.
diff --git a/cmd/wavehouse/main.go b/cmd/wavehouse/main.go
index c75eb853..7ae3e21a 100644
--- a/cmd/wavehouse/main.go
+++ b/cmd/wavehouse/main.go
@@ -292,7 +292,7 @@ func run() int {
 	// handler (write counts), and the Hub that projects/serializes each event once
 	// per (topic, role) and pushes it to that role's subscribers.
 	sseMetrics := stream.NewMetrics()
-	streamHub := stream.NewHub(policyStore, sseMetrics)
+	streamHub := stream.NewHub(policyStore, registry, sseMetrics)
 
 	// Start policy watch for cluster-wide updates.
 	go policyStore.Watch(ctx)
diff --git a/docs/src/content/docs/access-control.mdx b/docs/src/content/docs/access-control.mdx
index dda1e0b5..3804b0af 100644
--- a/docs/src/content/docs/access-control.mdx
+++ b/docs/src/content/docs/access-control.mdx
@@ -370,12 +370,12 @@ The same policy drives every data path, but not every field is meaningful on eve
 | ------- | -------- | -------- |
 | Structured read | `POST /v1/query?table={table}` | table+role `select` required, then `allow`/`deny_columns`, row `filter`, aggregation rules, and the per-role `max_rows` / `max_execution_time` / `max_rows_to_read` / `max_memory_usage` caps (over the [ClickHouse server-wide limits](/configuration#server-side-resource-limits)) |
 | Ingest (write) | `POST /v1/ingest?table={table}` | table+role `insert` required, then `allow`/`deny_columns` and `check` (enforced and auto-injected) |
-| Live stream | `GET /v1/stream` | table+role `select` required (a table the role can't read is skipped), then denied columns are masked from each event |
+| Live stream | `GET /v1/stream` | table+role `select` required (a table the role can't read is skipped), then denied columns are masked from each event and the role's row `filter` is applied per subscriber against their JWT claims (see caution below) |
 | Raw SQL | `POST /v1/admin/query` | `admin_role` only — no per-statement policy; the role gate is the entire authorization story |
 | Named pipe | `GET/POST /v1/pipes/{name}` | per-pipe `allowed_roles` (not the policy engine; see [Named Pipes](/pipes)). Resource limits come from ClickHouse's [server-wide settings](/configuration#server-side-resource-limits), not per-role policy caps |
 
-:::caution[Live streams enforce access, not row filters]
-SSE subscribers are checked for table-level `select` permission and have denied columns stripped from each event, but the row-level `filter` predicates and the resource limits (`max_rows`, `max_execution_time`, `max_rows_to_read`, `max_memory_usage`) are a property of the SQL query path and are **not** applied to the live event stream. If a role must never observe another tenant's rows in real time, don't grant it stream access to a shared table — scope the data at the table level.
+:::caution[Live streams enforce column and row policy, but not resource limits]
+SSE subscribers are checked for table-level `select` permission, have denied columns stripped from each event, and — like the query path — receive only the rows their role's row-level `filter` predicates admit, evaluated per subscriber against their JWT claims. Equality and set predicates (`_eq`, `_neq`, `_in`) — the usual tenant/user scoping — are enforced exactly: numeric ties resolve at full precision (an ID beyond `float64`'s 2^53 never falsely matches a neighbor) and an unparseable or `NaN` operand withholds the row. One representation caveat: a `Bool` (or `DateTime`) value compares by its canonical text form, without ClickHouse's cross-representation coercion — write the filter value the way your events carry it (`true`, not `1`). Ordering predicates (`_gt`, `_lt`) are best-effort: the stream compares numeric columns numerically (matching ClickHouse) but lacks ClickHouse's full per-type coercion, so an ordering filter on an exotic type (a `Decimal`/`Int128` beyond `float64` precision, or a `DateTime` ingested as a Unix number) can admit or withhold a row the query path wouldn't. The resource limits (`max_rows`, `max_execution_time`, `max_rows_to_read`, `max_memory_usage`) remain a property of the SQL query path and are **not** applied to the live event stream; if those caps are part of a role's isolation story, don't rely on them over the stream.
 :::
 
 ## Managing the policy
diff --git a/docs/src/content/docs/architecture.md b/docs/src/content/docs/architecture.md
index 4c6a8958..f35d5e80 100644
--- a/docs/src/content/docs/architecture.md
+++ b/docs/src/content/docs/architecture.md
@@ -85,7 +85,7 @@ The API layer uses [Chi](https://github.com/go-chi/chi) for routing with Request
 
 The SSE fan-out, factored out of `api/` so the delivery hot path ([#294](https://github.com/Wave-RF/WaveHouse/issues/294)) lives next to the keepalive primitives it shares. One abstraction per file.
 
-- **hub.go** — `Hub`, the event fan-out. Subscribers register under `(topic, role)`; `Broadcast` decodes each event once, applies each subscribed role's column policy once, builds one SSE frame per role, and fans it to every member of that role's `Bucket` — collapsing the prior per-subscriber `unmarshal → evaluate → filter → marshal` into one pass per distinct `(role, table)` output shape (the [#294](https://github.com/Wave-RF/WaveHouse/issues/294) lever; the measured ceiling was ~2 270 deliveries/s from re-projecting per subscriber). The `(topic, role)` key is sufficient because column visibility derives only from the role+table policy entry, never from JWT claims (claims feed only the row-level `WHERE`/`CHECK`, which the stream path does not apply). `ReplayFrame` shares the same projection for the handler's per-connection gap-fill.
+- **hub.go** — `Hub`, the event fan-out. Subscribers register under `(topic, role)`; `Broadcast` decodes each event once, applies each subscribed role's column policy once, builds one SSE frame per role, and fans it to every member of that role's `Bucket` — collapsing the prior per-subscriber `unmarshal → evaluate → filter → marshal` into one pass per distinct `(role, table)` output shape (the [#294](https://github.com/Wave-RF/WaveHouse/issues/294) lever; the measured ceiling was ~2 270 deliveries/s from re-projecting per subscriber). The column projection is claims-independent, so it is shared across a role's whole bucket; the role's row-level `filter` predicate is not — it is resolved against each subscriber's JWT claims, so for a role that carries a filter `Broadcast` keeps the shared column projection but delivers it only to the subscribers whose claims admit each row (`ResolvedPermissions.RowVisible`, evaluated against the full event via the numeric-aware comparison seeded from the schema registry). This is the [#319](https://github.com/Wave-RF/WaveHouse/issues/319) fix that closes the query/stream row-level-security drift; roles without a filter keep the pure once-per-role fast path. `ReplayFrame` shares the same projection and per-connection row check for the handler's gap-fill.
 - **subscriber.go** — `Subscriber`, the per-connection handle. It owns a single ready-to-write outbound queue of `Frame`s (each tagged with its `kind`, so the handler labels the write where it happens): producers — the keepalive wheel and the event `Hub` — fan frames in with `Send` (non-blocking; a full queue drops and the `Hub` counts it), and the handler drains `Frames()` to the client verbatim. The queue is sized for buffering live events (cap 64, up from the keepalive-only cap 1; #152 will make it a knob), and an `Evicted()` channel is the seam the slow-consumer follow-up closes to disconnect a wedged consumer.
 - **bucket.go** — `Bucket`, the reusable fan-out primitive: a concurrency-safe set of subscribers. `Push` delivers a shared `Frame` to each fire-and-forget (the keepalive wheel's ring); `Snapshot` exposes the members so the event `Hub` can fan out while inspecting each `Send` result (to count drops). The `Hub` holds one `Bucket` per `(topic, role)` so a projected frame is built once and sent to every member instead of re-projected per subscriber.
 - **heartbeat.go** — The keepalive wheel (`Heartbeater`). A single process-wide ticker fans a minimal `:` comment across the ring of `Bucket`s, waking ~1/N of live streams per tick so the writes don't synchronize. The effective per-connection keepalive period is `stream.keepalive_interval` (the wheel ticks every `keepalive_interval ÷ keepalive_buckets`, so one rotation spans the interval); the owning handler goroutine does the actual write, so the shared ticker never touches a `ResponseWriter` directly.
diff --git a/go.mod b/go.mod
index 7a1c7d32..0ea6370e 100644
--- a/go.mod
+++ b/go.mod
@@ -1,6 +1,6 @@
 module github.com/Wave-RF/WaveHouse
 
-go 1.26.4
+go 1.26.5
 
 tool (
 	github.com/Zxilly/go-size-analyzer/cmd/gsa
diff --git a/internal/api/errors_test.go b/internal/api/errors_test.go
index 358a3f34..3f2956e7 100644
--- a/internal/api/errors_test.go
+++ b/internal/api/errors_test.go
@@ -194,7 +194,7 @@ func TestAuthzDenied_LogsChiRoutePattern(t *testing.T) {
 	router := NewRouter(Dependencies{
 		Ingest:      NewIngestHandler(reg, &testutil.MockPublisher{}, logger),
 		Query:       &QueryHandler{},
-		SSE:         NewStreamHandler(stream.NewHub(nil, nil), nil),
+		SSE:         NewStreamHandler(stream.NewHub(nil, nil, nil), nil),
 		Health:      &HealthHandler{},
 		Schema:      NewSchemaHandler(reg),
 		AuthMW:      func(next http.Handler) http.Handler { return next },
diff --git a/internal/api/router_test.go b/internal/api/router_test.go
index bf9dcaac..a0c1d5ac 100644
--- a/internal/api/router_test.go
+++ b/internal/api/router_test.go
@@ -283,7 +283,7 @@ func TestNewRouter_RoutesRegistered(t *testing.T) {
 		{Name: "events", Columns: []discovery.Column{{Name: "id", Type: "String"}}},
 	})
 	pub := &testutil.MockPublisher{}
-	hub := stream.NewHub(nil, nil)
+	hub := stream.NewHub(nil, nil, nil)
 
 	deps := Dependencies{
 		Ingest:  NewIngestHandler(reg, pub, testutil.NopLogger()),
@@ -339,7 +339,7 @@ func TestNewRouter_RoutesRegistered(t *testing.T) {
 func TestNewRouter_CORSOnStream(t *testing.T) {
 	t.Parallel()
 
-	hub := stream.NewHub(nil, nil)
+	hub := stream.NewHub(nil, nil, nil)
 	router := NewRouter(Dependencies{
 		SSE:         NewStreamHandler(hub, nil),
 		Health:      &HealthHandler{},
@@ -411,7 +411,7 @@ func TestNewRouter_RawSQLAdminGate(t *testing.T) {
 
 	reg := discovery.NewSchemaRegistryFromMap(nil)
 	pub := &testutil.MockPublisher{}
-	hub := stream.NewHub(nil, nil)
+	hub := stream.NewHub(nil, nil, nil)
 
 	router := NewRouter(Dependencies{
 		Ingest:      NewIngestHandler(reg, pub, testutil.NopLogger()),
@@ -472,7 +472,7 @@ func TestNewRouter_OptionalDepsNil(t *testing.T) {
 
 	reg := discovery.NewSchemaRegistryFromMap(nil)
 	pub := &testutil.MockPublisher{}
-	hub := stream.NewHub(nil, nil)
+	hub := stream.NewHub(nil, nil, nil)
 
 	deps := Dependencies{
 		Ingest:      NewIngestHandler(reg, pub, testutil.NopLogger()),
@@ -523,7 +523,7 @@ func TestNewRouter_NotFoundEmitsJSON(t *testing.T) {
 
 	reg := discovery.NewSchemaRegistryFromMap(nil)
 	pub := &testutil.MockPublisher{}
-	hub := stream.NewHub(nil, nil)
+	hub := stream.NewHub(nil, nil, nil)
 	deps := Dependencies{
 		Ingest: NewIngestHandler(reg, pub, testutil.NopLogger()),
 		Query:  &QueryHandler{},
@@ -548,7 +548,7 @@ func TestNewRouter_MethodNotAllowedEmitsJSON(t *testing.T) {
 
 	reg := discovery.NewSchemaRegistryFromMap(nil)
 	pub := &testutil.MockPublisher{}
-	hub := stream.NewHub(nil, nil)
+	hub := stream.NewHub(nil, nil, nil)
 	deps := Dependencies{
 		Ingest: NewIngestHandler(reg, pub, testutil.NopLogger()),
 		Query:  &QueryHandler{},
@@ -645,7 +645,7 @@ func TestNewRouter_SchemaAdminOnly(t *testing.T) {
 	t.Parallel()
 	reg := discovery.NewSchemaRegistryFromMap(nil)
 	pub := &testutil.MockPublisher{}
-	hub := stream.NewHub(nil, nil)
+	hub := stream.NewHub(nil, nil, nil)
 
 	router := NewRouter(Dependencies{
 		Ingest:      NewIngestHandler(reg, pub, testutil.NopLogger()),
diff --git a/internal/api/stream.go b/internal/api/stream.go
index 44485224..a0727573 100644
--- a/internal/api/stream.go
+++ b/internal/api/stream.go
@@ -42,10 +42,13 @@ func (h *StreamHandler) Handle(w http.ResponseWriter, r *http.Request) {
 	}
 
 	// Resolve stream permissions for this request. The raw role from context is the
-	// bucket key: the Hub projects/serializes once per (topic, role), and column
-	// visibility derives only from the role+table policy entry — never from claims
-	// (see stream.Hub). Evaluate maps an empty role to the policy default_role.
+	// bucket key: the Hub serializes the column projection once per (topic, role),
+	// since column visibility derives only from the role+table policy entry. Claims
+	// are separate — they drive the row-level-security filter, which the Hub evaluates
+	// per subscriber (see stream.Hub), so they ride on the Subscriber rather than the
+	// bucket key. Evaluate maps an empty role to the policy default_role.
 	role := auth.RoleFromContext(r.Context())
+	claims, _ := auth.ClaimsFromContext(r.Context())
 
 	// TODO: impl scope
 	scope := ""
@@ -79,6 +82,7 @@ func (h *StreamHandler) Handle(w http.ResponseWriter, r *http.Request) {
 	// buffer in the subscriber queue instead of being missed (an overlap with
 	// replay yields duplicates, deduped client-side by id: — at-least-once).
 	sub := stream.NewSubscriber()
+	sub.SetClaims(claims)
 	h.Hub.Add(topic, role, sub)
 	defer h.Hub.Remove(topic, role, sub)
 
@@ -96,7 +100,7 @@ func (h *StreamHandler) Handle(w http.ResponseWriter, r *http.Request) {
 		// the replayed frame. A write error means the client is gone, so stop the
 		// gap-fill and let the deferred cleanup unwind.
 		sendReplay := func(data []byte) bool {
-			f, ok := h.Hub.ReplayFrame(role, data)
+			f, ok := h.Hub.ReplayFrame(role, claims, data)
 			if !ok {
 				return true // filtered for this role — skip
 			}
diff --git a/internal/api/stream_test.go b/internal/api/stream_test.go
index 5d67b763..88db1d99 100644
--- a/internal/api/stream_test.go
+++ b/internal/api/stream_test.go
@@ -22,7 +22,7 @@ import (
 
 func TestSSE_RejectsMissingOrInvalidTable(t *testing.T) {
 	t.Parallel()
-	h := &StreamHandler{Hub: stream.NewHub(nil, nil)}
+	h := &StreamHandler{Hub: stream.NewHub(nil, nil, nil)}
 
 	cases := []struct {
 		name    string
@@ -50,7 +50,7 @@ func TestSSE_RejectsMissingOrInvalidTable(t *testing.T) {
 
 func TestSSE_AcceptsSafeTableName(t *testing.T) {
 	t.Parallel()
-	h := &StreamHandler{Hub: stream.NewHub(nil, nil)}
+	h := &StreamHandler{Hub: stream.NewHub(nil, nil, nil)}
 
 	// Use a request context that's already cancelled so the handler exits
 	// the live-stream select loop immediately instead of blocking the test.
@@ -75,7 +75,7 @@ func TestSSE_EmitsHeartbeatsWhenIdle(t *testing.T) {
 	hb := stream.NewHeartbeater(20*time.Millisecond, 1)
 	go hb.Run(t.Context())
 
-	h := &StreamHandler{Hub: stream.NewHub(nil, nil), Heartbeater: hb}
+	h := &StreamHandler{Hub: stream.NewHub(nil, nil, nil), Heartbeater: hb}
 
 	ctx, cancel := context.WithCancel(context.Background())
 	req := httptest.NewRequestWithContext(ctx, http.MethodGet, "/v1/stream?table=clicks", nil)
@@ -114,7 +114,7 @@ func TestSSE_WheelTickRacesHandlerTeardown(t *testing.T) {
 	defer cancel()
 	go hb.Run(ctx)
 
-	h := &StreamHandler{Hub: stream.NewHub(nil, nil), Heartbeater: hb}
+	h := &StreamHandler{Hub: stream.NewHub(nil, nil, nil), Heartbeater: hb}
 
 	const conns = 40
 	var wg sync.WaitGroup
diff --git a/internal/discovery/validation.go b/internal/discovery/validation.go
index 90af29be..b2dbdf51 100644
--- a/internal/discovery/validation.go
+++ b/internal/discovery/validation.go
@@ -48,9 +48,9 @@ func Validate(schema *TableSchema, data map[string]any) error {
 	return nil
 }
 
-// isTypeCompatible checks whether a Go/JSON value can be stored in the given ClickHouse type.
-func isTypeCompatible(chType string, val any) bool {
-	// Robustly unwrap nested modifiers (e.g., LowCardinality(Nullable(String)))
+// unwrapType strips Nullable(...) and LowCardinality(...) modifiers (nested in any
+// order, e.g. LowCardinality(Nullable(String))) down to the base ClickHouse type.
+func unwrapType(chType string) string {
 	for {
 		if strings.HasPrefix(chType, "Nullable(") && strings.HasSuffix(chType, ")") {
 			chType = chType[9 : len(chType)-1]
@@ -60,8 +60,13 @@ func isTypeCompatible(chType string, val any) bool {
 			chType = chType[15 : len(chType)-1]
 			continue
 		}
-		break
+		return chType
 	}
+}
+
+// isTypeCompatible checks whether a Go/JSON value can be stored in the given ClickHouse type.
+func isTypeCompatible(chType string, val any) bool {
+	chType = unwrapType(chType)
 
 	switch {
 	// String-compatible types accept Strings, Numbers (coerced), and Bools
@@ -144,6 +149,14 @@ func isTypeCompatible(chType string, val any) bool {
 	}
 }
 
+// IsNumericType reports whether chType is a ClickHouse numeric type (integer, float,
+// or decimal), unwrapping Nullable/LowCardinality modifiers first. The stream
+// row-filter evaluator uses it to choose numeric vs lexicographic comparison for a
+// column, so ordering predicates (>, <) on numbers match ClickHouse (9 < 100).
+func IsNumericType(chType string) bool {
+	return isNumericType(unwrapType(chType))
+}
+
 // isNumericType returns true for ClickHouse integer, float, and decimal types.
 func isNumericType(chType string) bool {
 	switch {
diff --git a/internal/discovery/validation_test.go b/internal/discovery/validation_test.go
index 1dbbb08a..4a4015d1 100644
--- a/internal/discovery/validation_test.go
+++ b/internal/discovery/validation_test.go
@@ -224,3 +224,32 @@ func TestIsNumericType(t *testing.T) {
 		})
 	}
 }
+
+// TestIsNumericType_Exported checks the exported wrapper the stream row-filter uses:
+// it must unwrap Nullable/LowCardinality (in any nesting) before classifying, so a
+// Nullable(UInt64) column still compares numerically.
+func TestIsNumericType_Exported(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		chType string
+		want   bool
+	}{
+		{"UInt64", true},
+		{"Nullable(UInt64)", true},
+		{"LowCardinality(Int32)", true},
+		{"LowCardinality(Nullable(Float64))", true},
+		{"Decimal(10,2)", true},
+		{"String", false},
+		{"Nullable(String)", false},
+		{"LowCardinality(String)", false},
+		{"DateTime", false},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.chType, func(t *testing.T) {
+			t.Parallel()
+			assert.Equal(t, tt.want, IsNumericType(tt.chType))
+		})
+	}
+}
diff --git a/internal/policy/policy.go b/internal/policy/policy.go
index 198bc5a3..c0f5a4f3 100644
--- a/internal/policy/policy.go
+++ b/internal/policy/policy.go
@@ -62,11 +62,16 @@ type Filter struct {
 
 // ResolvedPermissions is the result of evaluating a policy against JWT claims.
 type ResolvedPermissions struct {
-	Allowed             bool
-	AllowColumns        []string
-	DenyColumns         []string
-	WhereClause         string
-	WhereParams         []any
+	Allowed      bool
+	AllowColumns []string
+	DenyColumns  []string
+	WhereClause  string
+	WhereParams  []any
+	// rowFilter is the same row-level-security predicate as WhereClause/WhereParams,
+	// kept in resolved form so the stream path can evaluate it in memory (RowVisible)
+	// while the query path renders it to SQL. Both derive from one resolvePredicates
+	// call in Evaluate, so the two read surfaces can't drift. See rowfilter.go.
+	rowFilter           []resolvedPredicate
 	CheckClauses        map[string]any // column → required value (for inserts)
 	AllowedAggregations []string
 	DeniedAggregations  []string
@@ -189,7 +194,13 @@ func Evaluate(p *Policy, role, table, operation string, claims map[string]any) *
 				return &ResolvedPermissions{Allowed: false}
 			}
 		}
-		clauses, params := resolveFilters(perms.Filter, claims)
+		// Resolve the row-filter once into predicates, then render both read surfaces
+		// from that single source so they can't drift: the query path binds them into
+		// a SQL WHERE here; the stream path evaluates the same predicates in memory
+		// (ResolvedPermissions.RowVisible).
+		preds := resolvePredicates(perms.Filter, claims)
+		resolved.rowFilter = preds
+		clauses, params := predicatesToSQL(preds)
 		if len(clauses) > 0 {
 			resolved.WhereClause = strings.Join(clauses, " AND ")
 			resolved.WhereParams = params
@@ -214,52 +225,92 @@ func Evaluate(p *Policy, role, table, operation string, claims map[string]any) *
 	return resolved
 }
 
-// resolveFilters converts filter definitions with claim templates into SQL WHERE clauses.
-func resolveFilters(filters map[string]Filter, claims map[string]any) ([]string, []any) {
-	var clauses []string
-	var params []any
+// resolvedPredicate is one row-filter comparison with its claim templates already
+// resolved to concrete string values — the shared, render-agnostic form the query
+// path turns into SQL (predicatesToSQL) and the stream path evaluates in memory
+// (RowVisible). Op is one of "=", "!=", ">", "<", "in". Values holds one element
+// for the scalar operators and zero-or-more for "in" (empty ⇒ matches no rows).
+type resolvedPredicate struct {
+	Column string
+	Op     string
+	Values []string
+}
+
+// resolvePredicates resolves each filter's claim templates once into predicates.
+// Both read surfaces derive from this single result so they can't drift; the
+// operator order within a column (=, !=, >, <, in) mirrors the former inline SQL.
+func resolvePredicates(filters map[string]Filter, claims map[string]any) []resolvedPredicate {
+	var preds []resolvedPredicate
 	for col, f := range filters {
-		// Quote the policy-authored column the same way the query builder quotes
-		// caller columns, so a row-filter on a weird-but-legal column name (dots,
-		// spaces, keywords) is emitted safely.
-		qcol := chsql.QuoteIdent(col)
 		if f.Eq != nil {
-			val := resolveTemplate(*f.Eq, claims)
-			clauses = append(clauses, fmt.Sprintf("%s = ?", qcol))
-			params = append(params, val)
+			preds = append(preds, resolvedPredicate{col, "=", []string{resolveTemplate(*f.Eq, claims)}})
 		}
 		if f.Neq != nil {
-			val := resolveTemplate(*f.Neq, claims)
-			clauses = append(clauses, fmt.Sprintf("%s != ?", qcol))
-			params = append(params, val)
+			preds = append(preds, resolvedPredicate{col, "!=", []string{resolveTemplate(*f.Neq, claims)}})
 		}
 		if f.Gt != nil {
-			val := resolveTemplate(*f.Gt, claims)
-			clauses = append(clauses, fmt.Sprintf("%s > ?", qcol))
-			params = append(params, val)
+			preds = append(preds, resolvedPredicate{col, ">", []string{resolveTemplate(*f.Gt, claims)}})
 		}
 		if f.Lt != nil {
-			val := resolveTemplate(*f.Lt, claims)
-			clauses = append(clauses, fmt.Sprintf("%s < ?", qcol))
-			params = append(params, val)
+			preds = append(preds, resolvedPredicate{col, "<", []string{resolveTemplate(*f.Lt, claims)}})
 		}
 		if f.In != nil {
-			vals := resolveInValues(*f.In, claims)
-			if len(vals) == 0 {
+			preds = append(preds, resolvedPredicate{col, "in", toStrings(resolveInValues(*f.In, claims))})
+		}
+	}
+	return preds
+}
+
+// predicatesToSQL renders resolved predicates into WHERE clauses and bound params.
+func predicatesToSQL(preds []resolvedPredicate) ([]string, []any) {
+	var clauses []string
+	var params []any
+	for _, p := range preds {
+		// Quote the policy-authored column the same way the query builder quotes
+		// caller columns, so a row-filter on a weird-but-legal column name (dots,
+		// spaces, keywords) is emitted safely.
+		qcol := chsql.QuoteIdent(p.Column)
+		switch p.Op {
+		case "in":
+			if len(p.Values) == 0 {
 				// An empty/unresolvable set fails closed: a row filter scoped to no
 				// values matches no rows, never widening to all of them (the #224
 				// fail-open). `IN ()` is not valid SQL, so emit a constant false.
 				clauses = append(clauses, "1 = 0")
 			} else {
-				placeholders := strings.TrimSuffix(strings.Repeat("?,", len(vals)), ",")
+				placeholders := strings.TrimSuffix(strings.Repeat("?,", len(p.Values)), ",")
 				clauses = append(clauses, fmt.Sprintf("%s IN (%s)", qcol, placeholders))
-				params = append(params, vals...)
+				for _, v := range p.Values {
+					params = append(params, v)
+				}
 			}
+		default:
+			clauses = append(clauses, fmt.Sprintf("%s %s ?", qcol, p.Op))
+			params = append(params, p.Values[0])
 		}
 	}
 	return clauses, params
 }
 
+// resolveFilters converts filter definitions with claim templates into SQL WHERE
+// clauses. Retained as the predicates→SQL composition the query-path tests target.
+func resolveFilters(filters map[string]Filter, claims map[string]any) ([]string, []any) {
+	return predicatesToSQL(resolvePredicates(filters, claims))
+}
+
+// toStrings normalizes resolveInValues' []any (already stringified elements) to the
+// []string a resolvedPredicate carries.
+func toStrings(vals []any) []string {
+	if len(vals) == 0 {
+		return nil
+	}
+	out := make([]string, len(vals))
+	for i, v := range vals {
+		out[i] = fmt.Sprint(v)
+	}
+	return out
+}
+
 // resolveTemplate resolves {{ jwt.claim.path }} templates against JWT claims.
 // If a claim path cannot be resolved, the template placeholder is replaced with
 // an empty string to prevent "<nil>" from leaking into SQL filters.
diff --git a/internal/policy/rowfilter.go b/internal/policy/rowfilter.go
new file mode 100644
index 00000000..b008165b
--- /dev/null
+++ b/internal/policy/rowfilter.go
@@ -0,0 +1,149 @@
+package policy
+
+import (
+	"math"
+	"math/big"
+	"strconv"
+	"strings"
+)
+
+// HasRowFilter reports whether this role/table entry carries a row-level-security
+// predicate. The stream fan-out uses it to decide whether an event can be projected
+// once for a whole role bucket (no filter) or must be checked per subscriber against
+// that subscriber's claims (filter present). A nil receiver (no policy applies) has
+// no filter.
+func (p *ResolvedPermissions) HasRowFilter() bool {
+	return p != nil && len(p.rowFilter) > 0
+}
+
+// RowVisible reports whether row satisfies every resolved row-filter predicate — the
+// in-memory twin of the query path's WHERE clause, evaluated against a decoded event
+// so the stream applies the same row-level security the query path does. Predicates
+// are ANDed; the query path joins them with AND too.
+//
+// numericCols maps column name → true when that column's ClickHouse type is numeric
+// (the caller supplies it from the table schema; see discovery.IsNumericType).
+// Numeric columns compare numerically, so 9 < 100 as ClickHouse would; every other
+// column — and any column absent from the map, e.g. when no schema is available —
+// compares as text. This is exact for the equality/set operators row-level security
+// actually uses (=, !=, in) and best-effort for ordering (>, <): it cannot perfectly
+// mirror ClickHouse's per-type coercion for exotic types (Decimal / Int128 beyond
+// float64 precision, Date/DateTime stored as Unix numbers). Every ambiguous or
+// uncomparable case fails closed — the row is hidden, never leaked — so the boundary
+// costs availability, not confidentiality.
+//
+// A nil receiver (no policy applies) makes every row visible.
+func (p *ResolvedPermissions) RowVisible(row map[string]any, numericCols map[string]bool) bool {
+	if p == nil {
+		return true
+	}
+	// A denied role sees no rows — fail closed, mirroring the !Allowed guard on
+	// IsColumnAllowed, so a denied receiver never reads as "no filter ⇒ all visible".
+	if !p.Allowed {
+		return false
+	}
+	for _, pred := range p.rowFilter {
+		if !pred.matches(row, numericCols[pred.Column]) {
+			return false
+		}
+	}
+	return true
+}
+
+// matches evaluates one predicate against the row, failing closed (false) whenever
+// the value is absent or can't be compared as required.
+func (pred resolvedPredicate) matches(row map[string]any, numeric bool) bool {
+	raw, ok := row[pred.Column]
+	if !ok {
+		return false // column not in the event ⇒ can't prove the row is allowed
+	}
+	switch pred.Op {
+	case "=":
+		c, ok := compareScalar(raw, pred.Values[0], numeric)
+		return ok && c == 0
+	case "!=":
+		c, ok := compareScalar(raw, pred.Values[0], numeric)
+		return ok && c != 0
+	case ">":
+		c, ok := compareScalar(raw, pred.Values[0], numeric)
+		return ok && c > 0
+	case "<":
+		c, ok := compareScalar(raw, pred.Values[0], numeric)
+		return ok && c < 0
+	case "in":
+		for _, v := range pred.Values {
+			if c, ok := compareScalar(raw, v, numeric); ok && c == 0 {
+				return true
+			}
+		}
+		return false
+	default:
+		return false
+	}
+}
+
+// compareScalar compares an event value against a resolved filter value, returning
+// -1/0/+1 and ok=false when the comparison can't be made (a non-scalar event value,
+// or a numeric comparison whose operands don't parse as numbers). numeric selects
+// numeric vs lexicographic ordering.
+func compareScalar(rowVal any, filterVal string, numeric bool) (int, bool) {
+	s, ok := scalarString(rowVal)
+	if !ok {
+		return 0, false
+	}
+	if numeric {
+		a, err1 := strconv.ParseFloat(s, 64)
+		b, err2 := strconv.ParseFloat(filterVal, 64)
+		// NaN must be rejected explicitly: ParseFloat accepts "NaN", and NaN's
+		// three-way comparison reads as "equal to everything" below — a fail-open.
+		if err1 != nil || err2 != nil || math.IsNaN(a) || math.IsNaN(b) {
+			return 0, false
+		}
+		switch {
+		case a < b:
+			return -1, true
+		case a > b:
+			return 1, true
+		default:
+			// Float equality is not proof of equality: distinct integers beyond
+			// 2^53 (string-encoded on ingest exactly to survive JS precision loss)
+			// collapse to one float64, and rounding is monotonic so only the
+			// equal case is in doubt. Resolve the tie at full precision.
+			return compareExact(s, filterVal)
+		}
+	}
+	return strings.Compare(s, filterVal), true
+}
+
+// compareExact compares two numeric strings at arbitrary precision — the tie-break
+// for operands float64 cannot tell apart. ok=false when either side isn't an exact
+// rational (±Inf, malformed), failing the predicate closed.
+func compareExact(a, b string) (int, bool) {
+	ra, ok := new(big.Rat).SetString(a)
+	if !ok {
+		return 0, false
+	}
+	rb, ok := new(big.Rat).SetString(b)
+	if !ok {
+		return 0, false
+	}
+	return ra.Cmp(rb), true
+}
+
+// scalarString renders a JSON-decoded scalar as the canonical string compared
+// against a (string-valued) filter. Non-scalars (arrays, objects, null) return
+// ok=false so the predicate fails closed rather than guessing. JSON numbers arrive
+// as float64 via encoding/json; -1 precision emits the shortest round-trip form
+// without an exponent, so integer IDs read back as "123", not "1.23e+02".
+func scalarString(v any) (string, bool) {
+	switch x := v.(type) {
+	case string:
+		return x, true
+	case float64:
+		return strconv.FormatFloat(x, 'f', -1, 64), true
+	case bool:
+		return strconv.FormatBool(x), true
+	default:
+		return "", false
+	}
+}
diff --git a/internal/policy/rowfilter_test.go b/internal/policy/rowfilter_test.go
new file mode 100644
index 00000000..6e6103cc
--- /dev/null
+++ b/internal/policy/rowfilter_test.go
@@ -0,0 +1,141 @@
+package policy
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+// evalRowFilter builds a one-role/one-table policy carrying filter and returns the
+// permissions resolved against claims, so tests exercise the full
+// resolvePredicates → RowVisible path the stream fan-out uses.
+func evalRowFilter(t *testing.T, filter map[string]Filter, claims map[string]any) *ResolvedPermissions {
+	t.Helper()
+	p := &Policy{Tables: map[string]TablePolicy{
+		"t": {Select: map[string]RolePermissions{"r": {Filter: filter}}},
+	}}
+	return Evaluate(p, "r", "t", "select", claims)
+}
+
+// TestRowVisible_EqualityScoping covers the canonical row-level-security shape —
+// tenant_id = {{ jwt.tenant }} — including the fail-closed on a missing column that
+// stops an event without the filtered field from slipping through.
+func TestRowVisible_EqualityScoping(t *testing.T) {
+	t.Parallel()
+	perms := evalRowFilter(t, map[string]Filter{"tenant_id": {Eq: new("{{ jwt.tenant }}")}}, map[string]any{"tenant": "acme"})
+	is := assert.New(t)
+	is.True(perms.HasRowFilter())
+	is.True(perms.RowVisible(map[string]any{"tenant_id": "acme"}, nil))
+	is.False(perms.RowVisible(map[string]any{"tenant_id": "globex"}, nil))
+	is.False(perms.RowVisible(map[string]any{"other": "x"}, nil), "missing filtered column ⇒ fail closed")
+}
+
+// TestRowVisible_InSet covers _in against an array claim (multi-tenant scoping) and
+// the fail-closed empty-set case (absent claim never widens to all rows).
+func TestRowVisible_InSet(t *testing.T) {
+	t.Parallel()
+	perms := evalRowFilter(t, map[string]Filter{"tenant_id": {In: new("{{ jwt.tenants }}")}}, map[string]any{"tenants": []any{"a", "b"}})
+	assert.True(t, perms.RowVisible(map[string]any{"tenant_id": "b"}, nil))
+	assert.False(t, perms.RowVisible(map[string]any{"tenant_id": "c"}, nil))
+
+	empty := evalRowFilter(t, map[string]Filter{"tenant_id": {In: new("{{ jwt.tenants }}")}}, nil)
+	assert.True(t, empty.HasRowFilter())
+	assert.False(t, empty.RowVisible(map[string]any{"tenant_id": "a"}, nil), "absent claim ⇒ empty set ⇒ no row matches")
+}
+
+func TestRowVisible_Neq(t *testing.T) {
+	t.Parallel()
+	perms := evalRowFilter(t, map[string]Filter{"status": {Neq: new("deleted")}}, nil)
+	assert.True(t, perms.RowVisible(map[string]any{"status": "active"}, nil))
+	assert.False(t, perms.RowVisible(map[string]any{"status": "deleted"}, nil))
+}
+
+// TestRowVisible_Ordering_NumericVsLexicographic is the schema-informed behavior: an
+// event value of 9 is numerically LESS than 100 but lexicographically GREATER
+// ("9" > "100"). With the column marked numeric the row is hidden (matching
+// ClickHouse); the no-schema lexicographic fallback would leak it — the footgun the
+// schema closes.
+func TestRowVisible_Ordering_NumericVsLexicographic(t *testing.T) {
+	t.Parallel()
+	perms := evalRowFilter(t, map[string]Filter{"amount": {Gt: new("100")}}, nil)
+	row := map[string]any{"amount": float64(9)}
+
+	assert.False(t, perms.RowVisible(row, map[string]bool{"amount": true}), "numeric: 9 is not > 100")
+	assert.True(t, perms.RowVisible(row, nil), `lexicographic fallback: "9" > "100"`)
+	assert.True(t, perms.RowVisible(map[string]any{"amount": float64(250)}, map[string]bool{"amount": true}))
+}
+
+// TestRowVisible_NumericEquality_FloatFormatting: a JSON float64(100) equals the
+// string filter value "100" under numeric comparison, so integer-valued numeric
+// columns aren't tripped up by float formatting.
+func TestRowVisible_NumericEquality_FloatFormatting(t *testing.T) {
+	t.Parallel()
+	perms := evalRowFilter(t, map[string]Filter{"amount": {Eq: new("100")}}, nil)
+	num := map[string]bool{"amount": true}
+	assert.True(t, perms.RowVisible(map[string]any{"amount": float64(100)}, num))
+	assert.False(t, perms.RowVisible(map[string]any{"amount": float64(101)}, num))
+}
+
+// TestRowVisible_NaN_FailsClosed: strconv.ParseFloat accepts "NaN", and NaN's
+// three-way comparison would otherwise read as "equal to everything" — a fail-open
+// that delivers a row the query path's WHERE excludes. Either operand parsing to
+// NaN must withhold the row instead.
+func TestRowVisible_NaN_FailsClosed(t *testing.T) {
+	t.Parallel()
+	num := map[string]bool{"amount": true}
+
+	byRow := evalRowFilter(t, map[string]Filter{"amount": {Eq: new("100")}}, nil)
+	assert.False(t, byRow.RowVisible(map[string]any{"amount": "NaN"}, num), "NaN row value must not equal 100")
+
+	byClaim := evalRowFilter(t, map[string]Filter{"amount": {Eq: new("{{ jwt.cap }}")}}, map[string]any{"cap": "NaN"})
+	assert.False(t, byClaim.RowVisible(map[string]any{"amount": float64(100)}, num), "NaN claim must not match any row")
+
+	neq := evalRowFilter(t, map[string]Filter{"amount": {Neq: new("NaN")}}, nil)
+	assert.False(t, neq.RowVisible(map[string]any{"amount": float64(100)}, num), "NaN is uncomparable, so even != fails closed")
+}
+
+// TestRowVisible_NumericEquality_ExactBeyondFloat64: ingest accepts string-encoded
+// numerics precisely so 64-bit IDs survive JS precision loss; equality must not
+// collapse distinct IDs that round to the same float64 (adjacent values past 2^53).
+func TestRowVisible_NumericEquality_ExactBeyondFloat64(t *testing.T) {
+	t.Parallel()
+	num := map[string]bool{"id": true}
+
+	perms := evalRowFilter(t, map[string]Filter{"id": {Eq: new("9007199254740993")}}, nil)
+	assert.False(t, perms.RowVisible(map[string]any{"id": "9007199254740992"}, num), "float64-equal neighbors are not equal")
+	assert.True(t, perms.RowVisible(map[string]any{"id": "9007199254740993"}, num))
+	assert.True(t, perms.RowVisible(map[string]any{"id": "9007199254740993.0"}, num), "same value in a different rendering still matches")
+
+	neq := evalRowFilter(t, map[string]Filter{"id": {Neq: new("9007199254740993")}}, nil)
+	assert.True(t, neq.RowVisible(map[string]any{"id": "9007199254740992"}, num), "the exact tie-break keeps distinct IDs unequal for !=")
+}
+
+func TestRowVisible_NilReceiver_AllVisible(t *testing.T) {
+	t.Parallel()
+	var perms *ResolvedPermissions
+	assert.True(t, perms.RowVisible(map[string]any{"x": "y"}, nil))
+	assert.False(t, perms.HasRowFilter())
+}
+
+// TestRowVisible_NonScalar_FailsClosed: a nested/array/null event value can't be
+// compared to a scalar filter value, so the predicate fails closed rather than guess.
+func TestRowVisible_NonScalar_FailsClosed(t *testing.T) {
+	t.Parallel()
+	perms := evalRowFilter(t, map[string]Filter{"tenant_id": {Eq: new("acme")}}, nil)
+	assert.False(t, perms.RowVisible(map[string]any{"tenant_id": []any{"acme"}}, nil))
+	assert.False(t, perms.RowVisible(map[string]any{"tenant_id": nil}, nil))
+}
+
+// TestRowVisible_MultiplePredicates_AllMustPass: predicates are ANDed, matching the
+// query path's "AND"-joined WHERE clause.
+func TestRowVisible_MultiplePredicates_AllMustPass(t *testing.T) {
+	t.Parallel()
+	perms := evalRowFilter(t, map[string]Filter{
+		"tenant_id": {Eq: new("{{ jwt.tenant }}")},
+		"amount":    {Gt: new("100")},
+	}, map[string]any{"tenant": "acme"})
+	num := map[string]bool{"amount": true}
+	assert.True(t, perms.RowVisible(map[string]any{"tenant_id": "acme", "amount": float64(250)}, num))
+	assert.False(t, perms.RowVisible(map[string]any{"tenant_id": "acme", "amount": float64(9)}, num), "amount fails")
+	assert.False(t, perms.RowVisible(map[string]any{"tenant_id": "globex", "amount": float64(250)}, num), "tenant fails")
+}
diff --git a/internal/stream/hub.go b/internal/stream/hub.go
index e84821fa..934556f4 100644
--- a/internal/stream/hub.go
+++ b/internal/stream/hub.go
@@ -5,25 +5,29 @@ import (
 	"encoding/json"
 	"sync"
 
+	"github.com/Wave-RF/WaveHouse/internal/discovery"
 	"github.com/Wave-RF/WaveHouse/internal/ingest"
 	"github.com/Wave-RF/WaveHouse/internal/policy"
 )
 
-// Hub fans live events out to SSE subscribers, projecting and serializing each
-// event ONCE per (topic, role) instead of once per subscriber — the #294 lever.
-// Subscribers register under (topic, role); Broadcast decodes the event once,
-// applies each subscribed role's column policy once, builds one SSE frame per
-// role, and fans it to every member of that role's Bucket.
+// Hub fans live events out to SSE subscribers. Column projection is serialized ONCE
+// per (topic, role) instead of once per subscriber — the #294/#353 lever — because
+// column visibility depends solely on the role+table policy entry, never on JWT
+// claims, so the projected frame is identical for every subscriber of a role.
 //
-// The (topic, role) key is sufficient because the live path's only per-subscriber
-// transform is column filtering, which depends solely on the role+table policy
-// entry — never on JWT claims (claims feed only row-level WHERE/CHECK, which the
-// stream path does not apply). See projectData.
+// Row-level security is the exception: a role's row-filter (RLS predicate) is
+// resolved against each subscriber's claims, so two subscribers of the same role can
+// be entitled to different rows. For a role that carries a row-filter, Broadcast
+// therefore keeps the shared column projection but evaluates row visibility PER
+// subscriber (ResolvedPermissions.RowVisible) before delivering — closing the
+// query/stream RLS drift in #319. Roles without a row-filter keep the pure
+// once-per-role fast path unchanged. See projectColumns.
 type Hub struct {
-	mu     sync.RWMutex
-	topics map[string]*topicRoutes
-	policy *policy.Store // nil ⇒ policy filtering not configured (legacy passthrough)
-	metric *Metrics      // nil-safe
+	mu       sync.RWMutex
+	topics   map[string]*topicRoutes
+	policy   *policy.Store             // nil ⇒ policy filtering not configured (legacy passthrough)
+	registry *discovery.SchemaRegistry // nil ⇒ no column types; row-filter ordering compares lexicographically
+	metric   *Metrics                  // nil-safe
 }
 
 // topicRoutes holds the per-role buckets subscribed to one topic.
@@ -33,9 +37,11 @@ type topicRoutes struct {
 
 // NewHub builds an event hub. A nil policy store passes every event through
 // unfiltered (the unwired-tests case); a non-nil store whose Get returns nil is a
-// total lockout (a deleted/absent policy denies everyone). metric may be nil.
-func NewHub(policyStore *policy.Store, metric *Metrics) *Hub {
-	return &Hub{topics: make(map[string]*topicRoutes), policy: policyStore, metric: metric}
+// total lockout (a deleted/absent policy denies everyone). A nil registry disables
+// numeric-aware row-filter comparison (ordering predicates fall back to
+// lexicographic); metric may be nil.
+func NewHub(policyStore *policy.Store, registry *discovery.SchemaRegistry, metric *Metrics) *Hub {
+	return &Hub{topics: make(map[string]*topicRoutes), policy: policyStore, registry: registry, metric: metric}
 }
 
 // Add registers sub to receive events for (topic, role), creating the role bucket
@@ -100,10 +106,12 @@ type roleBucket struct {
 	bucket Bucket
 }
 
-// Broadcast projects raw — a published EventMessage JSON delivered on topic —
-// once per subscribed role and fans the finished SSE frame to that role's bucket.
-// The expensive work (decode, evaluate, marshal) happens once per distinct role,
-// not once per subscriber.
+// Broadcast projects raw — a published EventMessage JSON delivered on topic — and
+// fans the finished SSE frame to each subscribed role's bucket. The column
+// projection (decode, evaluate, marshal) happens once per distinct role. For a role
+// that carries a row-level-security filter, that shared frame is still delivered only
+// to the subscribers whose claims admit this row (evaluated per subscriber); a role
+// without a filter takes the pure once-per-role fast path.
 func (h *Hub) Broadcast(topic string, raw []byte) {
 	// Snapshot the role->bucket set under the read lock; do the unmarshal / project
 	// / serialize outside it. Skip the decode entirely when nobody is listening.
@@ -125,13 +133,43 @@ func (h *Hub) Broadcast(topic string, raw []byte) {
 	decoded := json.Unmarshal(raw, &evt) == nil && evt.TableName != ""
 	p, filter := h.snapshotPolicy()
 
+	// Column types for numeric-aware row-filter comparison — resolved lazily at most
+	// once per event, only when some role actually carries a row-filter, and reused
+	// across every filtered role and subscriber.
+	var numericCols map[string]bool
+	numericResolved := false
+
 	for _, rb := range roleBuckets {
-		wire, ok := project(p, filter, rb.role, &evt, raw, decoded)
+		wire, perms, ok := projectColumns(p, filter, rb.role, &evt, raw, decoded)
 		if !ok {
 			continue // denied table / invalid payload for this role
 		}
 		frame := Frame{Kind: KindEvent, Data: wire}
+
+		if !perms.HasRowFilter() {
+			// No row-level security for this role: one projection serves the whole
+			// bucket, regardless of per-subscriber claims (the #294/#353 fast path).
+			for _, sub := range rb.bucket.Snapshot() {
+				if !sub.Send(frame) {
+					h.metric.FrameDropped(KindEvent)
+				}
+			}
+			continue
+		}
+
+		// Row-level security applies. The column projection is claims-independent and
+		// shared, but whether each subscriber may see THIS row depends on its claims, so
+		// evaluate visibility per subscriber. Predicates read the full event data (a
+		// filter may key on a column the role can't SELECT), not the projected columns.
+		if !numericResolved {
+			numericCols = h.numericCols(evt.TableName)
+			numericResolved = true
+		}
 		for _, sub := range rb.bucket.Snapshot() {
+			subPerms := policy.Evaluate(p, rb.role, evt.TableName, "select", sub.claims)
+			if !subPerms.RowVisible(evt.Data, numericCols) {
+				continue // this row is filtered out for this subscriber
+			}
 			if !sub.Send(frame) {
 				h.metric.FrameDropped(KindEvent)
 			}
@@ -139,6 +177,26 @@ func (h *Hub) Broadcast(topic string, raw []byte) {
 	}
 }
 
+// numericCols maps each of the table's columns to whether its ClickHouse type is
+// numeric, so the row-filter evaluator compares numeric columns numerically (9 < 100)
+// rather than lexicographically. nil when no schema is available (unknown table, or a
+// Hub built without a registry), in which case ordering predicates fall back to
+// lexicographic comparison.
+func (h *Hub) numericCols(table string) map[string]bool {
+	if h.registry == nil {
+		return nil
+	}
+	schema := h.registry.Get(table)
+	if schema == nil {
+		return nil
+	}
+	m := make(map[string]bool, len(schema.Columns))
+	for _, c := range schema.Columns {
+		m[c.Name] = discovery.IsNumericType(c.Type)
+	}
+	return m
+}
+
 // snapshotPolicy returns the current policy and whether filtering is configured.
 // filter is false only when no store is wired (legacy passthrough); a wired store
 // returning a nil policy is a deliberate lockout that Evaluate denies.
@@ -149,50 +207,60 @@ func (h *Hub) snapshotPolicy() (p *policy.Policy, filter bool) {
 	return h.policy.Get(), true
 }
 
-// ReplayFrame projects a single gap-fill event for one role into a ready-to-write
-// replay frame, or ok=false to skip it (denied table / invalid payload). It is a
-// method so replay shares the Hub's policy store with the live fan-out — the
-// handler can't accidentally project replay against a different (or nil) policy.
-// The per-connection replay path uses this; the live path uses Broadcast, which
-// projects once per role instead of once per connection.
-func (h *Hub) ReplayFrame(role string, raw []byte) (Frame, bool) {
+// ReplayFrame projects a single gap-fill event for one role+claims into a
+// ready-to-write replay frame, or ok=false to skip it (denied table, invalid
+// payload, or a row the claims aren't entitled to see). It is a method so replay
+// shares the Hub's policy store and schema registry with the live fan-out — the
+// handler can't accidentally project replay against a different (or nil) policy. The
+// per-connection replay path uses this; the live path uses Broadcast. Replay is
+// already per-connection, so it evaluates row-level security against this
+// connection's claims directly.
+func (h *Hub) ReplayFrame(role string, claims map[string]any, raw []byte) (Frame, bool) {
 	var evt ingest.EventMessage
 	decoded := json.Unmarshal(raw, &evt) == nil && evt.TableName != ""
 	p, filter := h.snapshotPolicy()
-	wire, ok := project(p, filter, role, &evt, raw, decoded)
+	wire, perms, ok := projectColumns(p, filter, role, &evt, raw, decoded)
 	if !ok {
 		return Frame{}, false
 	}
+	if perms.HasRowFilter() {
+		subPerms := policy.Evaluate(p, role, evt.TableName, "select", claims)
+		if !subPerms.RowVisible(evt.Data, h.numericCols(evt.TableName)) {
+			return Frame{}, false // this row is filtered out for these claims
+		}
+	}
 	return Frame{Kind: KindReplay, Data: wire}, true
 }
 
-// project applies role/table column policy to a decoded EventMessage (or passes a
-// non-EventMessage JSON payload through untouched), returning the SSE wire frame.
-// ok=false means skip: the role can't read the table, or the payload is unusable.
-//
-// claims are intentionally not consulted: column visibility derives only from the
-// role+table policy entry (AllowColumns/DenyColumns), so the projection is
-// byte-identical for every subscriber of a (role, table) regardless of claims —
-// which is exactly what lets one serialization serve the whole bucket. If row-level
-// filtering is ever added to the stream path, this key (and projection) must take
-// claims into account.
-func project(p *policy.Policy, filter bool, role string, evt *ingest.EventMessage, raw []byte, decoded bool) (wire []byte, ok bool) {
+// projectColumns applies role/table COLUMN policy to a decoded EventMessage (or
+// passes a non-EventMessage JSON payload through untouched), returning the SSE wire
+// frame and the resolved permissions. Column visibility derives only from the
+// role+table policy entry (AllowColumns/DenyColumns), never from claims, so this
+// frame is byte-identical for every subscriber of a (role, table) — the shared
+// projection that lets one serialization serve a whole bucket. Row-level security is
+// NOT applied here; the caller checks perms.RowVisible per subscriber (with that
+// subscriber's claims) when perms.HasRowFilter(). ok=false means skip: the role can't
+// read the table, or the payload is unusable. perms is nil for the legacy no-policy
+// passthrough (which has no row-filter).
+func projectColumns(p *policy.Policy, filter bool, role string, evt *ingest.EventMessage, raw []byte, decoded bool) (wire []byte, perms *policy.ResolvedPermissions, ok bool) {
 	if !decoded {
 		// Without a decoded EventMessage there's no table to evaluate policy against,
 		// so fail closed whenever policy is configured: a malformed-but-valid-JSON
 		// payload on ingest.<table> must not bypass column filtering. Pass through
 		// only when no policy store is wired at all (the legacy/test passthrough).
 		if filter || !json.Valid(raw) {
-			return nil, false
+			return nil, nil, false
 		}
-		return wireFrame("", raw), true
+		return wireFrame("", raw), nil, true
 	}
 
 	data := evt.Data
 	if filter {
-		perms := policy.Evaluate(p, role, evt.TableName, "select", nil)
+		// Column allow/deny is claims-independent, so evaluate it once with nil claims;
+		// the caller re-evaluates the row-filter per subscriber against real claims.
+		perms = policy.Evaluate(p, role, evt.TableName, "select", nil)
 		if !perms.Allowed {
-			return nil, false // role has no access to this table
+			return nil, nil, false // role has no access to this table
 		}
 		data = filterColumns(evt.Data, perms)
 	}
@@ -203,9 +271,9 @@ func project(p *policy.Policy, filter bool, role string, evt *ingest.EventMessag
 		"data":               data,
 	})
 	if err != nil {
-		return nil, false
+		return nil, nil, false
 	}
-	return wireFrame(evt.ReceivedTimestamp, payload), true
+	return wireFrame(evt.ReceivedTimestamp, payload), perms, true
 }
 
 // filterColumns returns a copy of data containing only columns the role may see.
diff --git a/internal/stream/hub_test.go b/internal/stream/hub_test.go
index 311587fe..f07d92cf 100644
--- a/internal/stream/hub_test.go
+++ b/internal/stream/hub_test.go
@@ -8,6 +8,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/Wave-RF/WaveHouse/internal/discovery"
 	"github.com/Wave-RF/WaveHouse/internal/ingest"
 	"github.com/Wave-RF/WaveHouse/internal/policy"
 	"github.com/stretchr/testify/assert"
@@ -37,6 +38,17 @@ func recvFrame(t *testing.T, sub *Subscriber) Frame {
 	}
 }
 
+// assertNoFrame fails if sub has any frame buffered — used to prove a row-filtered
+// subscriber received nothing for a row it isn't entitled to see.
+func assertNoFrame(t *testing.T, sub *Subscriber) {
+	t.Helper()
+	select {
+	case f := <-sub.Frames():
+		t.Fatalf("expected no frame, got %q", f.Data)
+	default:
+	}
+}
+
 // frameData parses the JSON object on the "data:" line of an SSE frame.
 func frameData(t *testing.T, f Frame) map[string]any {
 	t.Helper()
@@ -53,7 +65,7 @@ func frameData(t *testing.T, f Frame) map[string]any {
 
 func TestHub_ProjectsOncePerRole_FanOutToAllSubscribers(t *testing.T) {
 	t.Parallel()
-	hub := NewHub(nil, nil) // nil store ⇒ passthrough, no filtering
+	hub := NewHub(nil, nil, nil) // nil store ⇒ passthrough, no filtering
 	const topic = "ingest.clicks"
 
 	a, b := NewSubscriber(), NewSubscriber()
@@ -85,7 +97,7 @@ func TestHub_ProjectsPerRole_ColumnFilterAndDenial(t *testing.T) {
 			},
 		},
 	}
-	hub := NewHub(policy.NewMemoryStore(p), nil)
+	hub := NewHub(policy.NewMemoryStore(p), nil, nil)
 	const topic = "ingest.clicks"
 
 	viewer := NewSubscriber()
@@ -121,7 +133,7 @@ func TestHub_ProjectsPerRole_DistinctRolesGetDistinctFrames(t *testing.T) {
 			},
 		},
 	}
-	hub := NewHub(policy.NewMemoryStore(p), nil)
+	hub := NewHub(policy.NewMemoryStore(p), nil, nil)
 	const topic = "ingest.clicks"
 
 	viewer, editor := NewSubscriber(), NewSubscriber()
@@ -148,9 +160,132 @@ func TestHub_ProjectsPerRole_DistinctRolesGetDistinctFrames(t *testing.T) {
 	assert.NotEqual(t, fv.Data, fe.Data, "distinct role projections produce distinct bytes")
 }
 
+// rowFilterPolicy scopes role "viewer" to column "page" only, and to rows whose
+// tenant_id equals the caller's {{ jwt.tenant }} claim. The filter keys on tenant_id
+// — a column viewer may NOT select — so it also exercises the rule that row
+// visibility is evaluated against the FULL event, then columns are projected.
+func rowFilterPolicy() *policy.Policy {
+	return &policy.Policy{
+		Tables: map[string]policy.TablePolicy{
+			"clicks": {
+				Select: map[string]policy.RolePermissions{
+					"viewer": {
+						AllowColumns: []string{"page"},
+						Filter:       map[string]policy.Filter{"tenant_id": {Eq: new("{{ jwt.tenant }}")}},
+					},
+				},
+			},
+		},
+	}
+}
+
+// TestHub_RowFilter_PerSubscriberIsolation is the #319 fix: two subscribers of the
+// SAME role but different tenant claims each receive only their own tenant's rows
+// over the live stream — the row-filter the query path applies is now applied here
+// too, per subscriber.
+func TestHub_RowFilter_PerSubscriberIsolation(t *testing.T) {
+	t.Parallel()
+	hub := NewHub(policy.NewMemoryStore(rowFilterPolicy()), nil, nil)
+	const topic = "ingest.clicks"
+
+	acme := NewSubscriber()
+	acme.SetClaims(map[string]any{"tenant": "acme"})
+	globex := NewSubscriber()
+	globex.SetClaims(map[string]any{"tenant": "globex"})
+	hub.Add(topic, "viewer", acme)
+	hub.Add(topic, "viewer", globex)
+
+	// An acme row reaches only the acme subscriber, projected to the allowed column.
+	hub.Broadcast(topic, rawEvent(t, "clicks", "2026-06-26T00:00:00Z",
+		map[string]any{"tenant_id": "acme", "page": "/a", "secret": "x"}))
+	inner := frameData(t, recvFrame(t, acme))["data"].(map[string]any)
+	assert.Equal(t, "/a", inner["page"])
+	assert.NotContains(t, inner, "tenant_id", "the filtered column is not in viewer's projection")
+	assert.NotContains(t, inner, "secret", "denied column stripped")
+	assertNoFrame(t, globex)
+
+	// A globex row reaches only the globex subscriber.
+	hub.Broadcast(topic, rawEvent(t, "clicks", "2026-06-26T00:00:01Z",
+		map[string]any{"tenant_id": "globex", "page": "/g"}))
+	assert.Equal(t, "/g", frameData(t, recvFrame(t, globex))["data"].(map[string]any)["page"])
+	assertNoFrame(t, acme)
+}
+
+// TestHub_RowFilter_MissingColumn_FailsClosed: an event that lacks the filtered
+// column can't be proven visible, so it is withheld rather than leaked.
+func TestHub_RowFilter_MissingColumn_FailsClosed(t *testing.T) {
+	t.Parallel()
+	hub := NewHub(policy.NewMemoryStore(rowFilterPolicy()), nil, nil)
+	const topic = "ingest.clicks"
+
+	acme := NewSubscriber()
+	acme.SetClaims(map[string]any{"tenant": "acme"})
+	hub.Add(topic, "viewer", acme)
+
+	hub.Broadcast(topic, rawEvent(t, "clicks", "2026-06-26T00:00:00Z",
+		map[string]any{"page": "/a"})) // no tenant_id
+	assertNoFrame(t, acme)
+}
+
+// TestHub_RowFilter_SharedProjectionAcrossSameClaims: the column projection is still
+// serialized once per role and shared — two subscribers with the same (matching)
+// claims receive the identical frame bytes. Only the visibility decision is
+// per-subscriber, not the serialization.
+func TestHub_RowFilter_SharedProjectionAcrossSameClaims(t *testing.T) {
+	t.Parallel()
+	hub := NewHub(policy.NewMemoryStore(rowFilterPolicy()), nil, nil)
+	const topic = "ingest.clicks"
+
+	a, b := NewSubscriber(), NewSubscriber()
+	a.SetClaims(map[string]any{"tenant": "acme"})
+	b.SetClaims(map[string]any{"tenant": "acme"})
+	hub.Add(topic, "viewer", a)
+	hub.Add(topic, "viewer", b)
+
+	hub.Broadcast(topic, rawEvent(t, "clicks", "2026-06-26T00:00:00Z",
+		map[string]any{"tenant_id": "acme", "page": "/a"}))
+
+	fa, fb := recvFrame(t, a), recvFrame(t, b)
+	require.NotEmpty(t, fa.Data)
+	require.NotEmpty(t, fb.Data)
+	assert.Same(t, &fa.Data[0], &fb.Data[0], "one serialization shared across same-role subscribers")
+}
+
+// TestHub_RowFilter_NumericOrdering_SchemaInformed drives the registry-backed path:
+// with a numeric column type in the schema, an `amount > 100` filter compares
+// numerically, so amount=9 is withheld (a lexicographic "9" > "100" comparison would
+// have leaked it) and amount=250 is delivered.
+func TestHub_RowFilter_NumericOrdering_SchemaInformed(t *testing.T) {
+	t.Parallel()
+	reg := discovery.NewSchemaRegistryFromMap([]*discovery.TableSchema{
+		{Name: "clicks", Columns: []discovery.Column{
+			{Name: "amount", Type: "UInt64"},
+			{Name: "page", Type: "String"},
+		}},
+	})
+	p := &policy.Policy{
+		Tables: map[string]policy.TablePolicy{
+			"clicks": {Select: map[string]policy.RolePermissions{
+				"viewer": {Filter: map[string]policy.Filter{"amount": {Gt: new("100")}}},
+			}},
+		},
+	}
+	hub := NewHub(policy.NewMemoryStore(p), reg, nil)
+	const topic = "ingest.clicks"
+
+	sub := NewSubscriber() // constant filter value ⇒ no claims needed
+	hub.Add(topic, "viewer", sub)
+
+	hub.Broadcast(topic, rawEvent(t, "clicks", "t1", map[string]any{"amount": float64(9), "page": "/a"}))
+	assertNoFrame(t, sub) // 9 is not numerically > 100
+
+	hub.Broadcast(topic, rawEvent(t, "clicks", "t2", map[string]any{"amount": float64(250), "page": "/b"}))
+	assert.Equal(t, float64(250), frameData(t, recvFrame(t, sub))["data"].(map[string]any)["amount"])
+}
+
 func TestHub_TopicIsolation(t *testing.T) {
 	t.Parallel()
-	hub := NewHub(nil, nil)
+	hub := NewHub(nil, nil, nil)
 	clicks, views := NewSubscriber(), NewSubscriber()
 	hub.Add("ingest.clicks", "public", clicks)
 	hub.Add("ingest.views", "public", views)
@@ -193,7 +328,7 @@ func TestHub_PassthroughAndFailClosed(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			t.Parallel()
-			hub := NewHub(tt.store, nil)
+			hub := NewHub(tt.store, nil, nil)
 			const topic = "ingest.custom"
 			sub := NewSubscriber()
 			hub.Add(topic, "public", sub)
@@ -214,7 +349,7 @@ func TestHub_PassthroughAndFailClosed(t *testing.T) {
 
 func TestHub_AddRemoveGCsBucketsAndTopics(t *testing.T) {
 	t.Parallel()
-	hub := NewHub(nil, nil)
+	hub := NewHub(nil, nil, nil)
 	const topic = "ingest.clicks"
 	sub := NewSubscriber()
 
@@ -235,7 +370,7 @@ func TestHub_AddRemoveGCsBucketsAndTopics(t *testing.T) {
 
 func TestHub_BroadcastNoSubscribers_NoOp(t *testing.T) {
 	t.Parallel()
-	hub := NewHub(nil, nil)
+	hub := NewHub(nil, nil, nil)
 	assert.NotPanics(t, func() {
 		hub.Broadcast("ingest.nobody", rawEvent(t, "clicks", "t", map[string]any{"a": float64(1)}))
 	})
@@ -252,7 +387,7 @@ func TestHub_SlowConsumerDropIncrementsMetric(t *testing.T) {
 		otel.SetMeterProvider(savedMP)
 	})
 
-	hub := NewHub(nil, NewMetrics())
+	hub := NewHub(nil, nil, NewMetrics())
 	const topic = "ingest.clicks"
 	sub := newSubscriber(1) // cap-1: the second undrained broadcast drops
 	hub.Add(topic, "public", sub)
@@ -273,7 +408,7 @@ func TestHub_ReplayFrame(t *testing.T) {
 			"clicks": {Select: map[string]policy.RolePermissions{"viewer": {AllowColumns: []string{"page"}}}},
 		},
 	}
-	hub := NewHub(policy.NewMemoryStore(p), nil)
+	hub := NewHub(policy.NewMemoryStore(p), nil, nil)
 	raw := rawEvent(t, "clicks", "2026-06-26T00:00:00Z", map[string]any{"page": "/home", "secret": "x"})
 
 	tests := []struct {
@@ -287,7 +422,7 @@ func TestHub_ReplayFrame(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			t.Parallel()
-			f, ok := hub.ReplayFrame(tt.role, raw)
+			f, ok := hub.ReplayFrame(tt.role, nil, raw)
 			require.Equal(t, tt.want, ok)
 			if !tt.want {
 				return
@@ -300,9 +435,35 @@ func TestHub_ReplayFrame(t *testing.T) {
 	}
 }
 
+// TestHub_ReplayFrame_RowFilter exercises the row-filter branch of ReplayFrame: the
+// #319 fix applies row-level security on the per-connection replay path too, so a
+// gap-fill event is projected only when the connection's claims satisfy the filter.
+func TestHub_ReplayFrame_RowFilter(t *testing.T) {
+	t.Parallel()
+	hub := NewHub(policy.NewMemoryStore(rowFilterPolicy()), nil, nil)
+	raw := rawEvent(t, "clicks", "2026-06-26T00:00:00Z",
+		map[string]any{"tenant_id": "acme", "page": "/a", "secret": "x"})
+
+	t.Run("matching claims replay the row, projected to allowed columns", func(t *testing.T) {
+		t.Parallel()
+		f, ok := hub.ReplayFrame("viewer", map[string]any{"tenant": "acme"}, raw)
+		require.True(t, ok)
+		assert.Equal(t, KindReplay, f.Kind)
+		inner := frameData(t, f)["data"].(map[string]any)
+		assert.Equal(t, "/a", inner["page"])
+		assert.NotContains(t, inner, "secret", "denied column stripped on replay too")
+	})
+
+	t.Run("non-matching claims withhold the row", func(t *testing.T) {
+		t.Parallel()
+		_, ok := hub.ReplayFrame("viewer", map[string]any{"tenant": "globex"}, raw)
+		require.False(t, ok, "row must be withheld when claims don't satisfy the filter")
+	})
+}
+
 func TestHub_ConcurrentAddRemoveBroadcast_Race(t *testing.T) {
 	t.Parallel()
-	hub := NewHub(nil, nil)
+	hub := NewHub(nil, nil, nil)
 	const topic = "ingest.clicks"
 	raw := rawEvent(t, "clicks", "t", map[string]any{"a": float64(1)})
 
diff --git a/internal/stream/subscriber.go b/internal/stream/subscriber.go
index 253980e5..3a22b909 100644
--- a/internal/stream/subscriber.go
+++ b/internal/stream/subscriber.go
@@ -23,6 +23,13 @@ type Subscriber struct {
 	out    chan Frame // ready-to-write frames; see defaultSubscriberQueue
 	bucket Bucket     // set on Add so the keepalive wheel's Remove is O(1); nil until added
 
+	// claims are this connection's JWT claims, evaluated against a role's row-level
+	// security filter so the Hub can decide, per subscriber, whether each event row is
+	// visible (see Hub.Broadcast). Set once via SetClaims before Add, then read-only —
+	// so the fan-out goroutine reads it without synchronization. nil ⇒ no claims (a
+	// tokenless subscriber), which fails any claim-scoped row-filter closed.
+	claims map[string]any
+
 	// evict is closed (once) to ask the owning handler to disconnect a wedged slow
 	// consumer; the handler selects on Evicted() and tears the stream down, after
 	// which the client reconnects and gap-fills via Last-Event-ID. The seam is wired
@@ -41,6 +48,12 @@ func newSubscriber(size int) *Subscriber {
 	return &Subscriber{out: make(chan Frame, size), evict: make(chan struct{})}
 }
 
+// SetClaims attaches this connection's JWT claims, which the Hub evaluates against a
+// role's row-level-security filter to decide, per subscriber, whether each event row
+// is visible. Call it before registering the subscriber with the Hub; the claims are
+// then read-only for the subscriber's lifetime, so the fan-out reads them race-free.
+func (s *Subscriber) SetClaims(claims map[string]any) { s.claims = claims }
+
 // Frames is the queue of ready-to-write frames; the handler writes whatever
 // arrives here to the client verbatim.
 func (s *Subscriber) Frames() <-chan Frame {
diff --git a/tests/e2e/sdk/streaming.test.ts b/tests/e2e/sdk/streaming.test.ts
index c6b4a5ea..2a1c9314 100644
--- a/tests/e2e/sdk/streaming.test.ts
+++ b/tests/e2e/sdk/streaming.test.ts
@@ -1,6 +1,13 @@
 import type { Policy } from "@wavehouse/sdk";
 import { afterAll, beforeAll, describe, expect, it } from "vitest";
-import { adminClient, dataClient, publicClient, testId, waitForCondition } from "./helpers.js";
+import {
+  adminClient,
+  authClient,
+  dataClient,
+  publicClient,
+  testId,
+  waitForCondition,
+} from "./helpers.js";
 import { suiteTables } from "./tables.js";
 
 describe("Streaming", () => {
@@ -18,10 +25,18 @@ describe("Streaming", () => {
     const publicPolicy = structuredClone(baselinePolicy);
     publicPolicy.default_role = "anon";
 
-    // Explicitly allow the 'anon' role to SELECT (stream) from this suite's tables
+    // Explicitly allow the 'anon' role to SELECT (stream) from this suite's tables.
+    // 'scoped' additionally carries a per-subscriber row filter — streamed rows are
+    // limited to the caller's own country claim — so the SSE fan-out exercises the
+    // row-level-security path (ResolvedPermissions.RowVisible) end to end, not just
+    // column projection.
     publicPolicy.tables[T.clicks].select = {
       ...(publicPolicy.tables[T.clicks].select || {}),
       anon: { allow_columns: ["*"] },
+      scoped: {
+        allow_columns: ["*"],
+        filter: { country: { _eq: "{{ jwt.country }}" } },
+      },
     };
     publicPolicy.tables[T.events].select = {
       ...(publicPolicy.tables[T.events].select || {}),
@@ -115,5 +130,55 @@ describe("Streaming", () => {
         stream.close();
       }
     });
+
+    it("applies the role's row filter per subscriber (row-level scoping)", async () => {
+      // Two subscribers, same 'scoped' role but different country claims. The role's
+      // filter (country = {{ jwt.country }}) must be evaluated per subscriber against
+      // the full event, so each sees only its own country's rows over the live stream —
+      // the #319 row-level-security guarantee, exercised end to end on SSE.
+      const inserter = dataClient(); // viewer may insert into this suite's tables
+      const usClient = authClient("scoped", { country: "US" });
+      const caClient = authClient("scoped", { country: "CA" });
+      const usEvents: any[] = [];
+      const caEvents: any[] = [];
+      const usId = testId();
+      const caId = testId();
+
+      const usStream = usClient.from(T.clicks).stream();
+      const caStream = caClient.from(T.clicks).stream();
+      let unsubUs: (() => void) | undefined;
+      let unsubCa: (() => void) | undefined;
+      try {
+        unsubUs = usStream.subscribe({
+          next: (e) => usEvents.push(e),
+          error: (err) => console.error("US SSE error:", err),
+        });
+        unsubCa = caStream.subscribe({
+          next: (e) => caEvents.push(e),
+          error: (err) => console.error("CA SSE error:", err),
+        });
+        await usStream.connected(20_000);
+        await caStream.connected(20_000);
+
+        // One row per country; the filter must route each to only its matching subscriber.
+        const base = { page: "/scoped", user_id: "u", session_id: "s", duration_ms: 1 };
+        await inserter.from(T.clicks).insert({ ...base, event_id: usId, country: "US" });
+        await inserter.from(T.clicks).insert({ ...base, event_id: caId, country: "CA" });
+
+        // Each subscriber receives its own country's row. Waiting for BOTH positives
+        // proves both rows were broadcast, so the cross-absence checks below are real
+        // (a row filtered out at the source can never arrive later).
+        await waitForCondition(() => usEvents.some((e) => e.data?.event_id === usId), 10_000);
+        await waitForCondition(() => caEvents.some((e) => e.data?.event_id === caId), 10_000);
+
+        expect(usEvents.some((e) => e.data?.event_id === caId)).toBe(false);
+        expect(caEvents.some((e) => e.data?.event_id === usId)).toBe(false);
+      } finally {
+        if (unsubUs) unsubUs();
+        if (unsubCa) unsubCa();
+        usStream.close();
+        caStream.close();
+      }
+    });
   });
 });
diff --git a/tests/integration/setup_test.go b/tests/integration/setup_test.go
index bb3aecdb..348cabda 100644
--- a/tests/integration/setup_test.go
+++ b/tests/integration/setup_test.go
@@ -312,7 +312,7 @@ func buildServer(ch *chInstance, embeddedMQ *mq.EmbeddedNATS, registry *discover
 	js := embeddedMQ.JetStream()
 
 	policyStore := policy.NewMemoryStore(&policy.Policy{AdminRole: "admin"})
-	streamHub := stream.NewHub(policyStore, nil)
+	streamHub := stream.NewHub(policyStore, registry, nil)
 
 	deps := api.Dependencies{
 		Ingest: api.NewIngestHandler(registry, embeddedMQ, logger),