From 865cc13813012a5e661357f68910ea7619ebd6c4 Mon Sep 17 00:00:00 2001 From: Sohum Desai Date: Tue, 22 Sep 2026 14:10:45 -0400 Subject: [PATCH 1/4] feat(mcp-server): migrate prediction contract-status WebSocket stream to SDK (PREDICT-8823) Routes the contractStatus and orders@account WebSocket channels through @gemini-markets/sdk/server's public/private streams instead of the legacy hand-rolled client, threading the already-authenticated SdkClient into WebSocketManager at both production call sites (server.ts, alerts daemon). Trade/depth/bookTicker/ticker dispatch on the legacy client is untouched, per the ticket's scope. Co-Authored-By: Claude Sonnet 5 --- .../mcp-server/src/alerts/daemon/index.ts | 2 +- packages/mcp-server/src/server.ts | 2 +- .../mcp-server/src/websocket/manager.test.ts | 552 +++++++++++------- packages/mcp-server/src/websocket/manager.ts | 218 ++++--- 4 files changed, 482 insertions(+), 292 deletions(-) diff --git a/packages/mcp-server/src/alerts/daemon/index.ts b/packages/mcp-server/src/alerts/daemon/index.ts index 3824614..80b03b9 100644 --- a/packages/mcp-server/src/alerts/daemon/index.ts +++ b/packages/mcp-server/src/alerts/daemon/index.ts @@ -106,7 +106,7 @@ async function main(): Promise { const httpClient = new GeminiHttpClient(); const sdkClient = await createSdkClient(); const marketStore = new MarketDataStore(); - const wsManager = new WebSocketManager(config.wsUrl, marketStore); + const wsManager = new WebSocketManager(config.wsUrl, sdkClient, marketStore); await wsManager.initialize(); const store = new AlertStore(); diff --git a/packages/mcp-server/src/server.ts b/packages/mcp-server/src/server.ts index ab3b0b8..51f308d 100644 --- a/packages/mcp-server/src/server.ts +++ b/packages/mcp-server/src/server.ts @@ -64,7 +64,7 @@ export function createServer(sdkClient: SdkClient): Server { ); const client = new GeminiHttpClient(); - const wsManager = new WebSocketManager(config.wsUrl); + const wsManager = new WebSocketManager(config.wsUrl, sdkClient); const allTools: ToolDefinition[] = [ ...createMarketTools(client), diff --git a/packages/mcp-server/src/websocket/manager.test.ts b/packages/mcp-server/src/websocket/manager.test.ts index ca89461..10c088e 100644 --- a/packages/mcp-server/src/websocket/manager.test.ts +++ b/packages/mcp-server/src/websocket/manager.test.ts @@ -5,13 +5,16 @@ import { WebSocketManager, toChannelSymbol, toEventTimeMs } from './manager.js'; import { config } from '../config.js'; import type { MarketDataStore } from '../store/index.js'; import type { CachedContractStatus, CachedOrderUpdate } from '../types/websocket.js'; - -// Shared by the wire-level contractStatus tests below: the server sends its -// fixture synchronously right after acking the subscribe request, so it may -// already be in the store by the time subscribeContractStatus() resolves — -// check directly first, and only fall back to waiting on the next onUpdate -// (with a bounded timeout, so a real regression fails fast instead of -// hanging) if it genuinely hasn't arrived yet. +import type { SdkClient } from '../client/sdk.js'; + +// Shared by the contractStatus/orders@account tests below: the fake stream +// resolves its `ready` promise (and, for the message tests, delivers its +// fixture frame) synchronously once subscribed — so the update may already +// be in the store by the time subscribeContractStatus()/ +// subscribeAccountOrders() resolves. Check directly first, and only fall +// back to waiting on the next onUpdate (with a bounded timeout, so a real +// regression fails fast instead of hanging) if it genuinely hasn't arrived +// yet. function waitForContractStatus( store: MarketDataStore, symbol: string, @@ -34,7 +37,7 @@ function waitForContractStatus( }); } -// Same reasoning as waitForContractStatus, for the order-update wire tests. +// Same reasoning as waitForContractStatus, for the order-update tests. function waitForOrder( store: MarketDataStore, orderId: string, @@ -80,7 +83,105 @@ test('toEventTimeMs leaves a millisecond-scale value unchanged', () => { assert.strictEqual(toEventTimeMs(1_710_000_000_000), 1_710_000_000_000); }); -async function withEchoServer(run: (manager: WebSocketManager, receivedParams: string[]) => Promise): Promise { +// --------------------------------------------------------------------------- +// Fake SDK client / WebSocketStream test double. +// +// PREDICT-8823 moved subscribeContractStatus()/subscribeAccountOrders() off +// the legacy hand-rolled GeminiWebSocketClient onto the SDK's +// `client.websocket.public.contractStatus()` / +// `client.websocket.private.orders({scope:'account'})` streams. Those +// streams are already built, tested and covered (reconnect/backoff/ +// message-limit/integer-safety) in the SDK's own websocket.test.ts — this +// file only needs a minimal double satisfying the surface +// WebSocketManager actually consumes: `.on('message', cb)` and `.ready` +// (awaited to know the subscribe ack landed) plus `.close()` (called from +// WebSocketManager.disconnect()). +// --------------------------------------------------------------------------- + +class FakeWebSocketStream { + readonly ready: Promise<{ result: null }>; + private resolveReady!: () => void; + private rejectReady!: (err: Error) => void; + private messageHandlers: Array<(msg: T) => void> = []; + closed = false; + + constructor() { + this.ready = new Promise((resolve, reject) => { + this.resolveReady = () => resolve({ result: null }); + this.rejectReady = reject; + }); + } + + on(event: 'message', cb: (msg: T) => void): this { + if (event === 'message') this.messageHandlers.push(cb); + return this; + } + + off(): this { + return this; + } + + close(): Promise { + this.closed = true; + return Promise.resolve(); + } + + /** Test helper: simulate the subscribe ack landing. */ + ackReady(): void { + this.resolveReady(); + } + + /** Test helper: simulate the subscribe being rejected by the exchange/SDK. */ + failReady(err: Error): void { + this.rejectReady(err); + } + + /** Test helper: simulate a pushed frame. */ + emitMessage(msg: T): void { + for (const handler of this.messageHandlers) handler(msg); + } +} + +interface FakeSdk { + sdkClient: SdkClient; + contractStatusStreams: FakeWebSocketStream[]; + orderStreams: FakeWebSocketStream[]; +} + +// By default the fake ack's the subscribe on the next microtask, mirroring +// a real subscribe ack arriving asynchronously over the wire. +function createFakeSdkClient(options?: { autoAck?: boolean }): FakeSdk { + const autoAck = options?.autoAck ?? true; + const contractStatusStreams: FakeWebSocketStream[] = []; + const orderStreams: FakeWebSocketStream[] = []; + + const sdkClient = { + websocket: { + public: { + contractStatus: () => { + const stream = new FakeWebSocketStream(); + contractStatusStreams.push(stream); + if (autoAck) queueMicrotask(() => stream.ackReady()); + return stream; + }, + }, + private: { + orders: () => { + const stream = new FakeWebSocketStream(); + orderStreams.push(stream); + if (autoAck) queueMicrotask(() => stream.ackReady()); + return stream; + }, + }, + }, + } as unknown as SdkClient; + + return { sdkClient, contractStatusStreams, orderStreams }; +} + +test('subscribe() sends the wire channel name with correct casing for spot vs prediction symbols', async () => { + // subscribe()/subscribeMultiple() still go through the legacy wire client — + // untouched by this migration — so this test keeps a real echo server. const receivedParams: string[] = []; const wss = new WebSocketServer({ port: 0 }); await new Promise((resolve) => wss.once('listening', resolve)); @@ -94,141 +195,124 @@ async function withEchoServer(run: (manager: WebSocketManager, receivedParams: s }); const { port } = wss.address() as { port: number }; - const manager = new WebSocketManager(`ws://localhost:${port}`); + const { sdkClient } = createFakeSdkClient(); + const manager = new WebSocketManager(`ws://localhost:${port}`, sdkClient); try { await manager.initialize(); - await run(manager, receivedParams); - } finally { - manager.disconnect(); - await new Promise((resolve, reject) => wss.close((err) => (err ? reject(err) : resolve()))); - } -} - -test('subscribe() sends the wire channel name with correct casing for spot vs prediction symbols', async () => { - await withEchoServer(async (manager, receivedParams) => { await manager.subscribe('GEMI-PRES2028-VANCE', 'bookTicker'); await manager.subscribe('BTCUSD', 'bookTicker'); assert.ok(receivedParams.includes('GEMI-PRES2028-VANCE@bookTicker')); assert.ok(receivedParams.includes('btcusd@bookTicker')); - }); + } finally { + manager.disconnect(); + await new Promise((resolve, reject) => wss.close((err) => (err ? reject(err) : resolve()))); + } }); -test('subscribeContractStatus() sends the literal global channel name, with no symbol prefix', async () => { - await withEchoServer(async (manager, receivedParams) => { - await manager.subscribeContractStatus(); +test('subscribeContractStatus() subscribes exactly once through the SDK public stream', async () => { + const { sdkClient, contractStatusStreams } = createFakeSdkClient(); + const manager = new WebSocketManager('ws://unused', sdkClient); - assert.deepStrictEqual(receivedParams, ['contractStatus']); - }); + await manager.subscribeContractStatus(); + + assert.strictEqual(contractStatusStreams.length, 1); + assert.deepStrictEqual(manager.getState().subscriptions, ['contractStatus']); }); -test('subscribeContractStatus() is idempotent — a second call does not re-subscribe over the wire', async () => { - await withEchoServer(async (manager, receivedParams) => { - await manager.subscribeContractStatus(); - await manager.subscribeContractStatus(); +test('subscribeContractStatus() is idempotent — a second call does not open a second SDK stream', async () => { + const { sdkClient, contractStatusStreams } = createFakeSdkClient(); + const manager = new WebSocketManager('ws://unused', sdkClient); - assert.deepStrictEqual(receivedParams, ['contractStatus']); - }); + await manager.subscribeContractStatus(); + await manager.subscribeContractStatus(); + + assert.strictEqual(contractStatusStreams.length, 1); }); test('subscribeContractStatus() deduplicates truly concurrent callers', async () => { - await withEchoServer(async (manager, receivedParams) => { - // Both calls start before either has awaited anything, so both would - // observe "not subscribed yet" without the pendingSubscriptions guard — - // this is the race the fix in subscribeOnce() targets, distinct from - // the sequential idempotency case above. - await Promise.all([manager.subscribeContractStatus(), manager.subscribeContractStatus()]); - - assert.deepStrictEqual(receivedParams, ['contractStatus']); - }); -}); + const { sdkClient, contractStatusStreams } = createFakeSdkClient(); + const manager = new WebSocketManager('ws://unused', sdkClient); -test('a real contractStatus wire message lands in the store with the contract ID intact', async () => { - const wss = new WebSocketServer({ port: 0 }); - await new Promise((resolve) => wss.once('listening', resolve)); + // Both calls start before either has awaited anything, so both would + // observe "not subscribed yet" without the pendingSubscriptions guard — + // this is the race the fix in subscribeOnce() targets, distinct from the + // sequential idempotency case above. + await Promise.all([manager.subscribeContractStatus(), manager.subscribeContractStatus()]); - // 17-18 digit contract IDs exceed Number.MAX_SAFE_INTEGER. Built as a raw - // string, not a JS numeric literal — `145828833218573125` as source code - // would itself get rounded by V8 at parse time, before it's even sent. - const BIG_CONTRACT_ID = '145828833218573125'; + assert.strictEqual(contractStatusStreams.length, 1); +}); - wss.on('connection', (socket) => { - socket.on('message', (data: RawData) => { - const msg = JSON.parse(data.toString()) as { id: string; params: string[] }; - socket.send(JSON.stringify({ id: msg.id, result: msg.params })); - socket.send( - `{"e":"contractStatus","E":1700000000000,"s":"GEMI-PRES2028-VANCE",` + - `"k":"PRES2028","c":"GEMI-PRES2028-VANCE","i":${BIG_CONTRACT_ID},` + - `"p":"0.50","o":"active","n":"settled"}` - ); - }); - }); +test('an SDK contractStatus frame lands in the store with the contract ID intact', async () => { + // 17-18 digit contract IDs exceed Number.MAX_SAFE_INTEGER. The SDK's own + // lossless parser hands these back as `bigint`, not `number` — modeled + // here as a real bigint, not a numeric literal, since a JS literal at + // this magnitude would itself round at parse time before we even get to + // the adapter under test. + const BIG_CONTRACT_ID = 145828833218573125n; - const { port } = wss.address() as { port: number }; - const manager = new WebSocketManager(`ws://localhost:${port}`); + const { sdkClient, contractStatusStreams } = createFakeSdkClient(); + const manager = new WebSocketManager('ws://unused', sdkClient); const beforeIngestion = Date.now(); - try { - await manager.initialize(); - await manager.subscribeContractStatus(); + await manager.subscribeContractStatus(); + contractStatusStreams[0]!.emitMessage({ + e: 'contractStatus', + E: 1_700_000_000_000, + s: 'GEMI-PRES2028-VANCE', + k: 'PRES2028', + c: 'GEMI-PRES2028-VANCE', + i: BIG_CONTRACT_ID, + p: '0.50', + o: 'active', + n: 'settled', + }); - const store = manager.getStore(); - const status = await waitForContractStatus(store, 'GEMI-PRES2028-VANCE'); - - const afterIngestion = Date.now(); - const { timestamp, ...rest } = status as CachedContractStatus; - - assert.deepStrictEqual(rest, { - symbol: 'GEMI-PRES2028-VANCE', - eventTicker: 'PRES2028', - contractTicker: 'GEMI-PRES2028-VANCE', - contractId: BIG_CONTRACT_ID, - previousStatus: 'active', - newStatus: 'settled', - strikePrice: '0.50', - eventTimeMs: 1_700_000_000_000, - }); - // `timestamp` is our own receipt-time bookkeeping (set via Date.now() at - // ingestion), distinct from the exchange's `eventTimeMs` — assert it - // against a real bound instead of comparing the object to itself. - assert.ok(timestamp >= beforeIngestion && timestamp <= afterIngestion); - } finally { - manager.disconnect(); - await new Promise((resolve, reject) => wss.close((err) => (err ? reject(err) : resolve()))); - } + const store = manager.getStore(); + const status = await waitForContractStatus(store, 'GEMI-PRES2028-VANCE'); + const afterIngestion = Date.now(); + const { timestamp, ...rest } = status as CachedContractStatus; + + assert.deepStrictEqual(rest, { + symbol: 'GEMI-PRES2028-VANCE', + eventTicker: 'PRES2028', + contractTicker: 'GEMI-PRES2028-VANCE', + contractId: '145828833218573125', + previousStatus: 'active', + newStatus: 'settled', + strikePrice: '0.50', + eventTimeMs: 1_700_000_000_000, + }); + // `timestamp` is our own receipt-time bookkeeping (set via Date.now() at + // ingestion), distinct from the exchange's `eventTimeMs` — assert it + // against a real bound instead of comparing the object to itself. + assert.ok(timestamp >= beforeIngestion && timestamp <= afterIngestion); }); -test('a contractStatus wire message without a strike price leaves strikePrice unset', async () => { - const wss = new WebSocketServer({ port: 0 }); - await new Promise((resolve) => wss.once('listening', resolve)); - - wss.on('connection', (socket) => { - socket.on('message', (data: RawData) => { - const msg = JSON.parse(data.toString()) as { id: string; params: string[] }; - socket.send(JSON.stringify({ id: msg.id, result: msg.params })); - // A real settlement event omits `p` entirely — only strike-setting - // events (and some contract types) carry a strike price. - socket.send('{"e":"contractStatus","E":1700000000000,"s":"GEMI-NOSTRIKE","k":"NS","c":"GEMI-NOSTRIKE","i":2,"o":"active","n":"settled"}'); - }); +test('an SDK contractStatus frame without a strike price leaves strikePrice unset', async () => { + const { sdkClient, contractStatusStreams } = createFakeSdkClient(); + const manager = new WebSocketManager('ws://unused', sdkClient); + + await manager.subscribeContractStatus(); + // A real settlement event omits `p` entirely — only strike-setting events + // (and some contract types) carry a strike price. + contractStatusStreams[0]!.emitMessage({ + e: 'contractStatus', + E: 1_700_000_000_000, + s: 'GEMI-NOSTRIKE', + k: 'NS', + c: 'GEMI-NOSTRIKE', + i: 2, + o: 'active', + n: 'settled', }); - const { port } = wss.address() as { port: number }; - const manager = new WebSocketManager(`ws://localhost:${port}`); - - try { - await manager.initialize(); - await manager.subscribeContractStatus(); + const store = manager.getStore(); + const status = await waitForContractStatus(store, 'GEMI-NOSTRIKE'); - const store = manager.getStore(); - const status = await waitForContractStatus(store, 'GEMI-NOSTRIKE'); - - assert.strictEqual(status?.newStatus, 'settled'); - assert.strictEqual(status?.strikePrice, undefined); - } finally { - manager.disconnect(); - await new Promise((resolve, reject) => wss.close((err) => (err ? reject(err) : resolve()))); - } + assert.strictEqual(status?.newStatus, 'settled'); + assert.strictEqual(status?.strikePrice, undefined); }); function withCredentials(apiKey: string, apiSecret: string, run: () => Promise): Promise { @@ -241,36 +325,46 @@ function withCredentials(apiKey: string, apiSecret: string, run: () => Promise { - await withCredentials('test-key', 'test-secret', () => - withEchoServer(async (manager, receivedParams) => { - await manager.subscribeAccountOrders(); - assert.deepStrictEqual(receivedParams, ['orders@account']); - }) - ); +test('subscribeAccountOrders() subscribes exactly once through the SDK private stream', async () => { + await withCredentials('test-key', 'test-secret', async () => { + const { sdkClient, orderStreams } = createFakeSdkClient(); + const manager = new WebSocketManager('ws://unused', sdkClient); + + await manager.subscribeAccountOrders(); + + assert.strictEqual(orderStreams.length, 1); + assert.deepStrictEqual(manager.getState().subscriptions, ['orders@account']); + }); }); -test('subscribeAccountOrders() throws clearly, without touching the wire, when credentials are not configured', async () => { - await withCredentials('', '', () => - withEchoServer(async (manager, receivedParams) => { - await assert.rejects(() => manager.subscribeAccountOrders(), /GEMINI_API_KEY and GEMINI_API_SECRET/); - assert.deepStrictEqual(receivedParams, []); - }) - ); +test('subscribeAccountOrders() throws clearly, without touching the SDK, when credentials are not configured', async () => { + await withCredentials('', '', async () => { + const { sdkClient, orderStreams } = createFakeSdkClient(); + const manager = new WebSocketManager('ws://unused', sdkClient); + + await assert.rejects(() => manager.subscribeAccountOrders(), /GEMINI_API_KEY and GEMINI_API_SECRET/); + assert.strictEqual(orderStreams.length, 0); + }); }); test('subscribeAccountOrders() deduplicates truly concurrent callers', async () => { - await withCredentials('test-key', 'test-secret', () => - withEchoServer(async (manager, receivedParams) => { - await Promise.all([manager.subscribeAccountOrders(), manager.subscribeAccountOrders()]); - assert.deepStrictEqual(receivedParams, ['orders@account']); - }) - ); + await withCredentials('test-key', 'test-secret', async () => { + const { sdkClient, orderStreams } = createFakeSdkClient(); + const manager = new WebSocketManager('ws://unused', sdkClient); + + await Promise.all([manager.subscribeAccountOrders(), manager.subscribeAccountOrders()]); + + assert.strictEqual(orderStreams.length, 1); + }); }); -test('a fill-shaped orderUpdate wire message lands in the order store, not the trade store', async () => { - const BIG_ORDER_ID = '145828833218573125'; // exceeds Number.MAX_SAFE_INTEGER - const BIG_TRADE_ID = '298374652910473625'; +test('a fill-shaped SDK orderUpdate frame lands in the order store, not the trade store', async () => { + // Both exceed Number.MAX_SAFE_INTEGER — modeled as real bigints, matching + // what the SDK's lossless parser actually hands back for IDs at this + // magnitude (see the contractStatus big-ID test above for why a numeric + // literal here wouldn't prove anything). + const BIG_ORDER_ID = 145828833218573125n; + const BIG_TRADE_ID = 298374652910473625n; // Order events use nanosecond timestamps, same convention as trade/ // bookTicker/depth/ticker — confirmed against the live AsyncAPI spec, // sdk-typescript's runtime validator, and directly against real @@ -279,103 +373,119 @@ test('a fill-shaped orderUpdate wire message lands in the order store, not the t // realistic 19-digit nanosecond value here, not a misleadingly small // ms-shaped one, so this test's own fixture can't be misread as evidence // that E is milliseconds. - const NANOS_E = '1789420240479000000'; + const NANOS_E = 1_789_420_240_479_000_000n; const expectedEventTimeMs = Math.floor(Number(NANOS_E) / 1_000_000); await withCredentials('test-key', 'test-secret', async () => { - const wss = new WebSocketServer({ port: 0 }); - await new Promise((resolve) => wss.once('listening', resolve)); - - wss.on('connection', (socket) => { - socket.on('message', (data: RawData) => { - const msg = JSON.parse(data.toString()) as { id: string; params: string[] }; - socket.send(JSON.stringify({ id: msg.id, result: msg.params })); - // A fill event: e:'orderUpdate' but ALSO carries t/q/m — the exact - // fields isTradeMessage duck-types on. This is the end-to-end proof - // that the ordering fix in handleMessage actually holds, not just - // the isOrderUpdateMessage guard in isolation. - socket.send( - `{"e":"orderUpdate","E":${NANOS_E},"T":${NANOS_E},` + - `"s":"GEMI-PRES2028-VANCE","i":${BIG_ORDER_ID},"c":"my-client-id",` + - `"S":"BUY","o":"LIMIT","X":"FILLED","O":"YES","p":"0.27","q":"100",` + - `"z":"0","Z":"100","L":"0.27","t":${BIG_TRADE_ID},"n":"0.01","m":true}` - ); - }); + const { sdkClient, orderStreams } = createFakeSdkClient(); + const manager = new WebSocketManager('ws://unused', sdkClient); + + await manager.subscribeAccountOrders(); + + // A fill event: e:'orderUpdate' but ALSO carries t/q/m — the exact + // fields the legacy isTradeMessage guard duck-typed on. Proves the SDK + // path still routes fills to the order store, not the trade store, now + // that the discriminator-ordering fix in the old handleMessage no + // longer applies (contractStatus/orderUpdate never reach handleMessage + // at all any more). + orderStreams[0]!.emitMessage({ + e: 'orderUpdate', + E: NANOS_E, + T: NANOS_E, + s: 'GEMI-PRES2028-VANCE', + i: BIG_ORDER_ID, + c: 'my-client-id', + S: 'BUY', + o: 'LIMIT', + X: 'FILLED', + O: 'YES', + p: '0.27', + q: '100', + z: '0', + Z: '100', + L: '0.27', + t: BIG_TRADE_ID, + n: '0.01', + m: true, }); - const { port } = wss.address() as { port: number }; - const manager = new WebSocketManager(`ws://localhost:${port}`); - - try { - await manager.initialize(); - await manager.subscribeAccountOrders(); - - const store = manager.getStore(); - const order = await waitForOrder(store, BIG_ORDER_ID); - - assert.ok(order, 'order update must have been captured'); - assert.strictEqual(order?.orderId, BIG_ORDER_ID); - assert.strictEqual(order?.tradeId, BIG_TRADE_ID); - assert.strictEqual(order?.status, 'FILLED'); - assert.strictEqual(order?.outcome, 'YES'); - assert.strictEqual(order?.symbol, 'GEMI-PRES2028-VANCE'); - assert.strictEqual(order?.isMaker, true); - assert.strictEqual(order?.eventTimeMs, expectedEventTimeMs); - // Sanity bound proving this is genuinely millisecond-scale (sometime - // after 2001), not the raw nanosecond value passed through unconverted. - assert.ok(order!.eventTimeMs > 1_000_000_000_000 && order!.eventTimeMs < 10_000_000_000_000); - - // The actual regression check: this must NOT have also landed in the - // trade/price cache via isTradeMessage's duck-typed match. - assert.strictEqual(store.getTrades('GEMI-PRES2028-VANCE').length, 0); - assert.strictEqual(store.getPrice('GEMI-PRES2028-VANCE'), undefined); - } finally { - manager.disconnect(); - await new Promise((resolve, reject) => wss.close((err) => (err ? reject(err) : resolve()))); - } + const store = manager.getStore(); + const order = await waitForOrder(store, '145828833218573125'); + + assert.ok(order, 'order update must have been captured'); + assert.strictEqual(order?.orderId, '145828833218573125'); + assert.strictEqual(order?.tradeId, '298374652910473625'); + assert.strictEqual(order?.status, 'FILLED'); + assert.strictEqual(order?.outcome, 'YES'); + assert.strictEqual(order?.symbol, 'GEMI-PRES2028-VANCE'); + assert.strictEqual(order?.isMaker, true); + assert.strictEqual(order?.eventTimeMs, expectedEventTimeMs); + // Sanity bound proving this is genuinely millisecond-scale (sometime + // after 2001), not the raw nanosecond value passed through unconverted. + assert.ok(order!.eventTimeMs > 1_000_000_000_000 && order!.eventTimeMs < 10_000_000_000_000); + + // The actual regression check: this must NOT have also landed in the + // trade/price cache. + assert.strictEqual(store.getTrades('GEMI-PRES2028-VANCE').length, 0); + assert.strictEqual(store.getPrice('GEMI-PRES2028-VANCE'), undefined); }); }); -test('a canceled orderUpdate wire message is captured with its reject reason', async () => { +test('a canceled SDK orderUpdate frame is captured with its reject reason', async () => { // The only other manager-level order fixture is a FILLED event — this // covers the non-fill terminal case (CANCELED, with a reject/cancel // reason and nothing executed) so a regression that drops `r` or // mishandles a terminal state without a fill wouldn't pass unnoticed. await withCredentials('test-key', 'test-secret', async () => { - const wss = new WebSocketServer({ port: 0 }); - await new Promise((resolve) => wss.once('listening', resolve)); - - wss.on('connection', (socket) => { - socket.on('message', (data: RawData) => { - const msg = JSON.parse(data.toString()) as { id: string; params: string[] }; - socket.send(JSON.stringify({ id: msg.id, result: msg.params })); - socket.send( - '{"e":"orderUpdate","E":1789420240479000000,"T":1789420240479000000,' + - '"s":"GEMI-CANCEL-CHECK","i":2,"S":"BUY","o":"LIMIT","X":"CANCELED",' + - '"O":"YES","p":"0.01","q":"1","z":"1","Z":"0","r":"Requested"}' - ); - }); + const { sdkClient, orderStreams } = createFakeSdkClient(); + const manager = new WebSocketManager('ws://unused', sdkClient); + + await manager.subscribeAccountOrders(); + + orderStreams[0]!.emitMessage({ + e: 'orderUpdate', + E: 1_789_420_240_479_000_000n, + T: 1_789_420_240_479_000_000n, + s: 'GEMI-CANCEL-CHECK', + i: 2, + S: 'BUY', + o: 'LIMIT', + X: 'CANCELED', + O: 'YES', + p: '0.01', + q: '1', + z: '1', + Z: '0', + r: 'Requested', }); - const { port } = wss.address() as { port: number }; - const manager = new WebSocketManager(`ws://localhost:${port}`); - - try { - await manager.initialize(); - await manager.subscribeAccountOrders(); - - const store = manager.getStore(); - const order = await waitForOrder(store, '2'); - - assert.ok(order, 'order update must have been captured'); - assert.strictEqual(order?.status, 'CANCELED'); - assert.strictEqual(order?.rejectReason, 'Requested'); - assert.strictEqual(order?.remainingQty, '1'); - assert.strictEqual(order?.executedQty, '0'); - assert.strictEqual(order?.tradeId, undefined); - } finally { - manager.disconnect(); - await new Promise((resolve, reject) => wss.close((err) => (err ? reject(err) : resolve()))); - } + const store = manager.getStore(); + const order = await waitForOrder(store, '2'); + + assert.ok(order, 'order update must have been captured'); + assert.strictEqual(order?.status, 'CANCELED'); + assert.strictEqual(order?.rejectReason, 'Requested'); + assert.strictEqual(order?.remainingQty, '1'); + assert.strictEqual(order?.executedQty, '0'); + assert.strictEqual(order?.tradeId, undefined); }); }); + +test('subscribeContractStatus() propagates a rejected subscribe ack and does not record a subscription', async () => { + const { sdkClient, contractStatusStreams } = createFakeSdkClient({ autoAck: false }); + const manager = new WebSocketManager('ws://unused', sdkClient); + + const attempt = manager.subscribeContractStatus(); + contractStatusStreams[0]!.failReady(new Error('subscribe rejected with status 400')); + + await assert.rejects(() => attempt, /subscribe rejected with status 400/); + assert.deepStrictEqual(manager.getState().subscriptions, []); + + // A retry after the failure must open a fresh stream rather than being + // treated as already subscribed or stuck on the failed pending promise. + const retry = manager.subscribeContractStatus(); + contractStatusStreams[1]!.ackReady(); + await retry; + assert.strictEqual(contractStatusStreams.length, 2); + assert.deepStrictEqual(manager.getState().subscriptions, ['contractStatus']); +}); diff --git a/packages/mcp-server/src/websocket/manager.ts b/packages/mcp-server/src/websocket/manager.ts index 38e9b03..4e79db9 100644 --- a/packages/mcp-server/src/websocket/manager.ts +++ b/packages/mcp-server/src/websocket/manager.ts @@ -1,7 +1,9 @@ -import { GeminiWebSocketClient, isTradeMessage, isDepthMessage, isBookTickerMessage, isTickerMessage, isContractStatusMessage, isOrderUpdateMessage, isSubscribeResponse } from '../client/websocket.js'; +import { GeminiWebSocketClient, isTradeMessage, isDepthMessage, isBookTickerMessage, isTickerMessage, isSubscribeResponse } from '../client/websocket.js'; import { MarketDataStore } from '../store/index.js'; import { config } from '../config.js'; -import type { WSMessage, WSConnectionStatus, WSManagerState, WSChannel } from '../types/websocket.js'; +import type { SdkClient } from '../client/sdk.js'; +import type { ContractStatus, OrderUpdate, WebSocketStream } from '@gemini-markets/sdk/server'; +import type { WSMessage, WSConnectionStatus, WSManagerState, WSChannel, CachedOrderUpdate } from '../types/websocket.js'; /** * Normalize a symbol into the casing Gemini's WS channel names expect. @@ -28,11 +30,66 @@ export function toEventTimeMs(rawTimestamp: number): number { return rawTimestamp >= NANOSECOND_MAGNITUDE_THRESHOLD ? Math.floor(rawTimestamp / 1_000_000) : rawTimestamp; } +// The SDK's lossless WebSocket parser types large integer fields (contract/ +// order/trade IDs) as `number | bigint` to avoid the precision loss a plain +// JSON.parse would cause on 17-18 digit values — see stream.ts's +// estimateFrameBytes and the SDK's own websocket.test.ts. `String()` on +// either a safe-integer `number` or a `bigint` yields the exact decimal +// digits with no exponential notation, which is exactly what this package's +// Cached*/store types (string IDs) need. +function idToString(id: number | bigint): string { + return String(id); +} + +/** + * Reshape the SDK's `ContractStatus` push frame into the positional args + * `MarketDataStore.updateContractStatus` expects. Contract status's `E` is + * milliseconds already (unlike order/trade/bookTicker's nanosecond-scale + * `E`), so it's passed straight through — no toEventTimeMs conversion, same + * as the legacy wire handler this replaces. + */ +function contractStatusArgsFromSdk(msg: ContractStatus): Parameters { + return [msg.s, msg.k, msg.c, idToString(msg.i), msg.o, msg.n, msg.p, Number(msg.E)]; +} + +/** + * Reshape the SDK's `OrderUpdate` push frame into the object + * `MarketDataStore.updateOrder` expects. The enum-typed fields (`S`/`o`/`X`/ + * `O`) carry the same literal string values as this package's Cached* + * unions at runtime; the cast just bridges the generated enum's nominal + * type to those plain string-literal unions. + */ +function orderUpdateFromSdk(msg: OrderUpdate): Omit { + return { + orderId: idToString(msg.i), + clientOrderId: msg.c, + symbol: msg.s, + side: msg.S as unknown as 'BUY' | 'SELL' | undefined, + orderType: msg.o as unknown as string | undefined, + status: msg.X as unknown as string, + outcome: msg.O as unknown as 'YES' | 'NO' | undefined, + price: msg.p, + stopPrice: msg.P, + quantity: msg.q, + remainingQty: msg.z, + executedQty: msg.Z, + lastExecutedPrice: msg.L, + tradeId: msg.t !== undefined ? idToString(msg.t) : undefined, + feeAmount: msg.n, + isMaker: msg.m, + rejectReason: msg.r, + // Same magnitude-detection treatment as the legacy orderUpdate wire + // handler applied to this field — see toEventTimeMs's doc comment. + eventTimeMs: toEventTimeMs(Number(msg.E)), + }; +} + /** * WebSocket manager that integrates client and store */ export class WebSocketManager { private client: GeminiWebSocketClient; + private sdkClient: SdkClient; private store: MarketDataStore; private status: WSConnectionStatus = 'disconnected'; private lastConnected?: number; @@ -43,10 +100,20 @@ export class WebSocketManager { // (the store isn't updated until the wire call resolves), so both send a // subscribe request — Gemini can deliver duplicate events or reject the // second one. Concurrent callers now await the same in-flight promise. + // Shared by both the legacy wire subscribe() path and the SDK-backed + // contractStatus/orders@account streams below — the channel-name keys + // ('btcusd@bookTicker' vs 'contractStatus'/'orders@account') never + // collide, so one map safely dedupes both. private pendingSubscriptions: Map> = new Map(); + // SDK-backed public/private streams for contractStatus and orders@account. + // Kept so disconnect() can release them; only set once subscribed + // successfully (see subscribeOnce below). + private contractStatusStream?: WebSocketStream; + private orderUpdateStream?: WebSocketStream; - constructor(wsUrl: string, store?: MarketDataStore) { + constructor(wsUrl: string, sdkClient: SdkClient, store?: MarketDataStore) { this.client = new GeminiWebSocketClient(wsUrl); + this.sdkClient = sdkClient; this.store = store || new MarketDataStore(); // Register message handler @@ -75,8 +142,12 @@ export class WebSocketManager { /** * Subscribe to a single channel, deduplicating concurrent callers and * recording the subscription only once the wire call actually succeeds. + * `doSubscribe` performs the actual subscribe — defaulting to the legacy + * wire client for bookTicker/trade/depth/ticker channels — so + * contractStatus/orders@account can plug in the SDK-backed streams below + * while sharing this same dedup guard. */ - private subscribeOnce(channelStr: string): Promise { + private subscribeOnce(channelStr: string, doSubscribe?: () => Promise): Promise { if (this.store.hasSubscription(channelStr)) { console.error(`[WSManager] Already subscribed to ${channelStr}`); return Promise.resolve(); @@ -85,8 +156,9 @@ export class WebSocketManager { const pending = this.pendingSubscriptions.get(channelStr); if (pending) return pending; - const promise = this.client - .subscribe([channelStr]) + const subscribeAction = doSubscribe ?? (() => this.client.subscribe([channelStr]).then(() => undefined)); + + const promise = subscribeAction() .then(() => { this.store.addSubscription(channelStr); console.error(`[WSManager] Subscribed to ${channelStr}`); @@ -183,27 +255,51 @@ export class WebSocketManager { /** * Subscribe to the global contractStatus channel (prediction-market - * strike/settlement lifecycle events). Unlike bookTicker/trade/depth, - * this has no per-symbol wire subscription — Gemini pushes every - * contract's status changes on one shared channel (confirmed against - * sdk-go's SubscribeContractStatus, which sends the literal channel name - * "contractStatus" regardless of the symbol callers filter by). Callers - * read a specific symbol's latest status back out of the store. + * strike/settlement lifecycle events) via the SDK's public + * `client.websocket.public.contractStatus()` stream. Unlike bookTicker/ + * trade/depth, this has no per-symbol wire subscription — Gemini pushes + * every contract's status changes on one shared channel (confirmed + * against sdk-go's SubscribeContractStatus, which sends the literal + * channel name "contractStatus" regardless of the symbol callers filter + * by). Callers read a specific symbol's latest status back out of the + * store. Reconnect/backoff for this stream is handled inside the SDK. */ async subscribeContractStatus(): Promise { - return this.subscribeOnce('contractStatus'); + return this.subscribeOnce('contractStatus', () => this.startContractStatusStream()); + } + + private async startContractStatusStream(): Promise { + const stream = this.sdkClient.websocket.public.contractStatus(); + stream.on('message', (msg) => this.handleContractStatusMessage(msg)); + try { + await stream.ready; + } catch (err) { + void stream.close(); + throw err; + } + // Only retained once the subscribe ack lands, matching subscribeOnce's + // "record the subscription only once the wire call actually succeeds". + this.contractStatusStream = stream; + } + + private handleContractStatusMessage(msg: ContractStatus): void { + try { + this.store.updateContractStatus(...contractStatusArgsFromSdk(msg)); + } catch (err) { + console.error('[WSManager] Error handling contractStatus message:', err); + } } /** * Subscribe to the authenticated orders@account channel (fill/cancel/ - * reject confirmation for every order on the account). Like + * reject confirmation for every order on the account) via the SDK's + * `client.websocket.private.orders({ scope: 'account' })` stream. Like * contractStatus, this is one global channel — no per-symbol wire - * subscription — and requires credentials at the WebSocket connection - * upgrade itself (handled in GeminiWebSocketClient.connect()), not a - * post-connect handshake. Throws immediately if credentials aren't - * configured, mirroring GeminiHttpClient.authenticatedPost's guard, - * rather than attempting the subscribe and getting a confusing late - * rejection from Gemini. + * subscription. Throws immediately if credentials aren't configured, + * mirroring GeminiHttpClient.authenticatedPost's guard, rather than + * attempting the subscribe and getting a confusing late rejection from + * either Gemini or the SDK's own "authenticated WebSocket operation + * requires auth" error. */ async subscribeAccountOrders(): Promise { if (!config.apiKey || !config.apiSecret) { @@ -212,7 +308,27 @@ export class WebSocketManager { 'must be set in the MCP server environment. This tool is unavailable in public-only mode.' ); } - return this.subscribeOnce('orders@account'); + return this.subscribeOnce('orders@account', () => this.startAccountOrdersStream()); + } + + private async startAccountOrdersStream(): Promise { + const stream = this.sdkClient.websocket.private.orders({ scope: 'account' }); + stream.on('message', (msg) => this.handleOrderUpdateMessage(msg)); + try { + await stream.ready; + } catch (err) { + void stream.close(); + throw err; + } + this.orderUpdateStream = stream; + } + + private handleOrderUpdateMessage(msg: OrderUpdate): void { + try { + this.store.updateOrder(orderUpdateFromSdk(msg)); + } catch (err) { + console.error('[WSManager] Error handling orderUpdate message:', err); + } } /** @@ -225,54 +341,12 @@ export class WebSocketManager { return; } - // Handle contract status messages (checked ahead of the legacy - // duck-typed guards below on principle — see toChannelSymbol's sibling - // fix for why an unrelated guard silently swallowing a new message - // shape is the kind of bug worth guarding against up front). - if (isContractStatusMessage(message)) { - this.store.updateContractStatus( - message.s, - message.k, - message.c, - message.i, - message.o, - message.n, - message.p, - message.E - ); - return; - } - - // Handle authenticated order lifecycle events — checked before - // isTradeMessage below on purpose. A fill event's t/q/m fields - // duck-type match isTradeMessage's check exactly; sdk-go's own - // dispatcher hit this and fixed it the same way (explicit - // discriminator first). Getting this order wrong means a fill - // silently corrupts the spot price/trade cache instead of reaching - // the order store. - if (isOrderUpdateMessage(message)) { - this.store.updateOrder({ - orderId: message.i, - clientOrderId: message.c, - symbol: message.s, - side: message.S, - orderType: message.o, - status: message.X, - outcome: message.O, - price: message.p, - stopPrice: message.P, - quantity: message.q, - remainingQty: message.z, - executedQty: message.Z, - lastExecutedPrice: message.L, - tradeId: message.t, - feeAmount: message.n, - isMaker: message.m, - rejectReason: message.r, - eventTimeMs: toEventTimeMs(message.E), - }); - return; - } + // contractStatus and orderUpdate/order messages no longer arrive here — + // subscribeContractStatus()/subscribeAccountOrders() now push those + // through the SDK's own WebSocketStream objects (see + // startContractStatusStream/startAccountOrdersStream above), which + // route straight to handleContractStatusMessage/handleOrderUpdateMessage + // without going through the legacy client's message handler at all. // Handle trade messages if (isTradeMessage(message)) { @@ -350,6 +424,12 @@ export class WebSocketManager { */ disconnect(): void { this.client.disconnect(); + // Release the SDK-backed contractStatus/orders@account streams too, if + // any were ever established — otherwise their underlying WebSocket + // sessions (and the SDK's own reconnect loop for them) would outlive + // this manager. + void this.contractStatusStream?.close(); + void this.orderUpdateStream?.close(); this.status = 'disconnected'; console.error('[WSManager] Disconnected'); } From 52f3c100a5bc9bbb9ead6d4cf1f86033cbafbd9e Mon Sep 17 00:00:00 2001 From: Sohum Desai Date: Tue, 22 Sep 2026 14:15:39 -0400 Subject: [PATCH 2/4] fix(mcp-server): address Semgrep false-positive on PREDICT-8823 The test-only 'ws://unused' placeholder passed to WebSocketManager's constructor never opens a socket (these tests exercise only the SDK-backed fake-client path), but Semgrep's detect-insecure-websocket rule pattern-matches the literal string regardless. Renamed to 'unused' so it no longer looks like a URL. Co-Authored-By: Claude Sonnet 5 --- .../mcp-server/src/websocket/manager.test.ts | 22 +++++++++---------- 1 file changed, 11 insertions(+), 11 deletions(-) diff --git a/packages/mcp-server/src/websocket/manager.test.ts b/packages/mcp-server/src/websocket/manager.test.ts index 10c088e..6fb1cae 100644 --- a/packages/mcp-server/src/websocket/manager.test.ts +++ b/packages/mcp-server/src/websocket/manager.test.ts @@ -213,7 +213,7 @@ test('subscribe() sends the wire channel name with correct casing for spot vs pr test('subscribeContractStatus() subscribes exactly once through the SDK public stream', async () => { const { sdkClient, contractStatusStreams } = createFakeSdkClient(); - const manager = new WebSocketManager('ws://unused', sdkClient); + const manager = new WebSocketManager('unused', sdkClient); await manager.subscribeContractStatus(); @@ -223,7 +223,7 @@ test('subscribeContractStatus() subscribes exactly once through the SDK public s test('subscribeContractStatus() is idempotent — a second call does not open a second SDK stream', async () => { const { sdkClient, contractStatusStreams } = createFakeSdkClient(); - const manager = new WebSocketManager('ws://unused', sdkClient); + const manager = new WebSocketManager('unused', sdkClient); await manager.subscribeContractStatus(); await manager.subscribeContractStatus(); @@ -233,7 +233,7 @@ test('subscribeContractStatus() is idempotent — a second call does not open a test('subscribeContractStatus() deduplicates truly concurrent callers', async () => { const { sdkClient, contractStatusStreams } = createFakeSdkClient(); - const manager = new WebSocketManager('ws://unused', sdkClient); + const manager = new WebSocketManager('unused', sdkClient); // Both calls start before either has awaited anything, so both would // observe "not subscribed yet" without the pendingSubscriptions guard — @@ -253,7 +253,7 @@ test('an SDK contractStatus frame lands in the store with the contract ID intact const BIG_CONTRACT_ID = 145828833218573125n; const { sdkClient, contractStatusStreams } = createFakeSdkClient(); - const manager = new WebSocketManager('ws://unused', sdkClient); + const manager = new WebSocketManager('unused', sdkClient); const beforeIngestion = Date.now(); await manager.subscribeContractStatus(); @@ -292,7 +292,7 @@ test('an SDK contractStatus frame lands in the store with the contract ID intact test('an SDK contractStatus frame without a strike price leaves strikePrice unset', async () => { const { sdkClient, contractStatusStreams } = createFakeSdkClient(); - const manager = new WebSocketManager('ws://unused', sdkClient); + const manager = new WebSocketManager('unused', sdkClient); await manager.subscribeContractStatus(); // A real settlement event omits `p` entirely — only strike-setting events @@ -328,7 +328,7 @@ function withCredentials(apiKey: string, apiSecret: string, run: () => Promise { await withCredentials('test-key', 'test-secret', async () => { const { sdkClient, orderStreams } = createFakeSdkClient(); - const manager = new WebSocketManager('ws://unused', sdkClient); + const manager = new WebSocketManager('unused', sdkClient); await manager.subscribeAccountOrders(); @@ -340,7 +340,7 @@ test('subscribeAccountOrders() subscribes exactly once through the SDK private s test('subscribeAccountOrders() throws clearly, without touching the SDK, when credentials are not configured', async () => { await withCredentials('', '', async () => { const { sdkClient, orderStreams } = createFakeSdkClient(); - const manager = new WebSocketManager('ws://unused', sdkClient); + const manager = new WebSocketManager('unused', sdkClient); await assert.rejects(() => manager.subscribeAccountOrders(), /GEMINI_API_KEY and GEMINI_API_SECRET/); assert.strictEqual(orderStreams.length, 0); @@ -350,7 +350,7 @@ test('subscribeAccountOrders() throws clearly, without touching the SDK, when cr test('subscribeAccountOrders() deduplicates truly concurrent callers', async () => { await withCredentials('test-key', 'test-secret', async () => { const { sdkClient, orderStreams } = createFakeSdkClient(); - const manager = new WebSocketManager('ws://unused', sdkClient); + const manager = new WebSocketManager('unused', sdkClient); await Promise.all([manager.subscribeAccountOrders(), manager.subscribeAccountOrders()]); @@ -378,7 +378,7 @@ test('a fill-shaped SDK orderUpdate frame lands in the order store, not the trad await withCredentials('test-key', 'test-secret', async () => { const { sdkClient, orderStreams } = createFakeSdkClient(); - const manager = new WebSocketManager('ws://unused', sdkClient); + const manager = new WebSocketManager('unused', sdkClient); await manager.subscribeAccountOrders(); @@ -438,7 +438,7 @@ test('a canceled SDK orderUpdate frame is captured with its reject reason', asyn // mishandles a terminal state without a fill wouldn't pass unnoticed. await withCredentials('test-key', 'test-secret', async () => { const { sdkClient, orderStreams } = createFakeSdkClient(); - const manager = new WebSocketManager('ws://unused', sdkClient); + const manager = new WebSocketManager('unused', sdkClient); await manager.subscribeAccountOrders(); @@ -473,7 +473,7 @@ test('a canceled SDK orderUpdate frame is captured with its reject reason', asyn test('subscribeContractStatus() propagates a rejected subscribe ack and does not record a subscription', async () => { const { sdkClient, contractStatusStreams } = createFakeSdkClient({ autoAck: false }); - const manager = new WebSocketManager('ws://unused', sdkClient); + const manager = new WebSocketManager('unused', sdkClient); const attempt = manager.subscribeContractStatus(); contractStatusStreams[0]!.failReady(new Error('subscribe rejected with status 400')); From 30bdc3ee027593e10320e77fe3f9acf75f81bfe9 Mon Sep 17 00:00:00 2001 From: Sohum Desai Date: Tue, 22 Sep 2026 15:02:11 -0400 Subject: [PATCH 3/4] test(mcp-server): assert account scope on PREDICT-8823's private stream fake svc-grace flagged that the private-stream fake ignored its arguments, so a regression to { scope: 'session' } would still pass every subscribeAccountOrders() test while subscribing to the wrong feed. Records the options passed to private.orders() and asserts { scope: 'account' } explicitly. Verified by temporarily breaking the real scope arg to 'session' and confirming this test fails before reverting. Co-Authored-By: Claude Sonnet 5 --- .../mcp-server/src/websocket/manager.test.ts | 16 +++++++++++++--- 1 file changed, 13 insertions(+), 3 deletions(-) diff --git a/packages/mcp-server/src/websocket/manager.test.ts b/packages/mcp-server/src/websocket/manager.test.ts index 6fb1cae..6eaccc4 100644 --- a/packages/mcp-server/src/websocket/manager.test.ts +++ b/packages/mcp-server/src/websocket/manager.test.ts @@ -146,6 +146,7 @@ interface FakeSdk { sdkClient: SdkClient; contractStatusStreams: FakeWebSocketStream[]; orderStreams: FakeWebSocketStream[]; + orderSubscriptionOptions: unknown[]; } // By default the fake ack's the subscribe on the next microtask, mirroring @@ -154,6 +155,7 @@ function createFakeSdkClient(options?: { autoAck?: boolean }): FakeSdk { const autoAck = options?.autoAck ?? true; const contractStatusStreams: FakeWebSocketStream[] = []; const orderStreams: FakeWebSocketStream[] = []; + const orderSubscriptionOptions: unknown[] = []; const sdkClient = { websocket: { @@ -166,7 +168,8 @@ function createFakeSdkClient(options?: { autoAck?: boolean }): FakeSdk { }, }, private: { - orders: () => { + orders: (opts: unknown) => { + orderSubscriptionOptions.push(opts); const stream = new FakeWebSocketStream(); orderStreams.push(stream); if (autoAck) queueMicrotask(() => stream.ackReady()); @@ -176,7 +179,7 @@ function createFakeSdkClient(options?: { autoAck?: boolean }): FakeSdk { }, } as unknown as SdkClient; - return { sdkClient, contractStatusStreams, orderStreams }; + return { sdkClient, contractStatusStreams, orderStreams, orderSubscriptionOptions }; } test('subscribe() sends the wire channel name with correct casing for spot vs prediction symbols', async () => { @@ -327,13 +330,20 @@ function withCredentials(apiKey: string, apiSecret: string, run: () => Promise { await withCredentials('test-key', 'test-secret', async () => { - const { sdkClient, orderStreams } = createFakeSdkClient(); + const { sdkClient, orderStreams, orderSubscriptionOptions } = createFakeSdkClient(); const manager = new WebSocketManager('unused', sdkClient); await manager.subscribeAccountOrders(); assert.strictEqual(orderStreams.length, 1); assert.deepStrictEqual(manager.getState().subscriptions, ['orders@account']); + // Regression guard: the account-wide channel name recorded above only + // proves *this manager* thinks it subscribed to the account scope — it + // doesn't prove that's what was actually requested from the SDK. Assert + // the literal options passed to private.orders() so a future change to + // `{ scope: 'session' }` fails here instead of silently subscribing to + // the wrong feed. + assert.deepStrictEqual(orderSubscriptionOptions, [{ scope: 'account' }]); }); }); From c5443b0e1dea48e58a8e5ff48c22be9bdf7957ee Mon Sep 17 00:00:00 2001 From: Sohum Desai Date: Tue, 22 Sep 2026 15:30:11 -0400 Subject: [PATCH 4/4] fix(mcp-server): preserve bigint precision through order timestamp division svc-grace flagged that orderUpdateFromSdk narrowed the SDK's nanosecond bigint E field to Number before dividing by 1e6, which can round the value across a millisecond boundary (doubles only have ~256ns of spacing at this magnitude). Added sdkEventTimeMs, which divides in bigint space first and only narrows the much smaller millisecond quotient. Reproduced the exact failure with a real bigint 50ns below a boundary, confirmed the new test fails against the old formula and passes against the fix. Also covers Grace's SDK-stream-cleanup finding: added tests asserting disconnect() closes both the contractStatus and orders@account SDK streams, which the fakes already tracked but nothing asserted. Co-Authored-By: Claude Sonnet 5 --- .../mcp-server/src/websocket/manager.test.ts | 56 +++++++++++++++++++ packages/mcp-server/src/websocket/manager.ts | 23 +++++++- 2 files changed, 76 insertions(+), 3 deletions(-) diff --git a/packages/mcp-server/src/websocket/manager.test.ts b/packages/mcp-server/src/websocket/manager.test.ts index 6eaccc4..9dd6c96 100644 --- a/packages/mcp-server/src/websocket/manager.test.ts +++ b/packages/mcp-server/src/websocket/manager.test.ts @@ -441,6 +441,38 @@ test('a fill-shaped SDK orderUpdate frame lands in the order store, not the trad }); }); +test('a bigint order timestamp within IEEE-754 rounding distance of a millisecond boundary is not shifted', async () => { + // 50ns before an exact millisecond boundary — well within the ~256ns + // spacing between representable doubles at this magnitude (see + // sdkEventTimeMs's doc comment). Narrowing to Number before dividing, as + // the pre-fix code did, rounds this fixture up to the boundary itself and + // lands one millisecond too high; dividing in bigint space first must not. + const BOUNDARY_MS = 1_789_420_240_479n; + const NANOS_E = BOUNDARY_MS * 1_000_000n - 50n; + const BIG_ORDER_ID = 145828833218573126n; + + await withCredentials('test-key', 'test-secret', async () => { + const { sdkClient, orderStreams } = createFakeSdkClient(); + const manager = new WebSocketManager('unused', sdkClient); + + await manager.subscribeAccountOrders(); + + orderStreams[0]!.emitMessage({ + e: 'orderUpdate', + E: NANOS_E, + s: 'GEMI-PRES2028-VANCE', + i: BIG_ORDER_ID, + X: 'NEW', + }); + + const store = manager.getStore(); + const order = await waitForOrder(store, '145828833218573126'); + + assert.ok(order, 'order update must have been captured'); + assert.strictEqual(order?.eventTimeMs, Number(BOUNDARY_MS) - 1); + }); +}); + test('a canceled SDK orderUpdate frame is captured with its reject reason', async () => { // The only other manager-level order fixture is a FILLED event — this // covers the non-fill terminal case (CANCELED, with a reject/cancel @@ -499,3 +531,27 @@ test('subscribeContractStatus() propagates a rejected subscribe ack and does not assert.strictEqual(contractStatusStreams.length, 2); assert.deepStrictEqual(manager.getState().subscriptions, ['contractStatus']); }); + +test('disconnect() closes the SDK contractStatus stream', async () => { + const { sdkClient, contractStatusStreams } = createFakeSdkClient(); + const manager = new WebSocketManager('unused', sdkClient); + + await manager.subscribeContractStatus(); + assert.strictEqual(contractStatusStreams[0]!.closed, false); + + manager.disconnect(); + assert.strictEqual(contractStatusStreams[0]!.closed, true); +}); + +test('disconnect() closes the SDK orders@account stream', async () => { + await withCredentials('test-key', 'test-secret', async () => { + const { sdkClient, orderStreams } = createFakeSdkClient(); + const manager = new WebSocketManager('unused', sdkClient); + + await manager.subscribeAccountOrders(); + assert.strictEqual(orderStreams[0]!.closed, false); + + manager.disconnect(); + assert.strictEqual(orderStreams[0]!.closed, true); + }); +}); diff --git a/packages/mcp-server/src/websocket/manager.ts b/packages/mcp-server/src/websocket/manager.ts index 4e79db9..0eaba4f 100644 --- a/packages/mcp-server/src/websocket/manager.ts +++ b/packages/mcp-server/src/websocket/manager.ts @@ -30,6 +30,23 @@ export function toEventTimeMs(rawTimestamp: number): number { return rawTimestamp >= NANOSECOND_MAGNITUDE_THRESHOLD ? Math.floor(rawTimestamp / 1_000_000) : rawTimestamp; } +// bigint-safe variant for the SDK's `number | bigint` timestamp fields. +// Narrowing a nanosecond-scale bigint to `number` *before* dividing (as +// toEventTimeMs does for its plain-number legacy callers) can round the +// value across a millisecond boundary — nanosecond epoch values are ~1.79e18 +// today, which only has ~256ns of spacing between representable doubles, so +// a raw value within that spacing of a millisecond boundary rounds up before +// the division ever happens. Dividing in bigint space first, and narrowing +// only the much smaller millisecond quotient, avoids that. +function sdkEventTimeMs(rawTimestamp: number | bigint): number { + if (typeof rawTimestamp === 'bigint') { + return rawTimestamp >= BigInt(NANOSECOND_MAGNITUDE_THRESHOLD) + ? Number(rawTimestamp / 1_000_000n) + : Number(rawTimestamp); + } + return toEventTimeMs(rawTimestamp); +} + // The SDK's lossless WebSocket parser types large integer fields (contract/ // order/trade IDs) as `number | bigint` to avoid the precision loss a plain // JSON.parse would cause on 17-18 digit values — see stream.ts's @@ -78,9 +95,9 @@ function orderUpdateFromSdk(msg: OrderUpdate): Omit