test: add Workbench session rail L0 benchmark

2026-07-20 13:05:52 +02:00
parent e810679d9c
commit 92d9053bea
1 changed files with 228 additions and 0 deletions
@@ -0,0 +1,228 @@
+// SPEC: PJ2026-01060505 Workbench performance; PJ2026-010401080313 Workbench realtime authority.
+// Responsibility: Deterministic L0 timing for session tag switching, Kafka replay, and session rail refresh.
+
+import type { ChatMessage, TraceEvent, WorkbenchSessionRecord } from "../src/types/index.ts";
+import { projectWorkbenchLiveKafkaMessage, projectWorkbenchLiveKafkaUserMessage, projectWorkbenchLiveKafkaUserSession } from "../src/stores/workbench-live-kafka-event.ts";
+import { createWorkbenchServerState, reduceWorkbenchServerState, selectActiveMessages, selectActiveSession, selectSessionList, type WorkbenchServerState } from "../src/stores/workbench-server-state.ts";
+import { mergeSessionIntoList, sessionToSessionTab, sortSessionTabs } from "../src/stores/workbench-session.ts";
+
+interface SampleSummary {
+  p50: number;
+  p95: number;
+  max: number;
+}
+
+interface ReplayFixture {
+  sessionId: string;
+  traceId: string;
+  events: TraceEvent[];
+}
+
+const rounds = positiveInteger(process.env.HWLAB_L0_PERF_ROUNDS, 30);
+const warmupRounds = positiveInteger(process.env.HWLAB_L0_PERF_WARMUP, 8);
+const sessionScales = [40, 100, 500];
+const messagesPerSession = positiveInteger(process.env.HWLAB_L0_PERF_MESSAGES, 40);
+const replayEventCounts = [100, 500, 1_000];
+
+const output: Record<string, unknown> = {
+  contractVersion: "workbench-l0-session-performance-v1",
+  runtime: `bun-${Bun.version}`,
+  rounds,
+  warmupRounds,
+  messagesPerSession,
+  units: "milliseconds",
+  sessionScales: {},
+  replayScales: {}
+};
+
+for (const sessionCount of sessionScales) {
+  const sessions = sessionFixtures(sessionCount, messagesPerSession);
+  const state = reduceWorkbenchServerState(createWorkbenchServerState(), { type: "session.list", sessions });
+  const activeIds = sessions.map((session) => session.sessionId);
+  let switchIndex = 0;
+  const activeSwitch = measure(() => {
+    const activeSessionId = activeIds[switchIndex % activeIds.length] ?? null;
+    switchIndex += 1;
+    const activeSession = selectActiveSession(state, activeSessionId);
+    const messages = selectActiveMessages(state, activeSessionId);
+    const tabs = sortSessionTabs(selectSessionList(state), activeSessionId, state.sessionStatusById);
+    consume(activeSession?.sessionId, messages.length, tabs[0]?.key);
+  });
+  const projection = measure(() => {
+    const tabs = sessions.map((session) => sessionToSessionTab(session, sessions[0]?.sessionId ?? null));
+    consume(tabs[0]?.label, tabs.at(-1)?.updatedAt);
+  });
+  const sort = measure(() => {
+    const tabs = sortSessionTabs(sessions, sessions[0]?.sessionId ?? null);
+    consume(tabs[0]?.key, tabs.at(-1)?.key);
+  });
+  const railRefresh = measure(() => {
+    const refreshed = refreshRailWithUserEvent(state, sessionCount - 1);
+    consume(refreshed.state.sessionOrder.length, refreshed.tabs[0]?.label);
+  });
+  (output.sessionScales as Record<string, unknown>)[String(sessionCount)] = {
+    projectTabs: projection,
+    projectAndSortTabs: sort,
+    activeTagSwitch: activeSwitch,
+    kafkaUserEventAndRailRefresh: railRefresh
+  };
+}
+
+for (const eventCount of replayEventCounts) {
+  const fixture = replayFixture(eventCount);
+  const result = measure(() => {
+    const replayed = replayKafkaEvents(fixture);
+    consume(replayed.state.sessionOrder.length, replayed.message?.runnerTrace?.events?.length, replayed.tabs[0]?.status);
+  }, Math.min(rounds, eventCount >= 1_000 ? 12 : rounds), Math.min(warmupRounds, 4));
+  (output.replayScales as Record<string, unknown>)[String(eventCount)] = result;
+}
+
+console.log(JSON.stringify(output, null, 2));
+
+function refreshRailWithUserEvent(initial: WorkbenchServerState, targetIndex: number): { state: WorkbenchServerState; tabs: ReturnType<typeof sortSessionTabs> } {
+  const sessionId = initial.sessionOrder[targetIndex] ?? initial.sessionOrder[0] ?? "ses_perf_0";
+  const traceId = `trc_refresh_${targetIndex}`;
+  const event: TraceEvent = {
+    type: "user",
+    eventType: "user",
+    messageId: `msg_refresh_${targetIndex}`,
+    userMessageId: `msg_refresh_${targetIndex}`,
+    text: "refresh rail title",
+    createdAt: "2026-07-20T12:00:00.000Z",
+    sessionId,
+    projectedSeq: 1
+  };
+  const previousMessages = initial.messagesBySessionId[sessionId] ?? [];
+  const userMessage = projectWorkbenchLiveKafkaUserMessage({ previous: null, traceId, sessionId, event, receivedAt: "2026-07-20T12:00:00.010Z" });
+  if (!userMessage) throw new Error("user replay fixture did not project");
+  let state = reduceWorkbenchServerState(initial, { type: "message.upsert", sessionId, message: userMessage });
+  const sessions = selectSessionList(state);
+  const existing = sessions.find((session) => session.sessionId === sessionId) ?? null;
+  const projectedSession = projectWorkbenchLiveKafkaUserSession({ previous: existing, message: userMessage, sessionId });
+  if (!projectedSession) throw new Error("user session fixture did not project");
+  state = reduceWorkbenchServerState(state, { type: "session.list", sessions: mergeSessionIntoList(sessions, { ...projectedSession, messages: [...previousMessages, userMessage] }) });
+  return { state, tabs: sortSessionTabs(selectSessionList(state), sessionId, state.sessionStatusById) };
+}
+
+function replayKafkaEvents(fixture: ReplayFixture): { state: WorkbenchServerState; message: ChatMessage | null; tabs: ReturnType<typeof sortSessionTabs> } {
+  let state = createWorkbenchServerState();
+  state = reduceWorkbenchServerState(state, { type: "session.detail", session: { sessionId: fixture.sessionId, messages: [] } });
+  let agentMessage: ChatMessage | null = null;
+  for (const event of fixture.events) {
+    if (event.type === "user") {
+      const userMessage = projectWorkbenchLiveKafkaUserMessage({ previous: null, traceId: fixture.traceId, sessionId: fixture.sessionId, event, receivedAt: String(event.createdAt) });
+      if (!userMessage) throw new Error("replay user fixture did not project");
+      state = reduceWorkbenchServerState(state, { type: "message.upsert", sessionId: fixture.sessionId, message: userMessage });
+      const session = projectWorkbenchLiveKafkaUserSession({ previous: selectActiveSession(state, fixture.sessionId), message: userMessage, sessionId: fixture.sessionId });
+      if (session) state = reduceWorkbenchServerState(state, { type: "session.list", sessions: mergeSessionIntoList(selectSessionList(state), session) });
+      continue;
+    }
+    agentMessage = projectWorkbenchLiveKafkaMessage({ previous: agentMessage, traceId: fixture.traceId, sessionId: fixture.sessionId, event, receivedAt: String(event.createdAt), turnStartedAt: String(fixture.events[0]?.createdAt) });
+    state = reduceWorkbenchServerState(state, { type: "message.upsert", sessionId: fixture.sessionId, message: agentMessage });
+  }
+  return { state, message: agentMessage, tabs: sortSessionTabs(selectSessionList(state), fixture.sessionId, state.sessionStatusById) };
+}
+
+function replayFixture(eventCount: number): ReplayFixture {
+  const sessionId = `ses_replay_${eventCount}`;
+  const traceId = `trc_replay_${eventCount}`;
+  const events: TraceEvent[] = [{
+    type: "user",
+    eventType: "user",
+    messageId: `msg_replay_${eventCount}_user`,
+    userMessageId: `msg_replay_${eventCount}_user`,
+    text: "measure retention replay",
+    createdAt: timestamp(0),
+    sessionId,
+    projectedSeq: 1
+  }];
+  for (let index = 1; index < eventCount; index += 1) {
+    const terminal = index === eventCount - 1;
+    events.push({
+      type: terminal ? "result" : index % 5 === 0 ? "assistant_progress" : "event",
+      eventType: terminal ? "terminal" : index % 5 === 0 ? "assistant_progress" : "backend",
+      text: index % 5 === 0 ? `progress ${index}` : undefined,
+      finalResponse: terminal ? { text: "completed replay" } : undefined,
+      status: terminal ? "completed" : "running",
+      terminal,
+      createdAt: timestamp(index),
+      projectedSeq: index + 1,
+      sourceSeq: index + 1,
+      source: "agentrun.event.v1",
+      sessionId,
+      traceId
+    });
+  }
+  return { sessionId, traceId, events };
+}
+
+function sessionFixtures(sessionCount: number, messageCount: number): WorkbenchSessionRecord[] {
+  return Array.from({ length: sessionCount }, (_, sessionIndex) => {
+    const sessionId = `ses_perf_${sessionIndex}`;
+    const messages = Array.from({ length: messageCount }, (_, messageIndex) => messageFixture(sessionId, sessionIndex, messageIndex));
+    return {
+      sessionId,
+      threadId: `thr_perf_${sessionIndex}`,
+      status: sessionIndex % 11 === 0 ? "running" : "completed",
+      lastTraceId: `trc_perf_${sessionIndex}_${messageCount - 1}`,
+      messageCount,
+      messages,
+      firstUserMessagePreview: `session ${sessionIndex} title`,
+      lastUserMessageAt: timestamp(sessionIndex * 100 + messageCount - 2),
+      updatedAt: timestamp(sessionIndex * 100 + messageCount - 1)
+    };
+  });
+}
+
+function messageFixture(sessionId: string, sessionIndex: number, messageIndex: number): ChatMessage {
+  const role = messageIndex % 2 === 0 ? "user" : "agent";
+  const traceId = `trc_perf_${sessionIndex}_${Math.floor(messageIndex / 2)}`;
+  return {
+    id: `msg_perf_${sessionIndex}_${messageIndex}`,
+    messageId: `msg_perf_${sessionIndex}_${messageIndex}`,
+    role,
+    title: role === "user" ? "用户" : "Code Agent",
+    text: role === "user" ? `session ${sessionIndex} prompt ${messageIndex}` : `response ${messageIndex}`,
+    status: role === "user" ? "sent" : messageIndex === messagesPerSession - 1 && sessionIndex % 11 === 0 ? "running" : "completed",
+    traceId,
+    sessionId,
+    createdAt: timestamp(sessionIndex * 100 + messageIndex)
+  };
+}
+
+function measure(operation: () => void, sampleRounds = rounds, warmups = warmupRounds): SampleSummary {
+  for (let index = 0; index < warmups; index += 1) operation();
+  const samples: number[] = [];
+  for (let index = 0; index < sampleRounds; index += 1) {
+    const startedAt = Bun.nanoseconds();
+    operation();
+    samples.push((Bun.nanoseconds() - startedAt) / 1_000_000);
+  }
+  samples.sort((left, right) => left - right);
+  return {
+    p50: rounded(percentile(samples, 0.5)),
+    p95: rounded(percentile(samples, 0.95)),
+    max: rounded(samples.at(-1) ?? 0)
+  };
+}
+
+function percentile(samples: number[], ratio: number): number {
+  return samples[Math.min(samples.length - 1, Math.max(0, Math.ceil(samples.length * ratio) - 1))] ?? 0;
+}
+
+function positiveInteger(value: string | undefined, fallback: number): number {
+  const parsed = Number(value);
+  return Number.isInteger(parsed) && parsed > 0 ? parsed : fallback;
+}
+
+function timestamp(offsetSeconds: number): string {
+  return new Date(Date.UTC(2026, 6, 20, 0, 0, offsetSeconds)).toISOString();
+}
+
+function rounded(value: number): number {
+  return Math.round(value * 1_000) / 1_000;
+}
+
+function consume(...values: unknown[]): void {
+  if (values.length === Number.MIN_SAFE_INTEGER) console.log(values);
+}