Unify live session streaming and evidence fetching

- Route live terminal data through a shared live session socket
- Fetch missing evidence for alerts and classifier hits
- Add live type definitions and storage/API tests

packages/storage/src/clickhouse.ts

@@ -418,6 +418,14 @@ const clampLimit = (limit: number): number => {
   return Math.max(1, Math.min(1000, Math.floor(limit)));
 };
 
+const clampLookupLimit = (limit: number): number => {
+  if (!Number.isFinite(limit)) {
+    return 100;
+  }
+
+  return Math.max(1, Math.min(5000, Math.floor(limit)));
+};
+
 const clampPositiveInt = (value: number, fallback = 1): number => {
   if (!Number.isFinite(value)) {
     return fallback;
@@ -450,6 +458,10 @@ const quoteString = (value: string): string => {
   return `'${escaped}'`;
 };
 
+const buildStringList = (values: string[]): string => {
+  return values.map((value) => quoteString(value)).join(", ");
+};
+
 const buildTracePrefixCondition = (tracePrefix: string | undefined): string | null => {
   if (!tracePrefix) {
     return null;
@@ -461,6 +473,15 @@ const buildTracePrefixCondition = (tracePrefix: string | undefined): string | nu
   return `startsWith(trace_id, ${quoteString(normalized)})`;
 };
 
+const buildBeforeTupleCondition = (
+  tsColumn: string,
+  seqColumn: string,
+  beforeTs: number,
+  beforeSeq: number
+): string => {
+  return `(${tsColumn}, ${seqColumn}) < (${clampCursor(beforeTs)}, ${clampCursor(beforeSeq)})`;
+};
+
 const normalizeNumericFields = (
   row: Record<string, unknown>,
   fields: string[]
@@ -1095,3 +1116,215 @@ export const fetchAlertsAfter = async (
   const alerts = records.map(fromAlertRecord);
   return AlertEventSchema.array().parse(alerts);
 };
+
+export const fetchOptionPrintsBefore = async (
+  client: ClickHouseClient,
+  beforeTs: number,
+  beforeSeq: number,
+  limit: number,
+  tracePrefix?: string
+): Promise<OptionPrint[]> => {
+  const safeLimit = clampLimit(limit);
+  const conditions = [buildBeforeTupleCondition("ts", "seq", beforeTs, beforeSeq)];
+  const traceCondition = buildTracePrefixCondition(tracePrefix);
+  if (traceCondition) {
+    conditions.push(traceCondition);
+  }
+
+  const result = await client.query({
+    query: `SELECT * FROM ${OPTION_PRINTS_TABLE} WHERE ${conditions.join(" AND ")} ORDER BY ts DESC, seq DESC LIMIT ${safeLimit}`,
+    format: "JSONEachRow"
+  });
+
+  const rows = await result.json<unknown[]>();
+  return OptionPrintSchema.array().parse(rows.map(normalizeOptionRow));
+};
+
+export const fetchOptionNBBOBefore = async (
+  client: ClickHouseClient,
+  beforeTs: number,
+  beforeSeq: number,
+  limit: number,
+  tracePrefix?: string
+): Promise<OptionNBBO[]> => {
+  const safeLimit = clampLimit(limit);
+  const conditions = [buildBeforeTupleCondition("ts", "seq", beforeTs, beforeSeq)];
+  const traceCondition = buildTracePrefixCondition(tracePrefix);
+  if (traceCondition) {
+    conditions.push(traceCondition);
+  }
+
+  const result = await client.query({
+    query: `SELECT * FROM ${OPTION_NBBO_TABLE} WHERE ${conditions.join(" AND ")} ORDER BY ts DESC, seq DESC LIMIT ${safeLimit}`,
+    format: "JSONEachRow"
+  });
+
+  const rows = await result.json<unknown[]>();
+  return OptionNBBOSchema.array().parse(rows.map(normalizeOptionNbboRow));
+};
+
+export const fetchEquityPrintsBefore = async (
+  client: ClickHouseClient,
+  beforeTs: number,
+  beforeSeq: number,
+  limit: number
+): Promise<EquityPrint[]> => {
+  const safeLimit = clampLimit(limit);
+  const result = await client.query({
+    query: `SELECT * FROM ${EQUITY_PRINTS_TABLE} WHERE ${buildBeforeTupleCondition("ts", "seq", beforeTs, beforeSeq)} ORDER BY ts DESC, seq DESC LIMIT ${safeLimit}`,
+    format: "JSONEachRow"
+  });
+
+  const rows = await result.json<unknown[]>();
+  return EquityPrintSchema.array().parse(rows.map(normalizeEquityRow));
+};
+
+export const fetchEquityPrintJoinsBefore = async (
+  client: ClickHouseClient,
+  beforeTs: number,
+  beforeSeq: number,
+  limit: number
+): Promise<EquityPrintJoin[]> => {
+  const safeLimit = clampLimit(limit);
+  const result = await client.query({
+    query: `SELECT * FROM ${EQUITY_PRINT_JOINS_TABLE} WHERE ${buildBeforeTupleCondition("source_ts", "seq", beforeTs, beforeSeq)} ORDER BY source_ts DESC, seq DESC LIMIT ${safeLimit}`,
+    format: "JSONEachRow"
+  });
+
+  const rows = await result.json<unknown[]>();
+  const records = rows
+    .map(normalizeEquityPrintJoinRow)
+    .filter((record): record is EquityPrintJoinRecord => record !== null);
+  return EquityPrintJoinSchema.array().parse(records.map(fromEquityPrintJoinRecord));
+};
+
+export const fetchFlowPacketsBefore = async (
+  client: ClickHouseClient,
+  beforeTs: number,
+  beforeSeq: number,
+  limit: number
+): Promise<FlowPacket[]> => {
+  const safeLimit = clampLimit(limit);
+  const result = await client.query({
+    query: `SELECT * FROM ${FLOW_PACKETS_TABLE} WHERE ${buildBeforeTupleCondition("source_ts", "seq", beforeTs, beforeSeq)} ORDER BY source_ts DESC, seq DESC LIMIT ${safeLimit}`,
+    format: "JSONEachRow"
+  });
+
+  const rows = await result.json<unknown[]>();
+  const records = rows
+    .map(normalizeFlowPacketRow)
+    .filter((record): record is FlowPacketRecord => record !== null);
+  return FlowPacketSchema.array().parse(records.map(fromFlowPacketRecord));
+};
+
+export const fetchClassifierHitsBefore = async (
+  client: ClickHouseClient,
+  beforeTs: number,
+  beforeSeq: number,
+  limit: number
+): Promise<ClassifierHitEvent[]> => {
+  const safeLimit = clampLimit(limit);
+  const result = await client.query({
+    query: `SELECT * FROM ${CLASSIFIER_HITS_TABLE} WHERE ${buildBeforeTupleCondition("source_ts", "seq", beforeTs, beforeSeq)} ORDER BY source_ts DESC, seq DESC LIMIT ${safeLimit}`,
+    format: "JSONEachRow"
+  });
+
+  const rows = await result.json<unknown[]>();
+  const records = rows
+    .map(normalizeClassifierHitRow)
+    .filter((record): record is ClassifierHitRecord => record !== null);
+  return ClassifierHitEventSchema.array().parse(records.map(fromClassifierHitRecord));
+};
+
+export const fetchAlertsBefore = async (
+  client: ClickHouseClient,
+  beforeTs: number,
+  beforeSeq: number,
+  limit: number
+): Promise<AlertEvent[]> => {
+  const safeLimit = clampLimit(limit);
+  const result = await client.query({
+    query: `SELECT * FROM ${ALERTS_TABLE} WHERE ${buildBeforeTupleCondition("source_ts", "seq", beforeTs, beforeSeq)} ORDER BY source_ts DESC, seq DESC LIMIT ${safeLimit}`,
+    format: "JSONEachRow"
+  });
+
+  const rows = await result.json<unknown[]>();
+  const records = rows
+    .map(normalizeAlertRow)
+    .filter((record): record is AlertRecord => record !== null);
+  return AlertEventSchema.array().parse(records.map(fromAlertRecord));
+};
+
+export const fetchInferredDarkBefore = async (
+  client: ClickHouseClient,
+  beforeTs: number,
+  beforeSeq: number,
+  limit: number
+): Promise<InferredDarkEvent[]> => {
+  const safeLimit = clampLimit(limit);
+  const result = await client.query({
+    query: `SELECT * FROM ${INFERRED_DARK_TABLE} WHERE ${buildBeforeTupleCondition("source_ts", "seq", beforeTs, beforeSeq)} ORDER BY source_ts DESC, seq DESC LIMIT ${safeLimit}`,
+    format: "JSONEachRow"
+  });
+
+  const rows = await result.json<unknown[]>();
+  const records = rows
+    .map(normalizeInferredDarkRow)
+    .filter((record): record is InferredDarkRecord => record !== null);
+  return InferredDarkEventSchema.array().parse(records.map(fromInferredDarkRecord));
+};
+
+export const fetchFlowPacketById = async (
+  client: ClickHouseClient,
+  id: string
+): Promise<FlowPacket | null> => {
+  const result = await client.query({
+    query: `SELECT * FROM ${FLOW_PACKETS_TABLE} WHERE id = ${quoteString(id)} ORDER BY source_ts DESC, seq DESC LIMIT 1`,
+    format: "JSONEachRow"
+  });
+
+  const rows = await result.json<unknown[]>();
+  const record = rows
+    .map(normalizeFlowPacketRow)
+    .find((row): row is FlowPacketRecord => row !== null);
+  return record ? FlowPacketSchema.parse(fromFlowPacketRecord(record)) : null;
+};
+
+export const fetchOptionPrintsByTraceIds = async (
+  client: ClickHouseClient,
+  traceIds: string[]
+): Promise<OptionPrint[]> => {
+  const ids = Array.from(new Set(traceIds.map((id) => id.trim()).filter(Boolean)));
+  if (ids.length === 0) {
+    return [];
+  }
+
+  const result = await client.query({
+    query: `SELECT * FROM ${OPTION_PRINTS_TABLE} WHERE trace_id IN (${buildStringList(ids)}) ORDER BY ts DESC, seq DESC LIMIT ${clampLookupLimit(ids.length)}`,
+    format: "JSONEachRow"
+  });
+
+  const rows = await result.json<unknown[]>();
+  return OptionPrintSchema.array().parse(rows.map(normalizeOptionRow));
+};
+
+export const fetchEquityPrintJoinsByIds = async (
+  client: ClickHouseClient,
+  ids: string[]
+): Promise<EquityPrintJoin[]> => {
+  const uniqueIds = Array.from(new Set(ids.map((id) => id.trim()).filter(Boolean)));
+  if (uniqueIds.length === 0) {
+    return [];
+  }
+
+  const result = await client.query({
+    query: `SELECT * FROM ${EQUITY_PRINT_JOINS_TABLE} WHERE id IN (${buildStringList(uniqueIds)}) ORDER BY source_ts DESC, seq DESC LIMIT ${clampLookupLimit(uniqueIds.length)}`,
+    format: "JSONEachRow"
+  });
+
+  const rows = await result.json<unknown[]>();
+  const records = rows
+    .map(normalizeEquityPrintJoinRow)
+    .filter((record): record is EquityPrintJoinRecord => record !== null);
+  return EquityPrintJoinSchema.array().parse(records.map(fromEquityPrintJoinRecord));
+};

packages/storage/tests/flow-packets.test.ts

@@ -1,4 +1,5 @@
 import { describe, expect, it } from "bun:test";
+import { createClickHouseClient, fetchFlowPacketById, fetchFlowPacketsBefore } from "../src/clickhouse";
 import {
   flowPacketsTableDDL,
   FLOW_PACKETS_TABLE,
@@ -36,4 +37,24 @@ describe("flow-packets storage helpers", () => {
     expect(restored.features).toEqual(packet.features);
     expect(restored.join_quality).toEqual(packet.join_quality);
   });
+
+  it("builds before-history and id lookup queries", async () => {
+    const queries: string[] = [];
+    const client = createClickHouseClient({ url: "http://127.0.0.1:8123" });
+    client.query = async ({ query }) => {
+      queries.push(query);
+      return {
+        async json<T>() {
+          return [] as T;
+        }
+      };
+    };
+
+    await fetchFlowPacketsBefore(client, 200, 3, 15);
+    await fetchFlowPacketById(client, "fp-1");
+
+    expect(queries[0]).toContain("(source_ts, seq) < (200, 3)");
+    expect(queries[0]).toContain("ORDER BY source_ts DESC, seq DESC LIMIT 15");
+    expect(queries[1]).toContain("WHERE id = 'fp-1'");
+  });
 });

packages/storage/tests/option-prints.test.ts

@@ -1,4 +1,5 @@
 import { describe, expect, it } from "bun:test";
+import { createClickHouseClient, fetchOptionPrintsBefore, fetchOptionPrintsByTraceIds } from "../src/clickhouse";
 import { normalizeOptionPrint, optionPrintsTableDDL, OPTION_PRINTS_TABLE } from "../src/option-prints";
 
 const basePrint = {
@@ -24,4 +25,25 @@ describe("option-prints storage helpers", () => {
     expect(ddl).toContain(OPTION_PRINTS_TABLE);
     expect(ddl).toContain("CREATE TABLE IF NOT EXISTS");
   });
+
+  it("builds before/history and trace lookup queries", async () => {
+    const queries: string[] = [];
+    const client = createClickHouseClient({ url: "http://127.0.0.1:8123" });
+    client.query = async ({ query }) => {
+      queries.push(query);
+      return {
+        async json<T>() {
+          return [] as T;
+        }
+      };
+    };
+
+    await fetchOptionPrintsBefore(client, 100, 5, 20, "alpaca");
+    await fetchOptionPrintsByTraceIds(client, ["trace-1", "trace-2"]);
+
+    expect(queries[0]).toContain("(ts, seq) < (100, 5)");
+    expect(queries[0]).toContain("startsWith(trace_id, 'alpaca')");
+    expect(queries[0]).toContain("ORDER BY ts DESC, seq DESC LIMIT 20");
+    expect(queries[1]).toContain("trace_id IN ('trace-1', 'trace-2')");
+  });
 });

packages/types/src/index.ts

@@ -1,2 +1,3 @@
 export * from "./events";
+export * from "./live";
 export * from "./sp500";

packages/types/src/live.ts

@@ -0,0 +1,182 @@
+import { z } from "zod";
+import {
+  AlertEventSchema,
+  ClassifierHitEventSchema,
+  EquityCandleSchema,
+  EquityPrintJoinSchema,
+  EquityPrintSchema,
+  FlowPacketSchema,
+  InferredDarkEventSchema,
+  OptionNBBOSchema,
+  OptionPrintSchema
+} from "./events";
+
+export const CursorSchema = z.object({
+  ts: z.number().int().nonnegative(),
+  seq: z.number().int().nonnegative()
+});
+
+export type Cursor = z.infer<typeof CursorSchema>;
+
+export const LiveGenericChannelSchema = z.enum([
+  "options",
+  "nbbo",
+  "equities",
+  "equity-joins",
+  "flow",
+  "classifier-hits",
+  "alerts",
+  "inferred-dark"
+]);
+
+export const LiveChannelSchema = z.enum([
+  "options",
+  "nbbo",
+  "equities",
+  "equity-joins",
+  "flow",
+  "classifier-hits",
+  "alerts",
+  "inferred-dark",
+  "equity-candles",
+  "equity-overlay"
+]);
+
+export type LiveChannel = z.infer<typeof LiveChannelSchema>;
+export type LiveGenericChannel = z.infer<typeof LiveGenericChannelSchema>;
+
+export const LiveSubscriptionSchema = z.discriminatedUnion("channel", [
+  z.object({
+    channel: LiveGenericChannelSchema
+  }),
+  z.object({
+    channel: z.literal("equity-candles"),
+    underlying_id: z.string().min(1),
+    interval_ms: z.number().int().positive()
+  }),
+  z.object({
+    channel: z.literal("equity-overlay"),
+    underlying_id: z.string().min(1)
+  })
+]);
+
+export type LiveSubscription = z.infer<typeof LiveSubscriptionSchema>;
+
+const livePayloadSchemas = {
+  options: OptionPrintSchema,
+  nbbo: OptionNBBOSchema,
+  equities: EquityPrintSchema,
+  "equity-joins": EquityPrintJoinSchema,
+  flow: FlowPacketSchema,
+  "classifier-hits": ClassifierHitEventSchema,
+  alerts: AlertEventSchema,
+  "inferred-dark": InferredDarkEventSchema,
+  "equity-candles": EquityCandleSchema,
+  "equity-overlay": EquityPrintSchema
+} as const;
+
+export const FeedSnapshotSchema = z.object({
+  subscription: LiveSubscriptionSchema,
+  items: z.array(z.unknown()),
+  watermark: CursorSchema.nullable(),
+  next_before: CursorSchema.nullable()
+});
+
+export type FeedSnapshot<T> = {
+  subscription: LiveSubscription;
+  items: T[];
+  watermark: Cursor | null;
+  next_before: Cursor | null;
+};
+
+export const LiveSubscribeMessageSchema = z.object({
+  op: z.literal("subscribe"),
+  subscriptions: z.array(LiveSubscriptionSchema).min(1)
+});
+
+export type LiveSubscribeMessage = z.infer<typeof LiveSubscribeMessageSchema>;
+
+export const LiveUnsubscribeMessageSchema = z.object({
+  op: z.literal("unsubscribe"),
+  subscriptions: z.array(LiveSubscriptionSchema).min(1)
+});
+
+export type LiveUnsubscribeMessage = z.infer<typeof LiveUnsubscribeMessageSchema>;
+
+export const LivePingMessageSchema = z.object({
+  op: z.literal("ping")
+});
+
+export type LivePingMessage = z.infer<typeof LivePingMessageSchema>;
+
+export const LiveClientMessageSchema = z.discriminatedUnion("op", [
+  LiveSubscribeMessageSchema,
+  LiveUnsubscribeMessageSchema,
+  LivePingMessageSchema
+]);
+
+export type LiveClientMessage = z.infer<typeof LiveClientMessageSchema>;
+
+export const LiveReadyMessageSchema = z.object({
+  op: z.literal("ready")
+});
+
+export type LiveReadyMessage = z.infer<typeof LiveReadyMessageSchema>;
+
+export const LiveSnapshotMessageSchema = z.object({
+  op: z.literal("snapshot"),
+  snapshot: FeedSnapshotSchema
+});
+
+export type LiveSnapshotMessage = z.infer<typeof LiveSnapshotMessageSchema>;
+
+export const LiveEventMessageSchema = z.object({
+  op: z.literal("event"),
+  subscription: LiveSubscriptionSchema,
+  item: z.unknown(),
+  watermark: CursorSchema.nullable()
+});
+
+export type LiveEventMessage = z.infer<typeof LiveEventMessageSchema>;
+
+export const LiveHeartbeatMessageSchema = z.object({
+  op: z.literal("heartbeat"),
+  ts: z.number().int().nonnegative()
+});
+
+export type LiveHeartbeatMessage = z.infer<typeof LiveHeartbeatMessageSchema>;
+
+export const LiveErrorMessageSchema = z.object({
+  op: z.literal("error"),
+  message: z.string().min(1)
+});
+
+export type LiveErrorMessage = z.infer<typeof LiveErrorMessageSchema>;
+
+export const LiveServerMessageSchema = z.discriminatedUnion("op", [
+  LiveReadyMessageSchema,
+  LiveSnapshotMessageSchema,
+  LiveEventMessageSchema,
+  LiveHeartbeatMessageSchema,
+  LiveErrorMessageSchema
+]);
+
+export type LiveServerMessage = z.infer<typeof LiveServerMessageSchema>;
+
+export const getSubscriptionKey = (subscription: LiveSubscription): string => {
+  switch (subscription.channel) {
+    case "equity-candles":
+      return `${subscription.channel}|${subscription.underlying_id}|${subscription.interval_ms}`;
+    case "equity-overlay":
+      return `${subscription.channel}|${subscription.underlying_id}`;
+    default:
+      return subscription.channel;
+  }
+};
+
+export const parseLivePayload = (
+  channel: LiveChannel,
+  item: unknown
+): z.infer<(typeof livePayloadSchemas)[typeof channel]> => {
+  return livePayloadSchemas[channel].parse(item);
+};

packages/types/tests/live.test.ts

@@ -0,0 +1,69 @@
+import { describe, expect, it } from "bun:test";
+import {
+  CursorSchema,
+  LiveClientMessageSchema,
+  LiveServerMessageSchema,
+  getSubscriptionKey
+} from "../src/live";
+
+describe("live protocol types", () => {
+  it("builds stable keys for generic and parameterized subscriptions", () => {
+    expect(getSubscriptionKey({ channel: "flow" })).toBe("flow");
+    expect(
+      getSubscriptionKey({
+        channel: "equity-candles",
+        underlying_id: "SPY",
+        interval_ms: 60000
+      })
+    ).toBe("equity-candles|SPY|60000");
+    expect(getSubscriptionKey({ channel: "equity-overlay", underlying_id: "SPY" })).toBe(
+      "equity-overlay|SPY"
+    );
+  });
+
+  it("validates subscribe messages", () => {
+    const parsed = LiveClientMessageSchema.parse({
+      op: "subscribe",
+      subscriptions: [
+        { channel: "flow" },
+        { channel: "equity-candles", underlying_id: "SPY", interval_ms: 60000 }
+      ]
+    });
+
+    expect(parsed.op).toBe("subscribe");
+    expect(parsed.subscriptions).toHaveLength(2);
+  });
+
+  it("validates snapshot and event server messages", () => {
+    const cursor = CursorSchema.parse({ ts: 100, seq: 2 });
+    const snapshot = LiveServerMessageSchema.parse({
+      op: "snapshot",
+      snapshot: {
+        subscription: { channel: "alerts" },
+        items: [],
+        watermark: cursor,
+        next_before: null
+      }
+    });
+    const event = LiveServerMessageSchema.parse({
+      op: "event",
+      subscription: { channel: "equity-overlay", underlying_id: "SPY" },
+      item: {
+        source_ts: 100,
+        ingest_ts: 101,
+        seq: 1,
+        trace_id: "eq-1",
+        ts: 100,
+        underlying_id: "SPY",
+        price: 500,
+        size: 10,
+        exchange: "X",
+        offExchangeFlag: true
+      },
+      watermark: cursor
+    });
+
+    expect(snapshot.op).toBe("snapshot");
+    expect(event.op).toBe("event");
+  });
+});