root
229 lines
11 KiB
TypeScript
229 lines
11 KiB
TypeScript
// 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);
|
|
}
|