diff --git a/CHANGELOG.md b/CHANGELOG.md
index a17029c1..f5ee87b6 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -7,14 +7,27 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## Unreleased
 
-### Changed
-
-- **Boot-time schema discovery no longer crashes the binary** (`cmd/wavehouse/main.go`, `internal/api/health.go`, `internal/discovery/discovery.go`, `docs/src/content/docs/api.md`, `docs/src/content/docs/deployment.md`). Previously, any error from the initial `SchemaRegistry.Refresh` — connection-refused, missing database, transient network blip — caused `cmd/wavehouse` to call `os.Exit(1)`. The supervisor would restart the process every ~10s in an unbounded loop, port 8080 never bound, and operators got `connection refused` on probes even when ClickHouse was otherwise healthy. Now the first Refresh failure is non-fatal: a new `api.BootState` is set with the diagnostic, the server still binds `:8080`, and a background goroutine calls the new `SchemaRegistry.RetryRefresh(ctx, 2s, 60s, onAttempt)` with exponential backoff until success or shutdown. While `BootState.Err()` is non-nil, `/health` returns 503 with `{"status":"degraded","error":"…"}` so an operator can `curl /health` to learn why the gateway isn't ready to serve traffic instead of grepping a restart-loop log; `/ready` returns 503 with `{"status":"not ready",…}` for the same reason. Once `BootState.Set(nil)` fires (either from the initial sync `Refresh` or from `SchemaRegistry.RetryRefresh` success), `/health` stays at `200 OK` for the rest of the process lifetime — that flip is sticky and reflects "boot completed once." `/ready`, by contrast, remains conditional on current ClickHouse reachability: it can flip back to `503` on transient runtime blips after boot, since readiness is a probe of "can I serve traffic right now," not "did I ever boot." Resolves #95.
-
 ### Removed
