fix(server): return ThetaData stream-verification values from the websocket bridge (#807)

* fix(server): return ThetaData stream-verification values from the websocket bridge

The bundled WebSocket bridge acknowledged every stream request with `OK`, which is not one of the stream-request verification values ThetaData clients match on. A successful subscribe, unsubscribe, or stop now acknowledges with `SUBSCRIBED`.

It also returned `OK` when a subscribe arrived before streaming had started, even though nothing was installed. That was a false-positive acknowledgement: the client believed it was subscribed while no feed existed. That path now returns `ERROR` with a message saying streaming is not started.

The acknowledgement value is built from a small internal enum so the wire strings live in one place. ThetaData also documents `MAX_STREAMS_REACHED` and `INVALID_PERMS`, but the upstream FPSS error type does not carry enough to tell those apart from a generic failure, so both currently fall through to `ERROR` with a descriptive message; the specific values can be emitted once the error type surfaces the category.

Closes #801

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* docs(server): the websocket strike field is in dollars, not thousandths

The websocket bridge parses the option strike as dollars, but the docs still showed thousandths (570000) and called it the one exception to the dollars-everywhere rule, so a client following the docs would subscribe to the wrong contract. Show the strike as a dollar number.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* docs: the websocket strike envelope is dollars across the streaming docs

The strike-dollars change made the bundled server's websocket subscribe envelope take dollars, but the per-endpoint streaming option docs and the symbology page still showed thousandths (570000) and called the websocket the one exception to dollars-everywhere. Correct the examples and prose so a websocket client sends the dollar value.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* fix(docs): emit the websocket strike envelope in dollars from the generator

The streaming-options doc generator hard-coded the websocket subscribe
example with the strike as a scaled integer (570000) and prose calling
it thousandths, so a client copying the published envelope subscribed
to a $570,000 strike instead of the $570 option. The server's own
subscribe path takes dollars, so the docs contradicted the code.

Emit the strike in dollars (570) from the three option stream specs and
rewrite the envelope note to match. Invert the docs-consistency guard so
it bans the thousandths vocabulary and the 570000 literal on every page
that carries a websocket subscribe envelope, instead of whitelisting the
thousandths note as an allowed exception.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* fix(server): publish stream-verification tokens and make WS STOP apply

The WebSocket REQ_RESPONSE and DISCONNECTED frames rendered their outcome via the Rust Debug form, putting internal variant identifiers (`Subscribed`, `MaxStreamsReached`, `InvalidPerms`, and the `RemoveReason` names) onto the wire where clients expect the documented stream-verification vocabulary. Give `StreamResponseType` and `RemoveReason` an explicit `as_wire_str` mapping (the single source of `SUBSCRIBED` / `ERROR` / `MAX_STREAMS_REACHED` / `INVALID_PERMS` and the SCREAMING_SNAKE disconnect tokens) and emit that, so the variant identifier can never reach a client. `StreamResponseType` also gains a `Display` impl routed through the same mapping.

STOP previously acknowledged `SUBSCRIBED` unconditionally — even when streaming was never started or no stream existed — which told the client its feed was gone when nothing changed. STOP is a removal, not a status query, so it now snapshots the live subscription set, removes it, and acknowledges success only when the removal actually applied; an inactive session or a failed unsubscribe surfaces `ERROR` with a diagnostic, matching the subscribe path's discipline.

Correct the index price streaming doc example, which subscribed `Contract.stock("SPX")` for an index; SPX is an index, so the example now uses `Contract.index("SPX")` to match the other language rows. Regenerate the checked-in docs page from the corrected source.

Unit tests assert each enum variant maps to its exact wire token (the regression guard against the Debug leak), the STOP error/success token contract, and that no variant identifier appears in the serialized frame.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

---------

Co-authored-by: Claude <noreply@anthropic.com>

Author: userFRM

CHANGELOG.md

@@ -69,6 +69,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - The CLI's raw OHLC output (`--format json-raw` / `csv`) emits the `vwap` value in its own column. The raw value row was one field short of its header set, so `vwap` was dropped and `date`, `expiration`, `strike`, and `right` rendered under the wrong column names. The typed SDK surfaces and the CLI's presentation `json` were unaffected.
 - Python — the standalone `StreamingClient` streaming connect, reconnect, and subscribe / unsubscribe paths release the GIL across their blocking I/O (the TLS connect and handshake, and the per-subscription wire write), so other Python threads keep running while a connect or subscribe is in flight; the typed exception raised on failure is unchanged.
 - The standalone Python and TypeScript streaming clients now forward the full streaming and reconnect config, so every tuning knob — including the wait strategy and consumer-core affinity, host selection, watchdog and keepalive cadences, and the reconnect backoff and replay pacing — is honored, matching the unified client and the C ABI.
+- The bundled WebSocket server acknowledges stream requests with ThetaData's stream-verification values: a successful subscribe, unsubscribe, or stop now returns `SUBSCRIBED` instead of `OK`, and a subscribe that arrives before streaming has started returns `ERROR` with a descriptive message instead of a false-positive `OK` that claimed success while installing nothing.
 
 ### Security
 

crates/thetadatadx/build_support_bin/endpoints/docs_render/streaming.rs

@@ -200,7 +200,7 @@ const STREAMS: &[StreamSpec] = &[
         cpp_sub: "thetadatadx::Contract::option(\"SPY\", {.expiration = \"20260618\", .strike = \"570\", .right = \"C\"}).quote()",
         ws_req_type: "QUOTE",
         ws_sec_type: "OPTION",
-        ws_contract: Some(r#"{"symbol": "SPY", "expiration": 20260618, "strike": 570000, "right": "C"}"#),
+        ws_contract: Some(r#"{"symbol": "SPY", "expiration": 20260618, "strike": 570, "right": "C"}"#),
         group: "Options",
         label: "Quote",
     },
@@ -216,7 +216,7 @@ const STREAMS: &[StreamSpec] = &[
         cpp_sub: "thetadatadx::Contract::option(\"SPY\", {.expiration = \"20260618\", .strike = \"570\", .right = \"C\"}).trade()",
         ws_req_type: "TRADE",
         ws_sec_type: "OPTION",
-        ws_contract: Some(r#"{"symbol": "SPY", "expiration": 20260618, "strike": 570000, "right": "C"}"#),
+        ws_contract: Some(r#"{"symbol": "SPY", "expiration": 20260618, "strike": 570, "right": "C"}"#),
         group: "Options",
         label: "Trade",
     },
@@ -232,7 +232,7 @@ const STREAMS: &[StreamSpec] = &[
         cpp_sub: "thetadatadx::Contract::option(\"SPY\", {.expiration = \"20260618\", .strike = \"570\", .right = \"C\"}).open_interest()",
         ws_req_type: "OPEN_INTEREST",
         ws_sec_type: "OPTION",
-        ws_contract: Some(r#"{"symbol": "SPY", "expiration": 20260618, "strike": 570000, "right": "C"}"#),
+        ws_contract: Some(r#"{"symbol": "SPY", "expiration": 20260618, "strike": 570, "right": "C"}"#),
         group: "Options",
         label: "Open Interest",
     },
@@ -275,7 +275,7 @@ const STREAMS: &[StreamSpec] = &[
         prose: "Streams every index value update. Indices publish price prints through the trade feed, so each update delivers a `Trade` event whose `price` field carries the index value. Indices have no full-stream broadcast; subscribe per index.",
         event: "Trade",
         rust_sub: "Contract::index(\"SPX\").trade()",
-        python_sub: "Contract.stock(\"SPX\").trade()",
+        python_sub: "Contract.index(\"SPX\").trade()",
         ts_sub: "Contract.index('SPX').trade()",
         cpp_sub: "thetadatadx::Contract::index(\"SPX\").trade()",
         ws_req_type: "TRADE",
@@ -417,7 +417,7 @@ fn http_tab(spec: &StreamSpec) -> String {
     );
     if spec.ws_contract.is_some_and(|c| c.contains("strike")) {
         out.push_str(
-            "\nThe WebSocket envelope takes the strike in thousandths of a dollar (`570000` = $570.00); the SDKs take dollars.\n",
+            "\nThe WebSocket envelope takes the strike in dollars (`570` = $570.00), the same as the SDKs.\n",
         );
     }
     out

crates/thetadatadx/src/tdbe/types/enums.rs

@@ -603,6 +603,33 @@ pub enum StreamResponseType {
     InvalidPerms = 3,
 }
 
+impl StreamResponseType {
+    /// Stream-request verification token this outcome is published as on
+    /// the WebSocket wire.
+    ///
+    /// The wire vocabulary is fixed by the stream-verification contract
+    /// every client matches on: `SUBSCRIBED` / `ERROR` /
+    /// `MAX_STREAMS_REACHED` / `INVALID_PERMS`. This mapping is the single
+    /// source of those tokens, so the Rust variant identifier can never
+    /// reach the wire — emitting the debug form of the variant would break
+    /// every client written against the documented vocabulary.
+    #[must_use]
+    pub const fn as_wire_str(self) -> &'static str {
+        match self {
+            Self::Subscribed => "SUBSCRIBED",
+            Self::Error => "ERROR",
+            Self::MaxStreamsReached => "MAX_STREAMS_REACHED",
+            Self::InvalidPerms => "INVALID_PERMS",
+        }
+    }
+}
+
+impl core::fmt::Display for StreamResponseType {
+    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
+        f.write_str(self.as_wire_str())
+    }
+}
+
 /// Disconnect reason codes.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 #[repr(i16)]
@@ -699,6 +726,37 @@ impl RemoveReason {
             Self::InvalidCredentialsNullUser => "InvalidCredentialsNullUser",
         }
     }
+
+    /// Stable SCREAMING_SNAKE_CASE token published as the disconnect
+    /// `reason` on the WebSocket wire.
+    ///
+    /// This mapping is the single source of the wire reason vocabulary, so
+    /// the Rust variant identifier can never reach a client: emitting the
+    /// debug form of the variant would leak an internal type name in place
+    /// of the documented status token.
+    #[must_use]
+    pub const fn as_wire_str(self) -> &'static str {
+        match self {
+            Self::Unspecified => "UNSPECIFIED",
+            Self::InvalidCredentials => "INVALID_CREDENTIALS",
+            Self::InvalidLoginValues => "INVALID_LOGIN_VALUES",
+            Self::InvalidLoginSize => "INVALID_LOGIN_SIZE",
+            Self::GeneralValidationError => "GENERAL_VALIDATION_ERROR",
+            Self::TimedOut => "TIMED_OUT",
+            Self::ClientForcedDisconnect => "CLIENT_FORCED_DISCONNECT",
+            Self::AccountAlreadyConnected => "ACCOUNT_ALREADY_CONNECTED",
+            Self::SessionTokenExpired => "SESSION_TOKEN_EXPIRED",
+            Self::InvalidSessionToken => "INVALID_SESSION_TOKEN",
+            Self::FreeAccount => "FREE_ACCOUNT",
+            Self::TooManyRequests => "TOO_MANY_REQUESTS",
+            Self::NoStartDate => "NO_START_DATE",
+            Self::LoginTimedOut => "LOGIN_TIMED_OUT",
+            Self::ServerRestarting => "SERVER_RESTARTING",
+            Self::SessionTokenNotFound => "SESSION_TOKEN_NOT_FOUND",
+            Self::ServerUserDoesNotExist => "SERVER_USER_DOES_NOT_EXIST",
+            Self::InvalidCredentialsNullUser => "INVALID_CREDENTIALS_NULL_USER",
+        }
+    }
 }
 
 /// Market-calendar day classification.
@@ -783,3 +841,74 @@ impl std::fmt::Display for CalendarStatus {
 }
 
 include!("generated/enums_endpoint.rs");
+
+#[cfg(test)]
+mod wire_token_tests {
+    use super::{RemoveReason, StreamResponseType};
+
+    /// Every stream-response outcome maps to its exact stream-verification
+    /// token. This is the single source the WebSocket surface publishes,
+    /// so a drift here would leak the Rust variant name onto the wire and
+    /// break clients matching on the documented vocabulary.
+    #[test]
+    fn stream_response_wire_tokens_are_exact() {
+        assert_eq!(StreamResponseType::Subscribed.as_wire_str(), "SUBSCRIBED");
+        assert_eq!(StreamResponseType::Error.as_wire_str(), "ERROR");
+        assert_eq!(
+            StreamResponseType::MaxStreamsReached.as_wire_str(),
+            "MAX_STREAMS_REACHED"
+        );
+        assert_eq!(
+            StreamResponseType::InvalidPerms.as_wire_str(),
+            "INVALID_PERMS"
+        );
+        // `Display` routes through the same mapping.
+        assert_eq!(StreamResponseType::Subscribed.to_string(), "SUBSCRIBED");
+    }
+
+    /// Every disconnect reason maps to a stable SCREAMING_SNAKE token; the
+    /// Rust variant identifier must never reach the wire.
+    #[test]
+    fn remove_reason_wire_tokens_are_exact() {
+        for (reason, token) in [
+            (RemoveReason::Unspecified, "UNSPECIFIED"),
+            (RemoveReason::InvalidCredentials, "INVALID_CREDENTIALS"),
+            (RemoveReason::InvalidLoginValues, "INVALID_LOGIN_VALUES"),
+            (RemoveReason::InvalidLoginSize, "INVALID_LOGIN_SIZE"),
+            (
+                RemoveReason::GeneralValidationError,
+                "GENERAL_VALIDATION_ERROR",
+            ),
+            (RemoveReason::TimedOut, "TIMED_OUT"),
+            (
+                RemoveReason::ClientForcedDisconnect,
+                "CLIENT_FORCED_DISCONNECT",
+            ),
+            (
+                RemoveReason::AccountAlreadyConnected,
+                "ACCOUNT_ALREADY_CONNECTED",
+            ),
+            (RemoveReason::SessionTokenExpired, "SESSION_TOKEN_EXPIRED"),
+            (RemoveReason::InvalidSessionToken, "INVALID_SESSION_TOKEN"),
+            (RemoveReason::FreeAccount, "FREE_ACCOUNT"),
+            (RemoveReason::TooManyRequests, "TOO_MANY_REQUESTS"),
+            (RemoveReason::NoStartDate, "NO_START_DATE"),
+            (RemoveReason::LoginTimedOut, "LOGIN_TIMED_OUT"),
+            (RemoveReason::ServerRestarting, "SERVER_RESTARTING"),
+            (
+                RemoveReason::SessionTokenNotFound,
+                "SESSION_TOKEN_NOT_FOUND",
+            ),
+            (
+                RemoveReason::ServerUserDoesNotExist,
+                "SERVER_USER_DOES_NOT_EXIST",
+            ),
+            (
+                RemoveReason::InvalidCredentialsNullUser,
+                "INVALID_CREDENTIALS_NULL_USER",
+            ),
+        ] {
+            assert_eq!(reason.as_wire_str(), token, "{reason:?}");
+        }
+    }
+}

docs-site/docs/articles/symbology.md

@@ -22,7 +22,7 @@ An option contract is always the four-tuple **(symbol, expiration, strike, right
 
 Strikes are **human dollars everywhere on the SDK surface** — never scaled integers. Endpoints that take an optional `strike` / `right` treat omission as a wildcard: all strikes, both rights. Wildcard responses identify each row's contract via the `expiration`, `strike`, and `right` response fields.
 
-The one place a scaled strike survives: the server's [WebSocket subscribe envelope](/server/websocket) takes thousandths of a dollar (`570000` = $570.00), matching the upstream wire convention.
+Strikes are dollars across every surface, including the bundled server's [WebSocket subscribe envelope](/server/websocket) (`570` = $570.00). The scaled-integer form only exists on the raw upstream wire, which the SDK and the server both hide.
 
 ## Dates and times in
 

docs-site/docs/changelog.md

@@ -69,6 +69,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - The CLI's raw OHLC output (`--format json-raw` / `csv`) emits the `vwap` value in its own column. The raw value row was one field short of its header set, so `vwap` was dropped and `date`, `expiration`, `strike`, and `right` rendered under the wrong column names. The typed SDK surfaces and the CLI's presentation `json` were unaffected.
 - Python — the standalone `StreamingClient` streaming connect, reconnect, and subscribe / unsubscribe paths release the GIL across their blocking I/O (the TLS connect and handshake, and the per-subscription wire write), so other Python threads keep running while a connect or subscribe is in flight; the typed exception raised on failure is unchanged.
 - The standalone Python and TypeScript streaming clients now forward the full streaming and reconnect config, so every tuning knob — including the wait strategy and consumer-core affinity, host selection, watchdog and keepalive cadences, and the reconnect backoff and replay pacing — is honored, matching the unified client and the C ABI.
+- The bundled WebSocket server acknowledges stream requests with ThetaData's stream-verification values: a successful subscribe, unsubscribe, or stop now returns `SUBSCRIBED` instead of `OK`, and a subscribe that arrives before streaming has started returns `ERROR` with a descriptive message instead of a false-positive `OK` that claimed success while installing nothing.
 
 ### Security
 

docs-site/docs/server/websocket.md

@@ -28,19 +28,19 @@ The server bridges [streaming](/streaming/) onto a local WebSocket at `ws://127.
 | `id` | Your request id; echoed in the acknowledgement |
 | `contract` | Omit for `FULL_*` streams |
 
-Option contracts carry the four-tuple, with the strike in **thousandths of a dollar** (the one wire-format exception to the dollars-everywhere rule — see [Symbology](/articles/symbology)):
+Option contracts carry the four-tuple, with the strike in **dollars** (a JSON number, e.g. `570` or `570.0`):
 
 ```json
-{"symbol": "SPY", "expiration": 20250321, "strike": 570000, "right": "C"}
+{"symbol": "SPY", "expiration": 20250321, "strike": 570, "right": "C"}
 ```
 
-`{"msg_type": "STOP", "id": 2}` removes every active stream at once. Each command is acknowledged:
+`{"msg_type": "STOP", "id": 2}` removes every active stream at once. Each command is acknowledged with a stream-request verification value in the `response` field:
 
 ```json
-{ "header": { "type": "REQ_RESPONSE", "response": "OK", "req_id": 1 } }
+{ "header": { "type": "REQ_RESPONSE", "response": "SUBSCRIBED", "req_id": 1 } }
 ```
 
-Invalid commands answer with `"response": "ERROR"` and a message naming the offending field.
+`SUBSCRIBED` confirms the request was accepted; it acknowledges subscribe (`add: true`), unsubscribe (`add: false`), and `STOP`, since there is no removal-specific value. Rejected commands answer with `"response": "ERROR"` and an `error` field naming the cause — a bad envelope, an offending field, or a subscribe sent before streaming has started (which installs nothing, so it is never acknowledged as a success). The values `MAX_STREAMS_REACHED` and `INVALID_PERMS` are also part of the verification vocabulary.
 
 ## Event messages
 
@@ -50,7 +50,7 @@ Events arrive as JSON with a `header.type` of `QUOTE`, `TRADE`, `OHLC`, or `OPEN
 
 ```bash
 websocat ws://127.0.0.1:25520/v1/events
-{"msg_type": "STREAM", "sec_type": "OPTION", "req_type": "TRADE", "id": 1, "add": true, "contract": {"symbol": "SPY", "expiration": 20250321, "strike": 570000, "right": "C"}}
+{"msg_type": "STREAM", "sec_type": "OPTION", "req_type": "TRADE", "id": 1, "add": true, "contract": {"symbol": "SPY", "expiration": 20250321, "strike": 570, "right": "C"}}
 ```
 
 ## Limits

docs-site/docs/streaming/indices/price.md

@@ -45,7 +45,7 @@ def on_event(event):
 
 client.stream.start_streaming(on_event)
 
-sub = Contract.stock("SPX").trade()
+sub = Contract.index("SPX").trade()
 client.stream.subscribe(sub)
 
 # Remove this stream; the session stays open for other subscriptions.

docs-site/docs/streaming/options/open-interest.md

@@ -106,10 +106,10 @@ WebSocket streaming from the bundled [server binary](/server/). Send one JSON en
 
 ```bash
 websocat ws://127.0.0.1:25520/v1/events
-{"msg_type": "STREAM", "sec_type": "OPTION", "req_type": "OPEN_INTEREST", "id": 1, "add": true, "contract": {"symbol": "SPY", "expiration": 20260618, "strike": 570000, "right": "C"}}
+{"msg_type": "STREAM", "sec_type": "OPTION", "req_type": "OPEN_INTEREST", "id": 1, "add": true, "contract": {"symbol": "SPY", "expiration": 20260618, "strike": 570, "right": "C"}}
 ```
 
-The WebSocket envelope takes the strike in thousandths of a dollar (`570000` = $570.00); the SDKs take dollars.
+The WebSocket envelope takes the strike in dollars (`570` = $570.00), the same as the SDKs.
 
 </template>
 

docs-site/docs/streaming/options/quote.md

@@ -106,10 +106,10 @@ WebSocket streaming from the bundled [server binary](/server/). Send one JSON en
 
 ```bash
 websocat ws://127.0.0.1:25520/v1/events
-{"msg_type": "STREAM", "sec_type": "OPTION", "req_type": "QUOTE", "id": 1, "add": true, "contract": {"symbol": "SPY", "expiration": 20260618, "strike": 570000, "right": "C"}}
+{"msg_type": "STREAM", "sec_type": "OPTION", "req_type": "QUOTE", "id": 1, "add": true, "contract": {"symbol": "SPY", "expiration": 20260618, "strike": 570, "right": "C"}}
 ```
 
-The WebSocket envelope takes the strike in thousandths of a dollar (`570000` = $570.00); the SDKs take dollars.
+The WebSocket envelope takes the strike in dollars (`570` = $570.00), the same as the SDKs.
 
 </template>
 

docs-site/docs/streaming/options/trade.md

@@ -106,10 +106,10 @@ WebSocket streaming from the bundled [server binary](/server/). Send one JSON en
 
 ```bash
 websocat ws://127.0.0.1:25520/v1/events
-{"msg_type": "STREAM", "sec_type": "OPTION", "req_type": "TRADE", "id": 1, "add": true, "contract": {"symbol": "SPY", "expiration": 20260618, "strike": 570000, "right": "C"}}
+{"msg_type": "STREAM", "sec_type": "OPTION", "req_type": "TRADE", "id": 1, "add": true, "contract": {"symbol": "SPY", "expiration": 20260618, "strike": 570, "right": "C"}}
 ```
 
-The WebSocket envelope takes the strike in thousandths of a dollar (`570000` = $570.00); the SDKs take dollars.
+The WebSocket envelope takes the strike in dollars (`570` = $570.00), the same as the SDKs.
 
 </template>