+
+- **Hub wildcard subscriptions** (`internal/api/hub.go`, `internal/api/hub_test.go`): dropped the NATS-style `*` / `>` pattern matching from `Hub.Broadcast`, the wildcard pattern loop, the `sent` dedup map, the `matchTopic` helper, and the eight wildcard tests (plus `TestMatchTopic`). After the #89 MVP cuts every producer publishes a concrete `ingest.<table>` subject and the SDK only ever subscribes to one concrete subject, so the wildcard fan-out was unused machinery. Closes #100 (part of #87). Net −210 lines (mostly tests).
 - **`project-orchestrator.yml` workflow + its three composite-action artifacts** (`.github/workflows/project-orchestrator.yml`, `.github/actions/board-upsert-status/`, `.github/actions/set-linked-issues-status/`, `.github/scripts/board-fetch-item.sh`, `AGENTS.md`, `CHANGELOG.md`): −887 lines net. The orchestrator was the largest single source of cross-trigger complexity on this repo (3-4 workflow_run-chained runs per PR push, `statusCheckRollup` GraphQL perms quirks, integration-token `NONE` for private-org members) for behaviour that is mostly either provided natively by GitHub or a one-click manual operation on a 4-person team. Replaced by: reviewer-assign step in `housekeeping.yml` that fires once on `pull_request_target: opened` / `ready_for_review` (not per-synchronize, so it doesn't re-spam after `dismiss_stale_reviews_on_push`), plus GitHub's native Projects v2 workflows (`Auto-add to project`, `Item added`, `Pull request merged`) configured in the project UI. Trade-offs explicit in the PR body: drafts no longer auto-flip on bot-clean, `CHANGES_REQUESTED` doesn't auto-move the board card, linked-issue card mirroring is dropped. AGENTS.md §"Governance Files" + §"Task Board state machine" + §"Review tooling reference" all rewritten to match. `dependabot-automerge.yml` trimmed in parallel: no more board-upsert step (native handles placement), `PROJECT_BOARD_TOKEN` guard removed (no longer used in this workflow), reviewer list sourced from `board-config.env`'s `ADMINS` via `replace()`, major-bump comment uses the marker-comment upsert pattern from `housekeeping.yml`.
 - **`STATUS_*` and old `ADMINS` consumers in `board-config.env`** — STATUS option IDs had only orchestrator-side consumers and are now unreferenced. `ADMINS` was restored to `board-config.env` after the initial orchestrator-removal commit dropped it (Gemini and Claude both flagged the resulting drift across three inlined copies); both `housekeeping.yml` and `dependabot-automerge.yml` now load `ADMINS` from `board-config.env`. `admin-approval.yml` keeps its own inline copy with the documented latency-avoidance reasoning.
 
+### Changed
+
+- **BREAKING:** SSE/WS streaming endpoints replaced `?topic=` with `?table=` (issue #100, option B). The NATS subject convention no longer leaks into the public HTTP API; the server builds `ingest.<table>` internally. SSE returns `400 Bad Request` when `?table=` is missing or fails the `^[a-zA-Z_][a-zA-Z0-9_]*$` safe-identifier check (the same regex `internal/ingest` and `internal/query` use for SQL identifiers — crucially this rejects NATS wildcard characters `*` and `>` before they reach the gap-fill `FilterSubject`). WS `?table=` is optional — clients can still defer to in-band subscribe commands — but is validated by the same regex when present.
+- **BREAKING:** WebSocket in-band subscribe/unsubscribe commands use `"table"` (raw name) instead of `"topic"` (NATS subject):
+  - Before: `{"action":"subscribe","topic":"ingest.clicks"}`
+  - After:  `{"action":"subscribe","table":"clicks"}`
+- **BREAKING:** WebSocket outbound envelope is now labelled `"table"` (raw name) instead of `"topic"` (NATS subject):
+  - Before: `{"topic":"ingest.clicks","data":{"table_name":"clicks",...}}`
+  - After:  `{"table":"clicks","data":{"table_name":"clicks",...}}`
+- TypeScript SDK (`@wavehouse/sdk`) updated to the new contract: `SharedWSManager` is now keyed by table name (one less string concat per dispatch), `SSEOptions` / `WSOptions` carry `table` instead of `topic`, wildcard dispatch loop + `matchTopicPattern` helper removed from `ws-manager.ts`. The high-level SDK API (`wh.from('clicks').stream()`) is unchanged.
+- WebSocket `?since=` gap-fill now accepts `RFC3339Nano` (with fractional seconds) in addition to `RFC3339`, matching the SSE handler's existing acceptance of high-precision client timestamps.
+- WebSocket internal dispatch carries the subscribing table name through the merged channel instead of re-unmarshalling each payload to extract `table_name`. Drops the per-message redundant unmarshal and — for non-`EventMessage` payloads (raw-JSON pass-through) — keeps the outbound envelope's `"table"` field labelled with the actual subscription instead of falling back to `""`. No behaviour change for the standard `EventMessage` path.
+- Doc-sync for the `?topic=` → `?table=` rename: `docs/src/content/docs/getting-started.md` quick-start now shows `?table=clicks` (the previous `?topic=ingest.clicks` example would 400 against the new contract), and `docs/src/content/docs/sdk.md` `SharedWSManager` section now describes per-table (not per-topic) subscriptions and notes that the legacy NATS-style wildcards are no longer accepted server-side.
+- **Boot-time schema discovery no longer crashes the binary** (`cmd/wavehouse/main.go`, `internal/api/health.go`, `internal/discovery/discovery.go`, `docs/src/content/docs/api.md`, `docs/src/content/docs/deployment.md`). Previously, any error from the initial `SchemaRegistry.Refresh` — connection-refused, missing database, transient network blip — caused `cmd/wavehouse` to call `os.Exit(1)`. The supervisor would restart the process every ~10s in an unbounded loop, port 8080 never bound, and operators got `connection refused` on probes even when ClickHouse was otherwise healthy. Now the first Refresh failure is non-fatal: a new `api.BootState` is set with the diagnostic, the server still binds `:8080`, and a background goroutine calls the new `SchemaRegistry.RetryRefresh(ctx, 2s, 60s, onAttempt)` with exponential backoff until success or shutdown. While `BootState.Err()` is non-nil, `/health` returns 503 with `{"status":"degraded","error":"…"}` so an operator can `curl /health` to learn why the gateway isn't ready to serve traffic instead of grepping a restart-loop log; `/ready` returns 503 with `{"status":"not ready",…}` for the same reason. Once `BootState.Set(nil)` fires (either from the initial sync `Refresh` or from `SchemaRegistry.RetryRefresh` success), `/health` stays at `200 OK` for the rest of the process lifetime — that flip is sticky and reflects "boot completed once." `/ready`, by contrast, remains conditional on current ClickHouse reachability: it can flip back to `503` on transient runtime blips after boot, since readiness is a probe of "can I serve traffic right now," not "did I ever boot." Resolves #95.
+
 ### Fixed
 - **Ingest worker no longer infinite-retries permanent delete errors** (`internal/ingest/bento.go`, `internal/ingest/bento_test.go`, `docs/src/content/docs/architecture.md`, `AGENTS.md`): `jsInput.Read`'s `action: "delete"` block called `m.Nak()` on every `chConn.Exec` failure, which JetStream interprets as "redeliver immediately." A delete whose error was *deterministic* (syntax error, unknown table, malformed identifier) would loop forever — clogging the buffer consumer, burning CPU, and spamming logs with the same message. Phase 1 of issue #91: every delete-Exec error is now treated as permanent. The original NATS envelope is published to `dlq.<table>` (reusing the existing `bentoDLQDropped` counter when the DLQ publish itself fails) and the message is `DoubleAck`'d so it leaves the main queue. Issue #91 stays open after this lands as the Phase 2 tracker for transient-vs-permanent error classification (timeouts and network errors should still `Nak()` for retry); Phase 1 alone is the stopgap, Phase 2 makes the trade-off acceptable in production.
 - **CORS middleware: spec-compliant wildcard, no credentials, no header decoration on same-origin** (`internal/api/router.go`, `internal/api/router_test.go`, `AGENTS.md`, `config.yaml`, `docs/src/content/docs/configuration.md`, `docs/src/content/docs/deployment.md`): closes [#29](https://github.com/Wave-RF/WaveHouse/issues/29) and bookends [#30](https://github.com/Wave-RF/WaveHouse/issues/30). Three behavior changes, one rationale. (1) Dropped `Access-Control-Allow-Credentials: true` entirely — WaveHouse is a Bearer-token API (`Authorization: Bearer <jwt>`), cookies are never used (verified: no `http.Cookie` / `SetCookie` anywhere in the Go tree, TS SDK sends no `credentials: 'include'`), so credentials mode is unnecessary AND the previous combination of `Allow-Credentials: true` with `Allow-Origin: *` violated the CORS spec — browsers reject that pairing, which silently broke any client that ever set `credentials: 'include'`. (2) Requests with no `Origin` header (same-origin browser navigation, server-to-server, curl) now skip the CORS decoration entirely instead of unconditionally stamping `Allow-Methods`/`Allow-Headers`/`Allow-Credentials` on every response. (3) A preflight from a disallowed origin still returns 204 but with no CORS headers, which the browser treats as preflight failure — same outcome as before but without leaking the methods/headers list to origins that aren't allowed. Allowlist mode sets `Vary: Origin` on both hits *and* rejects so shared caches can't memoize a headerless reject under the URL alone and replay it to a later allowed-origin request. Test coverage in `router_test.go` pins each branch: wildcard echoes `*`, allowlist hit echoes origin + `Vary: Origin`, allowlist miss gets no `Allow-Origin` but still gets `Vary: Origin` (both for regular and OPTIONS), no-Origin requests pass through clean, and a table-driven test asserts `Allow-Credentials` is never emitted across wildcard / empty-allowlist / allowlist-hit. Posture is documented as `AGENTS.md` §"Key Design Decisions" item 16 so future contributors don't reintroduce credentials or cookie auth without a design discussion. Config sample updated with the dev recipe (point at `http://localhost:3000` etc. instead of `*` once a frontend is built).
diff --git a/README.md b/README.md
index a437c934..9e7b15a7 100644
--- a/README.md
+++ b/README.md
@@ -75,8 +75,8 @@ curl -s -X POST http://localhost:8080/v1/query \
   -H "Content-Type: application/json" \
   -d '{"sql": "SELECT * FROM clicks LIMIT 10"}'
 
-# Open a real-time SSE stream (Ctrl+C to stop)
-curl -N http://localhost:8080/v1/stream/sse
+# Open a real-time SSE stream for a specific table (Ctrl+C to stop)
+curl -N "http://localhost:8080/v1/stream/sse?table=clicks"
 ```
 
 WaveHouse is now accepting API requests on `http://localhost:8080`.
diff --git a/clients/ts/src/client.ts b/clients/ts/src/client.ts
index 4cd144ea..774f40a9 100644
--- a/clients/ts/src/client.ts
+++ b/clients/ts/src/client.ts
@@ -81,13 +81,12 @@ export class WaveHouseClient<DB extends Database = Database> {
     return sql<Row>(this._ctx, query, params, opts);
   }
 
-  /** @internal Create a stream for the given table/topic. */
+  /** @internal Create a stream for the given table. */
   private _createStream<T = Record<string, unknown>>(
     table: string,
     opts?: StreamOptions,
   ): StreamController<T> {
     const transportType = opts?.transport ?? this._config.transport ?? 'auto';
-    const topic = `ingest.${table}`;
 
     // The Smart 'auto' Logic
     let useWS = transportType === 'ws';
@@ -126,7 +125,7 @@ export class WaveHouseClient<DB extends Database = Database> {
         connect() {
           // Subscribe to the manager; forward events to the transport callbacks.
           const unsub = mgr.subscribe<T>(
-            topic,
+            table,
             (event) => this.onEvent?.(event),
             (status) => this.onStatus?.(status),
             (error) => this.onError?.(error),
@@ -154,7 +153,7 @@ export class WaveHouseClient<DB extends Database = Database> {
 
     const transport = new SSETransport<T>({
       baseURL: this._ctx.baseURL,
-      topic,
+      table,
       since: opts?.since,
     });
     const controller = new StreamController<T>(transport);
diff --git a/clients/ts/src/stream/sse.ts b/clients/ts/src/stream/sse.ts
index 02bbe000..8f22bd19 100644
--- a/clients/ts/src/stream/sse.ts
+++ b/clients/ts/src/stream/sse.ts
@@ -3,7 +3,7 @@ import type { StreamTransport } from './controller.js';
 
 export interface SSEOptions {
   baseURL: string;
-  topic: string;
+  table: string;
   since?: string;
 }
 
@@ -33,7 +33,7 @@ export class SSETransport<T = Record<string, unknown>> implements StreamTranspor
     }
 
     const url = new URL('/v1/stream/sse', this._opts.baseURL);
-    url.searchParams.set('topic', this._opts.topic);
+    url.searchParams.set('table', this._opts.table);
     if (this._opts.since) {
       url.searchParams.set('since', this._opts.since);
     }
diff --git a/clients/ts/src/stream/ws-manager.ts b/clients/ts/src/stream/ws-manager.ts
index ad89d2d2..3bc66370 100644
--- a/clients/ts/src/stream/ws-manager.ts
+++ b/clients/ts/src/stream/ws-manager.ts
@@ -14,14 +14,14 @@ interface Subscription {
  * Manages a single multiplexed WebSocket connection.
  *
  * Instead of one WebSocket per stream, all subscriptions share a single
- * connection. Topics are subscribed/unsubscribed via in-band JSON commands:
+ * connection. Tables are subscribed/unsubscribed via in-band JSON commands:
  *
- *   {"action":"subscribe","topic":"ingest.clicks"}
- *   {"action":"unsubscribe","topic":"ingest.clicks"}
+ *   {"action":"subscribe","table":"clicks"}
+ *   {"action":"unsubscribe","table":"clicks"}
  *
- * Incoming messages have a topic envelope:
+ * Incoming messages carry a table envelope:
  *
- *   {"topic":"ingest.clicks","data":{"table_name":"clicks",...}}
+ *   {"table":"clicks","data":{"table_name":"clicks",...}}
  */
 export class SharedWSManager {
   private _baseURL: string;
@@ -32,7 +32,6 @@ export class SharedWSManager {
   private _reconnectTimer: ReturnType<typeof setTimeout> | null = null;
   private _closed = false;
   private _connected = false;
-  private _pendingCommands: string[] = [];
 
   constructor(baseURL: string, auth?: () => Promise<string> | string) {
     this._baseURL = baseURL;
@@ -40,25 +39,25 @@ export class SharedWSManager {
   }
 
   /**
-   * Subscribe to a topic. Opens the WebSocket if not already connected.
+   * Subscribe to a table. Opens the WebSocket if not already connected.
    * Returns an unsubscribe function.
    */
   subscribe<T = Record<string, unknown>>(
-    topic: string,
+    table: string,
     callback: WSEventCallback<T>,
     onStatus?: WSStatusCallback,
     onError?: WSErrorCallback,
   ): () => void {
     const sub: Subscription = { callback, onStatus, onError };
 
-    let topicSubs = this._subs.get(topic);
-    const isNewTopic = !topicSubs || topicSubs.size === 0;
+    let tableSubs = this._subs.get(table);
+    const isNewTable = !tableSubs || tableSubs.size === 0;
 
-    if (!topicSubs) {
-      topicSubs = new Set();
-      this._subs.set(topic, topicSubs);
+    if (!tableSubs) {
+      tableSubs = new Set();
+      this._subs.set(table, tableSubs);
     }
-    topicSubs.add(sub);
+    tableSubs.add(sub);
 
     // Ensure connection is open.
     if (!this._ws && !this._closed) {
@@ -69,16 +68,31 @@ export class SharedWSManager {
       onStatus?.(this._connected ? 'live' : 'connecting');
     }
 
-    // Send subscribe command for new topics.
-    if (isNewTopic) {
-      this._send(JSON.stringify({ action: 'subscribe', topic }));
+    // Send subscribe frame only if the socket is open. If we're still
+    // connecting (or reconnecting), onopen will reconcile by re-subscribing
+    // every table in this._subs — queuing the frame here would just produce
+    // a duplicate subscribe once the socket opens.
+    if (isNewTable && this._wsOpen()) {
+      this._ws!.send(JSON.stringify({ action: 'subscribe', table }));
     }
 
     return () => {
-      topicSubs!.delete(sub);
-      if (topicSubs!.size === 0) {
-        this._subs.delete(topic);
-        this._send(JSON.stringify({ action: 'unsubscribe', topic }));
+      // Idempotent — look up the *current* set each call rather than the one
+      // captured at subscribe-time. If this closure runs twice (or after a
+      // subsequent subscribe replaced this table's set), the captured
+      // reference would otherwise let the second call delete the wrong entry
+      // from this._subs.
+      const current = this._subs.get(table);
+      if (!current || !current.has(sub)) return;
+      current.delete(sub);
+      if (current.size === 0) {
+        this._subs.delete(table);
+        if (this._wsOpen()) {
+          this._ws!.send(JSON.stringify({ action: 'unsubscribe', table }));
+        }
+        // If not open, onopen only re-subscribes what's still in this._subs —
+        // the just-deleted table is excluded, so no explicit unsubscribe
+        // needs to reach the server.
       }
       // Close connection if no subscriptions remain.
       if (this._subs.size === 0) {
@@ -87,6 +101,10 @@ export class SharedWSManager {
     };
   }
 
+  private _wsOpen(): boolean {
+    return this._ws !== null && this._ws.readyState === WebSocket.OPEN;
+  }
+
   /** Close the WebSocket and release all resources. */
   close(): void {
     this._closed = true;
@@ -131,53 +149,44 @@ export class SharedWSManager {
       this._connected = true;
       this._notifyAllStatus('live');
 
-      // Flush pending commands.
-      for (const cmd of this._pendingCommands) {
-        this._ws?.send(cmd);
-      }
-      this._pendingCommands = [];
-
-      // Re-subscribe all active topics.
-      for (const topic of this._subs.keys()) {
-        this._ws?.send(JSON.stringify({ action: 'subscribe', topic }));
+      // Reconcile server state from this._subs — sends one subscribe per
+      // currently-active table. Sub/unsub frames while the socket was opening
+      // (or reconnecting) intentionally aren't queued; this loop is the
+      // single source of truth for what the server should be streaming.
+      for (const table of this._subs.keys()) {
+        this._ws?.send(JSON.stringify({ action: 'subscribe', table }));
       }
     };
 
     this._ws.onmessage = (e) => {
       try {
+        // Server envelope: {table, data: {table_name, received_timestamp, data}}.
+        // For non-EventMessage (raw-JSON pass-through) payloads, the inner
+        // table_name may be absent — fall back to the envelope's table so the
+        // StreamEvent always carries the subscribing table. envelope.table is
+        // also what we key dispatch by, so the two are guaranteed to agree.
         const envelope = JSON.parse(e.data as string) as {
-          topic: string;
+          table: string;
           data: {
-            table_name: string;
-            received_timestamp: string;
+            table_name?: string;
+            received_timestamp?: string;
             data: unknown;
           };
         };
-        if (!envelope.topic || !envelope.data) return;
+        if (!envelope.table || !envelope.data) return;
 
         const event: StreamEvent = {
-          table: envelope.data.table_name,
-          timestamp: envelope.data.received_timestamp,
+          table: envelope.data.table_name ?? envelope.table,
+          timestamp: envelope.data.received_timestamp ?? '',
           data: envelope.data.data as Record<string, unknown>,
         };
 
-        // Dispatch to exact topic subscribers.
-        const exact = this._subs.get(envelope.topic);
-        if (exact) {
-          for (const sub of exact) {
+        const subs = this._subs.get(envelope.table);
+        if (subs) {
+          for (const sub of subs) {
             sub.callback(event);
           }
         }
-
-        // Dispatch to wildcard subscribers (e.g. "ingest.>").
-        for (const [pattern, subs] of this._subs) {
-          if (pattern === envelope.topic) continue; // already handled
-          if (matchTopicPattern(pattern, envelope.topic)) {
-            for (const sub of subs) {
-              sub.callback(event);
-            }
-          }
-        }
       } catch {
         // ignore malformed messages
       }
@@ -201,14 +210,6 @@ export class SharedWSManager {
     };
   }
 
-  private _send(data: string): void {
-    if (this._ws && this._ws.readyState === WebSocket.OPEN) {
-      this._ws.send(data);
-    } else {
-      this._pendingCommands.push(data);
-    }
-  }
-
   private _scheduleReconnect(): void {
     this._notifyAllStatus('reconnecting');
     const delay = Math.min(1000 * 2 ** this._reconnectAttempt, 30_000);
@@ -234,22 +235,3 @@ export class SharedWSManager {
     }
   }
 }
-
-/**
- * Client-side NATS-style topic matching for dispatching messages.
- *  - `*` matches exactly one token
- *  - `>` as the last token matches one or more tokens
- */
-function matchTopicPattern(pattern: string, subject: string): boolean {
-  const pTokens = pattern.split('.');
-  const sTokens = subject.split('.');
-
-  for (let i = 0; i < pTokens.length; i++) {
-    if (pTokens[i] === '>') {
-      return i < sTokens.length;
-    }
-    if (i >= sTokens.length) return false;
-    if (pTokens[i] !== '*' && pTokens[i] !== sTokens[i]) return false;
-  }
-  return pTokens.length === sTokens.length;
-}
diff --git a/clients/ts/src/stream/ws.ts b/clients/ts/src/stream/ws.ts
index f5c3cc0e..66f936d5 100644
--- a/clients/ts/src/stream/ws.ts
+++ b/clients/ts/src/stream/ws.ts
@@ -3,7 +3,7 @@ import type { StreamTransport } from './controller.js';
 
 export interface WSOptions {
   baseURL: string;
-  topic: string;
+  table: string;
   since?: string;
   auth?: () => Promise<string> | string;
 }
@@ -48,7 +48,7 @@ export class WSTransport<T = Record<string, unknown>> implements StreamTransport
   private async _doConnect(): Promise<void> {
     const wsBase = this._opts.baseURL.replace(/^http/, 'ws');
     const url = new URL('/v1/stream/ws', wsBase);
-    url.searchParams.set('topic', this._opts.topic);
+    url.searchParams.set('table', this._opts.table);
     if (this._opts.since) {
       url.searchParams.set('since', this._opts.since);
     }
@@ -70,15 +70,23 @@ export class WSTransport<T = Record<string, unknown>> implements StreamTransport
 
     this._ws.onmessage = (e) => {
       try {
-        const msg = JSON.parse(e.data as string) as {
-          table_name: string;
-          received_timestamp: string;
-          data: T;
+        // Server WS envelope: {table, data: {table_name, received_timestamp, data}}.
+        // For non-EventMessage (raw-JSON pass-through) payloads, table_name on
+        // the inner object may be absent — fall back to the envelope's table so
+        // the StreamEvent always carries the subscribing table name.
+        const envelope = JSON.parse(e.data as string) as {
+          table: string;
+          data: {
+            table_name?: string;
+            received_timestamp?: string;
+            data: T;
+          };
         };
+        if (!envelope.table || !envelope.data) return;
         const event: StreamEvent<T> = {
-          table: msg.table_name,
-          timestamp: msg.received_timestamp,
-          data: msg.data,
+          table: envelope.data.table_name ?? envelope.table,
+          timestamp: envelope.data.received_timestamp ?? '',
+          data: envelope.data.data,
         };
         this.onEvent?.(event);
       } catch {
diff --git a/docs/src/content/docs/api.md b/docs/src/content/docs/api.md
index da31141f..6cec2aef 100644
--- a/docs/src/content/docs/api.md
+++ b/docs/src/content/docs/api.md
@@ -321,8 +321,8 @@ Opens a persistent SSE connection for real-time event streaming. Supports histor
 
 | Param | Type | Default | Description |
 | ----- | ---- | ------- | ----------- |
-| `topic` | string | `ingest.>` | NATS subject to subscribe to. Supports NATS wildcards: `*` matches one token, `>` matches one or more remaining tokens. |
-| `since` | string | — | RFC 3339 timestamp. If provided, replays historical events from NATS before switching to live streaming. |
+| `table` | string | (required) | Table name to subscribe to. Must match `^[a-zA-Z_][a-zA-Z0-9_]*$` (rejects NATS wildcards `*` / `>`). Returns 400 if missing or invalid. |
+| `since` | string | — | RFC 3339 or RFC 3339 Nano timestamp. If provided, replays historical events from NATS before switching to live streaming. |
 | `token` | string | — | JWT token (alternative to `Authorization` header, useful for `EventSource`). Stripped from URL after extraction. |
 
 **Headers:**
@@ -338,36 +338,35 @@ id: 2026-03-24T12:00:00.123Z
 data: {"table_name":"clicks","received_timestamp":"2026-03-24T12:00:00.123Z","data":{"page":"/home","button":"signup"}}
 
 id: 2026-03-24T12:00:01.456Z
-data: {"table_name":"page_views","received_timestamp":"2026-03-24T12:00:01.456Z","data":{"url":"/dashboard"}}
+data: {"table_name":"clicks","received_timestamp":"2026-03-24T12:00:01.456Z","data":{"page":"/pricing"}}
 ```
 
+Each SSE connection is bound to a single `?table=`; to consume multiple tables, open one connection per table or use the WebSocket endpoint with in-band multiplexing.
+
 **Note:** When access control policies are active, streamed events are filtered per the caller's role — denied columns are removed and tables without select permission are skipped.
 
 **curl example:**
 
 ```bash
-# All tables
-curl -N http://localhost:8080/v1/stream/sse
-
-# Specific table
-curl -N "http://localhost:8080/v1/stream/sse?topic=ingest.clicks"
+# Subscribe to a specific table
+curl -N "http://localhost:8080/v1/stream/sse?table=clicks"
 
 # With gap-fill
-curl -N "http://localhost:8080/v1/stream/sse?since=2026-03-24T11:00:00Z"
+curl -N "http://localhost:8080/v1/stream/sse?table=clicks&since=2026-03-24T11:00:00Z"
 ```
 
 ---
 
 ### `GET /v1/stream/ws` — WebSocket Stream
 
-Opens a WebSocket connection for real-time event streaming. Supports in-band multiplexing — a single WebSocket can subscribe to multiple topics dynamically.
+Opens a WebSocket connection for real-time event streaming. Supports in-band multiplexing — a single WebSocket can subscribe to multiple tables dynamically.
 
 **Query Parameters:**
 
 | Param | Type | Default | Description |
 | ----- | ---- | ------- | ----------- |
-| `topic` | string | — | Optional initial topic to subscribe to (backward compatible). If omitted, the client must send subscribe commands. |
-| `since` | string | — | RFC 3339 timestamp for gap-fill on the initial `?topic=` subscription. |
+| `table` | string | — | Optional initial table to subscribe to. If omitted, the client must send subscribe commands. When present, must match `^[a-zA-Z_][a-zA-Z0-9_]*$` — invalid values return 400 before the WebSocket upgrade. |
+| `since` | string | — | RFC 3339 or RFC 3339 Nano timestamp for gap-fill on the initial `?table=` subscription. |
 | `token` | string | — | JWT token (alternative to `Authorization` header). Stripped from URL after extraction. |
 
 **In-band commands (client → server):**
@@ -375,17 +374,17 @@ Opens a WebSocket connection for real-time event streaming. Supports in-band mul
 After connecting, send JSON commands to manage subscriptions:
 
 ```json
-{"action": "subscribe", "topic": "ingest.clicks"}
-{"action": "subscribe", "topic": "ingest.page_views"}
-{"action": "unsubscribe", "topic": "ingest.clicks"}
+{"action": "subscribe", "table": "clicks"}
+{"action": "subscribe", "table": "page_views"}
+{"action": "unsubscribe", "table": "clicks"}
 ```
 
 **Outbound message format (server → client):**
 
-Each message is wrapped in an envelope with the topic:
+Each message is wrapped in an envelope labelled with the table name:
 
 ```json
-{"topic": "ingest.clicks", "data": {"table_name": "clicks", "received_timestamp": "...", "data": {...}}}
+{"table": "clicks", "data": {"table_name": "clicks", "received_timestamp": "...", "data": {...}}}
 ```
 
 **JavaScript example:**
@@ -393,12 +392,12 @@ Each message is wrapped in an envelope with the topic:
 ```javascript
 const ws = new WebSocket("ws://localhost:8080/v1/stream/ws?token=<jwt>");
 ws.onopen = () => {
-  ws.send(JSON.stringify({ action: "subscribe", topic: "ingest.clicks" }));
-  ws.send(JSON.stringify({ action: "subscribe", topic: "ingest.page_views" }));
+  ws.send(JSON.stringify({ action: "subscribe", table: "clicks" }));
+  ws.send(JSON.stringify({ action: "subscribe", table: "page_views" }));
 };
 ws.onmessage = (event) => {
-  const { topic, data } = JSON.parse(event.data);
-  console.log(`[${topic}]`, data.table_name, data.data);
+  const { table, data } = JSON.parse(event.data);
+  console.log(`[${table}]`, data.table_name, data.data);
 };
 ```
 
diff --git a/docs/src/content/docs/architecture.md b/docs/src/content/docs/architecture.md
index 31784c12..8b6365b3 100644
--- a/docs/src/content/docs/architecture.md
+++ b/docs/src/content/docs/architecture.md
@@ -75,7 +75,7 @@ The API layer uses [Chi](https://github.com/go-chi/chi) for routing with standar
 - **structured_query.go** — Handler for `POST /v1/tables/{table}/query`: validates query AST, enforces permissions, builds and executes SQL.
 - **ingest.go** — Accepts flat JSON body for `POST /v1/ingest/{table}`, validates against discovered schema, optional dedup, publishes to NATS subject `ingest.{table}`.
 - **query.go** — Executes SQL queries directly against ClickHouse. Results are cached. UUID/DateTime columns are converted to strings.
-- **stream_sse.go** / **stream_ws.go** — Real-time streaming via SSE and WebSocket. Default topic is `ingest.>` (all tables). Supports gap-fill from NATS JetStream using `DeliverByStartTime`.
+- **stream_sse.go** / **stream_ws.go** — Real-time streaming via SSE and WebSocket. Callers select a table with the `?table=` query parameter (required for SSE); WS additionally accepts in-band `{"action":"subscribe","table":"..."}` commands. Supports gap-fill from NATS JetStream using `DeliverByStartTime`.
 - **transform.go** — Shared `transformForClient` function: passes through `table_name`, `received_timestamp`, and `data` from the wire format.
 - **schema.go** — Schema discovery API: list all schemas, get one table, trigger refresh.
 - **dlq.go** — DLQ stats endpoint and `EnsureDLQStream` helper for creating the `WAVEHOUSE_DLQ` NATS stream.
diff --git a/docs/src/content/docs/development.md b/docs/src/content/docs/development.md
index 9e3d8c25..5a0e2804 100644
--- a/docs/src/content/docs/development.md
+++ b/docs/src/content/docs/development.md
@@ -96,14 +96,11 @@ curl -s -X POST http://localhost:8080/v1/query \
   -H "Content-Type: application/json" \
   -d '{"sql": "SELECT * FROM clicks LIMIT 10"}'
 
-# Open an SSE stream for all tables (Ctrl+C to stop)
-curl -N http://localhost:8080/v1/stream/sse
-
-# Open an SSE stream for a specific table
-curl -N "http://localhost:8080/v1/stream/sse?topic=ingest.clicks"
+# Open an SSE stream for a specific table (Ctrl+C to stop)
+curl -N "http://localhost:8080/v1/stream/sse?table=clicks"
 
 # With gap-fill (replays events since the given timestamp, then switches to live)
-curl -N "http://localhost:8080/v1/stream/sse?since=2026-03-24T11:00:00Z"
+curl -N "http://localhost:8080/v1/stream/sse?table=clicks&since=2026-03-24T11:00:00Z"
 
 # Health check (no auth required)
 curl http://localhost:8080/health
diff --git a/docs/src/content/docs/getting-started.md b/docs/src/content/docs/getting-started.md
index 17b2fca9..2a2b6267 100644
--- a/docs/src/content/docs/getting-started.md
+++ b/docs/src/content/docs/getting-started.md
@@ -78,16 +78,15 @@ Prefer a type-safe query builder over raw SQL? See the [structured query endpoin
 Every ingested event is broadcast to SSE and WebSocket subscribers **before** it's flushed to ClickHouse, so dashboards see new data with zero perceived lag.
 
 ```bash
-# All tables
-curl -N http://localhost:8080/v1/stream/sse
-
-# Specific table
-curl -N "http://localhost:8080/v1/stream/sse?topic=ingest.clicks"
+# Specific table (?table= is required)
+curl -N "http://localhost:8080/v1/stream/sse?table=clicks"
 
 # With historical replay (RFC 3339 timestamp)
-curl -N "http://localhost:8080/v1/stream/sse?since=2026-03-24T11:00:00Z"
+curl -N "http://localhost:8080/v1/stream/sse?table=clicks&since=2026-03-24T11:00:00Z"
 ```
 
+To consume multiple tables on a single connection, use the WebSocket endpoint (`/v1/stream/ws`) with in-band `subscribe` commands — see the [API reference](api.md) for the envelope format.
+
 ## Next steps
 
 - **[Architecture](architecture.md)** — how ingest, query, cache, and streaming fit together.
diff --git a/docs/src/content/docs/sdk.md b/docs/src/content/docs/sdk.md
index 54d6d704..49183568 100644
--- a/docs/src/content/docs/sdk.md
+++ b/docs/src/content/docs/sdk.md
@@ -535,12 +535,12 @@ interface StreamEvent<T> {
 
 ### `SharedWSManager`
 
-When using WebSocket transport, the SDK automatically multiplexes all topic subscriptions over a single WebSocket connection per client via `SharedWSManager`. This is transparent — `.stream()` calls route through it automatically.
+When using WebSocket transport, the SDK automatically multiplexes all per-table subscriptions over a single WebSocket connection per client via `SharedWSManager`. This is transparent — `.stream()` calls route through it automatically.
 
 Key behaviors:
-- Ref-counted subscriptions: unsubscribing removes only when the last subscriber for a topic disconnects.
-- Auto-reconnect with exponential backoff; all active topics resubscribed on reconnect.
-- NATS-style wildcard matching (`*` = one token, `>` = one-or-more) on the client side.
+- Ref-counted subscriptions: unsubscribing removes only when the last subscriber for a table disconnects.
+- Auto-reconnect with exponential backoff; all active table subscriptions resubscribed on reconnect.
+- Exact-match dispatch by table name. The legacy NATS-style wildcards (`*`, `>`) are no longer accepted server-side; subscribe to one concrete table per call.
 
 ### Client-Side Stream Filtering
 
diff --git a/internal/api/hub.go b/internal/api/hub.go
index 26e94155..0e6ce960 100644
--- a/internal/api/hub.go
+++ b/internal/api/hub.go
@@ -3,7 +3,6 @@ package api
 import (
 	"context"
 	"encoding/json"
-	"strings"
 	"sync"
 
 	"github.com/Wave-RF/WaveHouse/internal/mq"
@@ -84,59 +83,10 @@ func (h *Hub) Broadcast(topic string, msg *mq.Message) {
 	h.mu.RLock()
 	defer h.mu.RUnlock()
 
-	sent := make(map[chan []byte]struct{})
-
-	// Exact match. Only mark a channel as "sent" after the send actually
-	// succeeds — if the channel is full and we hit the default case, the
-	// wildcard loop below gets a second chance.
 	for ch := range h.subscribers[topic] {
 		select {
 		case ch <- data:
-			sent[ch] = struct{}{}
 		default:
 		}
 	}
-
-	// Wildcard match: iterate all subscriber patterns.
-	for pattern, chs := range h.subscribers {
-		if pattern == topic {
-			continue // already handled above
-		}
-		if !matchTopic(pattern, topic) {
-			continue
-		}
-		for ch := range chs {
-			if _, dup := sent[ch]; dup {
-				continue
-			}
-			select {
-			case ch <- data:
-				sent[ch] = struct{}{}
-			default:
-			}
-		}
-	}
-}
-
-// matchTopic checks whether a NATS-style pattern matches a subject.
-// Tokens are separated by ".".
-//   - "*" matches exactly one token
-//   - ">" as the last pattern token matches one or more remaining tokens
-func matchTopic(pattern, subject string) bool {
-	pTokens := strings.Split(pattern, ".")
-	sTokens := strings.Split(subject, ".")
-
-	for i, pt := range pTokens {
-		if pt == ">" {
-			// ">" must be the last token and matches 1+ remaining subject tokens.
-			return i < len(sTokens)
-		}
-		if i >= len(sTokens) {
-			return false
-		}
-		if pt != "*" && pt != sTokens[i] {
-			return false
-		}
-	}
-	return len(pTokens) == len(sTokens)
 }
diff --git a/internal/api/hub_test.go b/internal/api/hub_test.go
index 5519276c..79babd30 100644
--- a/internal/api/hub_test.go
+++ b/internal/api/hub_test.go
@@ -208,204 +208,3 @@ func TestHub_UnsubscribeCleansEmptyTopic(t *testing.T) {
 	hub.mu.RUnlock()
 	assert.False(t, exists, "topic should be removed when last subscriber leaves")
 }
-
-func TestHub_WildcardGreaterThan(t *testing.T) {
-	t.Parallel()
-	hub := NewHub()
-
-	// Subscribe with "ingest.>" wildcard.
-	ch := make(chan []byte, 10)
-	hub.Subscribe("ingest.>", ch)
-	defer hub.Unsubscribe("ingest.>", ch)
-
-	hub.Broadcast("ingest.clicks", &mq.Message{
-		Ctx:     context.Background(),
-		Subject: "ingest.clicks",
-		Data:    []byte(`{"table_name":"clicks"}`),
-	})
-
-	select {
-	case msg := <-ch:
-		payload := unwrapTestMessage(t, msg)
-		assert.Contains(t, string(payload), "clicks")
-	case <-time.After(time.Second):
-		t.Fatal("wildcard subscriber should have received ingest.clicks")
-	}
-}
-
-func TestHub_WildcardStar(t *testing.T) {
-	t.Parallel()
-	hub := NewHub()
-
-	ch := make(chan []byte, 10)
-	hub.Subscribe("ingest.*", ch)
-	defer hub.Unsubscribe("ingest.*", ch)
-
-	hub.Broadcast("ingest.clicks", &mq.Message{
-		Ctx:     context.Background(),
-		Subject: "ingest.clicks",
-		Data:    []byte(`{"table_name":"clicks"}`),
-	})
-
-	select {
-	case msg := <-ch:
-		payload := unwrapTestMessage(t, msg)
-		assert.Contains(t, string(payload), "clicks")
-	case <-time.After(time.Second):
-		t.Fatal("star wildcard subscriber should have received")
-	}
-}
-
-func TestHub_WildcardStarNoMultiToken(t *testing.T) {
-	t.Parallel()
-	hub := NewHub()
-
-	ch := make(chan []byte, 10)
-	hub.Subscribe("ingest.*", ch)
-	defer hub.Unsubscribe("ingest.*", ch)
-
-	// "ingest.*" should NOT match "ingest.clicks.subpath" (star = one token).
-	hub.Broadcast("ingest.clicks.subpath", &mq.Message{
-		Ctx:     context.Background(),
-		Subject: "ingest.clicks.subpath",
-		Data:    []byte(`{"table_name":"clicks"}`),
-	})
-
-	select {
-	case <-ch:
-		t.Fatal("star wildcard should NOT match multi-token subject")
-	case <-time.After(50 * time.Millisecond):
-		// expected — no message
-	}
-}
-
-func TestHub_WildcardGreaterThanMultiToken(t *testing.T) {
-	t.Parallel()
-	hub := NewHub()
-
-	ch := make(chan []byte, 10)
-	hub.Subscribe("ingest.>", ch)
-	defer hub.Unsubscribe("ingest.>", ch)
-
-	// "ingest.>" should match multi-token subjects.
-	hub.Broadcast("ingest.clicks.subpath", &mq.Message{
-		Ctx:     context.Background(),
-		Subject: "ingest.clicks.subpath",
-		Data:    []byte(`{"table_name":"clicks"}`),
-	})
-
-	select {
-	case msg := <-ch:
-		payload := unwrapTestMessage(t, msg)
-		assert.Contains(t, string(payload), "clicks")
-	case <-time.After(time.Second):
-		t.Fatal("> wildcard should match multi-token subjects")
-	}
-}
-
-func TestHub_WildcardDoesNotMatchExact(t *testing.T) {
-	t.Parallel()
-	hub := NewHub()
-
-	ch := make(chan []byte, 10)
-	hub.Subscribe("ingest.>", ch)
-	defer hub.Unsubscribe("ingest.>", ch)
-
-	// "ingest.>" should NOT match "ingest" alone (> requires 1+ tokens after).
-	hub.Broadcast("ingest", &mq.Message{
-		Ctx:     context.Background(),
-		Subject: "ingest",
-		Data:    []byte(`{"table_name":"ingest"}`),
-	})
-
-	select {
-	case <-ch:
-		t.Fatal("> should not match bare prefix")
-	case <-time.After(50 * time.Millisecond):
-		// good
-	}
-}
-
-func TestHub_BareGreaterThanMatchesAll(t *testing.T) {
-	t.Parallel()
-	hub := NewHub()
-
-	ch := make(chan []byte, 10)
-	hub.Subscribe(">", ch)
-	defer hub.Unsubscribe(">", ch)
-
-	hub.Broadcast("ingest.clicks", &mq.Message{
-		Ctx:     context.Background(),
-		Subject: "ingest.clicks",
-		Data:    []byte(`{"table_name":"clicks"}`),
-	})
-
-	select {
-	case <-ch:
-		// expected
-	case <-time.After(time.Second):
-		t.Fatal("bare > should match everything")
-	}
-}
-
-func TestHub_WildcardNoDuplicateDelivery(t *testing.T) {
-	t.Parallel()
-	hub := NewHub()
-
-	ch := make(chan []byte, 10)
-	// Subscribe with both exact and wildcard that would match.
-	hub.Subscribe("ingest.clicks", ch)
-	hub.Subscribe("ingest.>", ch)
-	defer hub.Unsubscribe("ingest.clicks", ch)
-	defer hub.Unsubscribe("ingest.>", ch)
-
-	hub.Broadcast("ingest.clicks", &mq.Message{
-		Ctx:     context.Background(),
-		Subject: "ingest.clicks",
-		Data:    []byte(`{"table_name":"clicks"}`),
-	})
-
-	// Should receive exactly one message, not two.
-	select {
-	case <-ch:
-		// got first
-	case <-time.After(time.Second):
-		t.Fatal("should have received at least one message")
-	}
-
-	select {
-	case <-ch:
-		t.Fatal("should NOT receive a duplicate")
-	case <-time.After(50 * time.Millisecond):
-		// good
-	}
-}
-
-func TestMatchTopic(t *testing.T) {
-	t.Parallel()
-	tests := []struct {
-		pattern string
-		subject string
-		want    bool
-	}{
-		{"ingest.clicks", "ingest.clicks", true},
-		{"ingest.clicks", "ingest.users", false},
-		{"ingest.*", "ingest.clicks", true},
-		{"ingest.*", "ingest.clicks.sub", false},
-		{"ingest.>", "ingest.clicks", true},
-		{"ingest.>", "ingest.clicks.sub", true},
-		{"ingest.>", "ingest", false},
-		{">", "anything", true},
-		{">", "a.b.c", true},
-		{"*.*", "ingest.clicks", true},
-		{"*.*", "ingest", false},
-		{"a.*.c", "a.b.c", true},
-		{"a.*.c", "a.b.d", false},
-	}
-	for _, tt := range tests {
-		t.Run(tt.pattern+"_vs_"+tt.subject, func(t *testing.T) {
-			t.Parallel()
-			assert.Equal(t, tt.want, matchTopic(tt.pattern, tt.subject))
-		})
-	}
-}
diff --git a/internal/api/stream_sse.go b/internal/api/stream_sse.go
index 60f2d9b3..ed993009 100644
--- a/internal/api/stream_sse.go
+++ b/internal/api/stream_sse.go
@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"net/http"
+	"regexp"
 	"time"
 
 	"github.com/Wave-RF/WaveHouse/internal/ingest"
@@ -16,6 +17,13 @@ import (
 	"go.opentelemetry.io/otel/propagation"
 )
 
+// validTableNameRe matches safe table identifiers and — critically — rejects
+// the NATS subject wildcards `*` and `>`. The `?table=` value is concatenated
+// into a NATS FilterSubject in the gap-fill path; without this guard,
+// `?table=>` would build `ingest.>` and replay every ingest subject.
+// Matches the same shape as ingest.safeIdentifierRe / query.validIdentifierRe.
+var validTableNameRe = regexp.MustCompile(`^[a-zA-Z_][a-zA-Z0-9_]*$`)
+
 // SSEHandler handles GET /v1/stream/sse.
 type SSEHandler struct {
 	Hub         *Hub
@@ -34,10 +42,16 @@ func (h *SSEHandler) Handle(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	topic := r.URL.Query().Get("topic")
-	if topic == "" {
-		topic = "ingest.>"
+	table := r.URL.Query().Get("table")
+	if table == "" {
+		writeJSONError(w, http.StatusBadRequest, "missing required query parameter: table")
+		return
+	}
+	if !validTableNameRe.MatchString(table) {
+		writeJSONError(w, http.StatusBadRequest, "invalid table name")
+		return
 	}
+	topic := "ingest." + table
 
 	// Resolve stream permissions for this request.
 	role := RoleFromContext(r.Context())
diff --git a/internal/api/stream_test.go b/internal/api/stream_test.go
index 141c8e5f..5710f16a 100644
--- a/internal/api/stream_test.go
+++ b/internal/api/stream_test.go
@@ -1,12 +1,17 @@
 package api
 
 import (
+	"context"
 	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
 	"testing"
 	"time"
 
 	"github.com/Wave-RF/WaveHouse/internal/ingest"
 	"github.com/Wave-RF/WaveHouse/internal/policy"
+	"github.com/Wave-RF/WaveHouse/internal/testutil"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -138,13 +143,13 @@ func TestWS_ApplyStreamPolicy_FiltersColumns(t *testing.T) {
 	}
 	raw, _ := json.Marshal(evt)
 
-	out := h.applyStreamPolicy(raw, "user", nil, "ingest.events")
+	out := h.applyStreamPolicy(raw, "user", nil, "events")
 	require.NotNil(t, out)
 
 	var got map[string]any
 	require.NoError(t, json.Unmarshal(out, &got))
-	// WS wraps in topic envelope.
-	assert.Equal(t, "ingest.events", got["topic"])
+	// WS wraps in table envelope.
+	assert.Equal(t, "events", got["table"])
 	inner := got["data"].(map[string]any)
 	data := inner["data"].(map[string]any)
 	assert.Equal(t, "click", data["name"])
@@ -162,17 +167,118 @@ func TestWS_ApplyStreamPolicy_NoPolicy(t *testing.T) {
 	}
 	raw, _ := json.Marshal(evt)
 
-	out := h.applyStreamPolicy(raw, "", nil, "ingest.clicks")
+	out := h.applyStreamPolicy(raw, "", nil, "clicks")
 	require.NotNil(t, out)
 
 	var got map[string]any
 	require.NoError(t, json.Unmarshal(out, &got))
-	// WS wraps in topic envelope.
-	assert.Equal(t, "ingest.clicks", got["topic"])
+	// WS wraps in table envelope.
+	assert.Equal(t, "clicks", got["table"])
 	inner := got["data"].(map[string]any)
 	assert.Equal(t, "clicks", inner["table_name"])
 }
 
+// TestSSE_RejectsMissingOrInvalidTable verifies that the SSE handler returns
+// 400 when the ?table= parameter is missing or contains characters that could
+// be interpreted as NATS wildcards (regression guard for the fix that landed
+// alongside the wildcard fan-out removal in #100).
+func TestSSE_RejectsMissingOrInvalidTable(t *testing.T) {
+	t.Parallel()
+	h := &SSEHandler{Hub: NewHub()}
+
+	cases := []struct {
+		name    string
+		table   string
+		errBody string
+	}{
+		{"missing", "", "missing required query parameter: table"},
+		{"nats greater wildcard", ">", "invalid table name"},
+		{"nats star wildcard", "*", "invalid table name"},
+		{"dot separator", "ingest.clicks", "invalid table name"},
+		{"nested wildcard", "ingest.>", "invalid table name"},
+		{"trailing wildcard", "clicks.>", "invalid table name"},
+		{"space", "click s", "invalid table name"},
+		{"leading digit", "1clicks", "invalid table name"},
+		{"empty after url decode", "", "missing required query parameter: table"},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			target := "/v1/stream/sse"
+			if tc.table != "" {
+				target += "?table=" + url.QueryEscape(tc.table)
+			}
+			req := httptest.NewRequestWithContext(context.Background(), http.MethodGet, target, nil)
+			w := httptest.NewRecorder()
+			h.Handle(w, req)
+			testutil.AssertJSONContains(t, w, http.StatusBadRequest, map[string]any{"error": tc.errBody})
+		})
+	}
+}
+
+func TestSSE_AcceptsSafeTableName(t *testing.T) {
+	t.Parallel()
+	h := &SSEHandler{Hub: NewHub()}
+
+	// Use a request context that's already cancelled so the handler exits
+	// the live-stream select loop immediately instead of blocking the test.
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	req := httptest.NewRequestWithContext(ctx, http.MethodGet, "/v1/stream/sse?table=clicks", nil)
+	w := httptest.NewRecorder()
+	h.Handle(w, req)
+	// Past the validation gate — header set to text/event-stream, not the
+	// 400-path application/json.
+	assert.Equal(t, http.StatusOK, w.Code)
+	assert.Equal(t, "text/event-stream", w.Header().Get("Content-Type"))
+}
+
+func TestWS_RejectsInvalidTableOnQuery(t *testing.T) {
+	t.Parallel()
+	h := &WSHandler{Hub: NewHub()}
+
+	cases := []string{">", "*", "ingest.>", "clicks ", "clicks.subpath"}
+	for _, tbl := range cases {
+		t.Run(tbl, func(t *testing.T) {
+			t.Parallel()
+			target := "/v1/stream/ws?table=" + url.QueryEscape(tbl)
+			req := httptest.NewRequestWithContext(context.Background(), http.MethodGet, target, nil)
+			w := httptest.NewRecorder()
+			h.Handle(w, req)
+			// Validation runs before websocket.Accept, so we get a plain 400
+			// rather than an upgrade-attempt error.
+			testutil.AssertJSONContains(t, w, http.StatusBadRequest, map[string]any{"error": "invalid table name"})
+		})
+	}
+}
+
+func TestValidTableNameRe(t *testing.T) {
+	t.Parallel()
+	cases := []struct {
+		in   string
+		want bool
+	}{
+		{"clicks", true},
+		{"page_views", true},
+		{"_internal", true},
+		{"events_v2", true},
+		{">", false},
+		{"*", false},
+		{"ingest.>", false},
+		{"clicks.subpath", false},
+		{"clicks ", false},
+		{"clicks*", false},
+		{"1clicks", false},
+		{"", false},
+	}
+	for _, tc := range cases {
+		t.Run(tc.in, func(t *testing.T) {
+			t.Parallel()
+			assert.Equal(t, tc.want, validTableNameRe.MatchString(tc.in))
+		})
+	}
+}
+
 func TestExtractEventTimestamp(t *testing.T) {
 	t.Parallel()
 	tests := []struct {
diff --git a/internal/api/stream_ws.go b/internal/api/stream_ws.go
index 2dcd0767..4cd07037 100644
--- a/internal/api/stream_ws.go
+++ b/internal/api/stream_ws.go
@@ -22,16 +22,16 @@ import (
 // WSHandler handles GET /v1/stream/ws.
 // Supports multiplexed subscriptions via in-band JSON commands:
 //
-//	{"action":"subscribe","topic":"ingest.clicks"}
-//	{"action":"unsubscribe","topic":"ingest.clicks"}
+//	{"action":"subscribe","table":"clicks"}
+//	{"action":"unsubscribe","table":"clicks"}
 //
-// Outbound messages are wrapped in a topic envelope:
+// Outbound messages are wrapped in a table envelope:
 //
-//	{"topic":"ingest.clicks","data":{...event...}}
+//	{"table":"clicks","data":{...event...}}
 //
-// For backward compatibility, the ?topic= query parameter auto-subscribes
-// on connect. If no ?topic= is set, the connection starts with no
-// subscriptions and waits for in-band subscribe commands.
+// The optional ?table= query parameter auto-subscribes on connect. If no
+// ?table= is set, the connection starts with no subscriptions and waits
+// for in-band subscribe commands.
 type WSHandler struct {
 	Hub            *Hub
 	JS             jetstream.JetStream
@@ -46,10 +46,20 @@ func NewWSHandler(hub *Hub, js jetstream.JetStream, allowedOrigins []string) *WS
 // wsCommand represents an in-band subscribe/unsubscribe command.
 type wsCommand struct {
 	Action string `json:"action"`
-	Topic  string `json:"topic"`
+	Table  string `json:"table"`
 }
 
 func (h *WSHandler) Handle(w http.ResponseWriter, r *http.Request) {
+	// Validate ?table= before upgrading. Empty is OK — the client can
+	// still subscribe via in-band commands after connect. Reject NATS
+	// wildcards / unsafe chars here so the gap-fill path that builds
+	// FilterSubject: "ingest."+table can't be tricked into a wildcard
+	// consumer.
+	if t := r.URL.Query().Get("table"); t != "" && !validTableNameRe.MatchString(t) {
+		writeJSONError(w, http.StatusBadRequest, "invalid table name")
+		return
+	}
+
 	origins := h.AllowedOrigins
 	if len(origins) == 0 {
 		origins = []string{"*"}
@@ -68,68 +78,75 @@ func (h *WSHandler) Handle(w http.ResponseWriter, r *http.Request) {
 
 	ctx := r.Context()
 
-	// Merged channel receives messages from all subscribed topics.
-	merged := make(chan []byte, 64)
+	// Merged channel receives messages from all subscribed tables. Each message
+	// carries the subscribing table name so the dispatcher doesn't need to
+	// re-unmarshal the payload to label the outbound envelope — and so the
+	// envelope's "table" field is correct even when the payload isn't an
+	// ingest.EventMessage (e.g. raw JSON pass-through).
+	merged := make(chan wsOutbound, 64)
 
-	// Track active topic subscriptions and their per-topic channels.
+	// Track active subscriptions by table name, with their per-table channels.
 	var mu sync.Mutex
-	subs := make(map[string]chan []byte) // topic → per-topic channel
+	subs := make(map[string]chan []byte) // table → per-table channel
 
-	subscribeTopic := func(topic string) {
+	subscribeTable := func(table string) {
 		mu.Lock()
 		defer mu.Unlock()
-		if _, exists := subs[topic]; exists {
+		if _, exists := subs[table]; exists {
 			return
 		}
 		ch := make(chan []byte, 64)
-		subs[topic] = ch
-		h.Hub.Subscribe(topic, ch)
+		subs[table] = ch
+		h.Hub.Subscribe("ingest."+table, ch)
 
-		// Pump per-topic channel into merged channel
+		// Pump per-table channel into merged channel, tagging each message with
+		// the subscribing table so the writer keeps the subscription context.
 		go func() {
 			for msg := range ch {
 				select {
-				case merged <- msg:
+				case merged <- wsOutbound{table: table, data: msg}:
 				default:
 				}
 			}
 		}()
 	}
 
-	unsubscribeTopic := func(topic string) {
+	unsubscribeTable := func(table string) {
 		mu.Lock()
-		ch, exists := subs[topic]
+		ch, exists := subs[table]
 		if !exists {
 			mu.Unlock()
 			return
 		}
-		delete(subs, topic)
+		delete(subs, table)
 		mu.Unlock()
-		h.Hub.Unsubscribe(topic, ch) // closes ch, which stops the pump goroutine
+		h.Hub.Unsubscribe("ingest."+table, ch) // closes ch, which stops the pump goroutine
 	}
 
 	unsubscribeAll := func() {
 		mu.Lock()
-		topics := make([]string, 0, len(subs))
+		tables := make([]string, 0, len(subs))
 		for t := range subs {
-			topics = append(topics, t)
+			tables = append(tables, t)
 		}
 		mu.Unlock()
-		for _, t := range topics {
-			unsubscribeTopic(t)
+		for _, t := range tables {
+			unsubscribeTable(t)
 		}
 	}
 	defer unsubscribeAll()
 
-	// Backward compat: auto-subscribe if ?topic= is set.
-	if topic := r.URL.Query().Get("topic"); topic != "" {
-		subscribeTopic(topic)
+	// Auto-subscribe if ?table= is set.
+	if table := r.URL.Query().Get("table"); table != "" {
+		subscribeTable(table)
 
-		// Gap fill from NATS.
+		// Gap fill from NATS. RFC3339Nano subsumes RFC3339 (the fractional
+		// component is optional), matching the SSE handler's acceptance of
+		// high-precision client timestamps.
 		if since := r.URL.Query().Get("since"); since != "" {
-			if ts, parseErr := time.Parse(time.RFC3339, since); parseErr == nil && h.JS != nil {
-				h.replayFromNATS(ctx, ts, topic, func(data []byte) bool {
-					out := h.applyStreamPolicy(data, role, map[string]any(claims), topic)
+			if ts, parseErr := time.Parse(time.RFC3339Nano, since); parseErr == nil && h.JS != nil {
+				h.replayFromNATS(ctx, ts, "ingest."+table, func(data []byte) bool {
+					out := h.applyStreamPolicy(data, role, map[string]any(claims), table)
 					if out == nil {
 						return true
 					}
@@ -147,14 +164,20 @@ func (h *WSHandler) Handle(w http.ResponseWriter, r *http.Request) {
 				return
 			}
 			var cmd wsCommand
-			if json.Unmarshal(data, &cmd) != nil || cmd.Topic == "" {
+			if json.Unmarshal(data, &cmd) != nil || cmd.Table == "" {
+				continue
+			}
+			// Hub lookups are exact-match so wildcards here are inert, but
+			// reject them anyway for consistency with the ?table= path and
+			// to keep the contract crisp.
+			if !validTableNameRe.MatchString(cmd.Table) {
 				continue
 			}
 			switch cmd.Action {
 			case "subscribe":
-				subscribeTopic(cmd.Topic)
+				subscribeTable(cmd.Table)
 			case "unsubscribe":
-				unsubscribeTopic(cmd.Topic)
+				unsubscribeTable(cmd.Table)
 			}
 		}
 	}()
@@ -166,24 +189,15 @@ func (h *WSHandler) Handle(w http.ResponseWriter, r *http.Request) {
 		select {
 		case <-ctx.Done():
 			return
-		// Determine the event's table_name to use as the topic in the envelope.
-		case data := <-merged:
+		case m := <-merged:
 			var envelope struct {
 				TraceHeaders map[string]string `json:"trace_headers"`
 				Payload      []byte            `json:"payload"`
 			}
-			if err := json.Unmarshal(data, &envelope); err != nil {
+			if err := json.Unmarshal(m.data, &envelope); err != nil {
 				continue
 			}
 
-			var rawEvt struct {
-				TableName string `json:"table_name"`
-			}
-			evtTopic := ""
-			if json.Unmarshal(envelope.Payload, &rawEvt) == nil && rawEvt.TableName != "" {
-				evtTopic = "ingest." + rawEvt.TableName
-			}
-
 			parentCtx := otel.GetTextMapPropagator().Extract(
 				context.Background(),
 				propagation.MapCarrier(envelope.TraceHeaders),
@@ -191,7 +205,7 @@ func (h *WSHandler) Handle(w http.ResponseWriter, r *http.Request) {
 
 			_, pushSpan := tracer.Start(parentCtx, "WS.PushEvent")
 
-			out := h.applyStreamPolicy(envelope.Payload, role, map[string]any(claims), evtTopic)
+			out := h.applyStreamPolicy(envelope.Payload, role, map[string]any(claims), m.table)
 			if out == nil {
 				pushSpan.End()
 				continue
@@ -206,16 +220,23 @@ func (h *WSHandler) Handle(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
+// wsOutbound pairs a message with the subscribing table name so the WS write
+// loop can label the outbound envelope without re-unmarshalling the payload.
+type wsOutbound struct {
+	table string
+	data  []byte
+}
+
 // applyStreamPolicy transforms raw event data for the client, filtering columns
 // based on the caller's policy permissions. Returns nil if the event should be skipped.
-// The result is wrapped in a topic envelope: {"topic":"...","data":{...}}.
-func (h *WSHandler) applyStreamPolicy(raw []byte, role string, claims map[string]any, topic string) []byte {
+// The result is wrapped in a table envelope: {"table":"...","data":{...}}.
+func (h *WSHandler) applyStreamPolicy(raw []byte, role string, claims map[string]any, table string) []byte {
 	var evt ingest.EventMessage
 	if err := json.Unmarshal(raw, &evt); err != nil || evt.TableName == "" {
 		if !json.Valid(raw) {
 			return nil
 		}
-		envelope := map[string]any{"topic": topic, "data": json.RawMessage(raw)}
+		envelope := map[string]any{"table": table, "data": json.RawMessage(raw)}
 		data, err := json.Marshal(envelope)
 		if err != nil {
 			return nil
@@ -238,7 +259,7 @@ func (h *WSHandler) applyStreamPolicy(raw []byte, role string, claims map[string
 		"data":               evt.Data,
 	}
 	envelope := map[string]any{
-		"topic": "ingest." + evt.TableName,
+		"table": evt.TableName,
 		"data":  inner,
 	}
 	data, err := json.Marshal(envelope)