feat: add Workbench L0 phase timing
This commit is contained in:
root
+100
-8
@@ -27,7 +27,7 @@ function help() {
|
||||
"hwlab-cli workbench health [--over-api]",
|
||||
"hwlab-cli workbench session create --actor-id ID [--provider-profile PROFILE] [--over-api]",
|
||||
"hwlab-cli workbench turn submit --actor-id ID --session-id ID --message TEXT [--trace-id ID] [--over-api]",
|
||||
"hwlab-cli workbench events inspect --session-id ID [--trace-id ID] --over-api [--timeout-ms MS] [--wait-for user,backend,assistant,terminal,final] [--min-events N]",
|
||||
"hwlab-cli workbench events inspect --session-id ID [--trace-id ID] --over-api [--timeout-ms MS] [--wait-for user,backend,assistant,terminal,final] [--min-events N] [--timing-detail summary|full]",
|
||||
"hwlab-cli workbench turn cancel --actor-id ID --trace-id ID [--over-api]",
|
||||
"hwlab-cli workbench service api|worker|web start|stop|restart|status|logs"
|
||||
],
|
||||
@@ -46,6 +46,7 @@ async function inspectEvents(parsed: Parsed, env: Record<string, string | undefi
|
||||
const traceId = parsed.values["trace-id"] ?? "";
|
||||
if (!sessionId && !traceId) throw codedError("workbench_realtime_scope_required", "--session-id or --trace-id is required");
|
||||
const timeoutMs = positiveInteger(parsed.values["timeout-ms"], 5000);
|
||||
const timingDetail = timingDetailValue(parsed.values["timing-detail"]);
|
||||
const waitFor = semanticWait(parsed.values["wait-for"]);
|
||||
const minEvents = parsed.values["min-events"] ? positiveInteger(parsed.values["min-events"], 1) : waitFor.length > 0 ? 0 : 1;
|
||||
const params = new URLSearchParams();
|
||||
@@ -54,7 +55,7 @@ async function inspectEvents(parsed: Parsed, env: Record<string, string | undefi
|
||||
const path = `/v1/workbench/events?${params.toString()}`;
|
||||
const controller = new AbortController();
|
||||
const timer = setTimeout(() => controller.abort(), timeoutMs);
|
||||
const frames: Array<{ name: string; data: Record<string, any> }> = [];
|
||||
const frames: SseFrame[] = [];
|
||||
let buffer = "";
|
||||
try {
|
||||
const response = await fetch(`${baseUrl.replace(/\/$/u, "")}${path}`, { signal: controller.signal });
|
||||
@@ -92,7 +93,7 @@ async function inspectEvents(parsed: Parsed, env: Record<string, string | undefi
|
||||
const semantics = semanticSummary(business);
|
||||
const missingSemantics = waitFor.filter((semantic) => !semantics.observedSemantics.includes(semantic));
|
||||
const ok = connected && inspectionSatisfied(business, minEvents, waitFor);
|
||||
return { ok, operation: "events.inspect", status: ok ? "passed" : "event-timeout", transport: "api-sse", baseUrl, route: `GET ${path}`, identity: { sessionId: sessionId || null, traceId: traceId || null }, connected, eventCount: business.length, eventTypes: business.map(eventType), timeoutMs, minEvents, waitFor, ...semantics, missingSemantics, error: ok ? undefined : { code: "workbench_kafka_sse_event_missing", message: `Workbench SSE did not deliver the required Kafka event conditions; missing=${missingSemantics.join(",") || (business.length < minEvents ? `min-events:${minEvents}` : "connection")}` }, valuesPrinted: false };
|
||||
return { ok, operation: "events.inspect", status: ok ? "passed" : "event-timeout", transport: "api-sse", baseUrl, route: `GET ${path}`, identity: { sessionId: sessionId || null, traceId: traceId || null }, connected, eventCount: business.length, eventTypes: business.map(eventType), timeoutMs, minEvents, waitFor, ...semantics, timing: timingSummary(business, timingDetail), missingSemantics, error: ok ? undefined : { code: "workbench_kafka_sse_event_missing", message: `Workbench SSE did not deliver the required Kafka event conditions; missing=${missingSemantics.join(",") || (business.length < minEvents ? `min-events:${minEvents}` : "connection")}` }, valuesPrinted: false };
|
||||
}
|
||||
|
||||
function commandFrom(parsed: Parsed, env: Record<string, string | undefined>): WorkbenchCommand {
|
||||
@@ -146,13 +147,14 @@ function semanticWait(value: unknown): EventSemantic[] {
|
||||
if (unsupported.length > 0) throw codedError("workbench_event_semantic_unsupported", `unsupported --wait-for semantics: ${unsupported.join(",")}; supported=${eventSemantics.join(",")}`);
|
||||
return [...new Set(requested)] as EventSemantic[];
|
||||
}
|
||||
function businessFrames(frames: Array<{ name: string; data: Record<string, any> }>) { return frames.filter((entry) => entry.name === "hwlab.event.v1"); }
|
||||
function inspectionSatisfied(business: Array<{ name: string; data: Record<string, any> }>, minEvents: number, waitFor: EventSemantic[]) {
|
||||
type SseFrame = { name: string; data: Record<string, any>; observedAt: string };
|
||||
function businessFrames(frames: SseFrame[]) { return frames.filter((entry) => entry.name === "hwlab.event.v1"); }
|
||||
function inspectionSatisfied(business: SseFrame[], minEvents: number, waitFor: EventSemantic[]) {
|
||||
if (business.length < minEvents) return false;
|
||||
const observed = semanticSummary(business).observedSemantics;
|
||||
return waitFor.every((semantic) => observed.includes(semantic));
|
||||
}
|
||||
function semanticSummary(business: Array<{ name: string; data: Record<string, any> }>) {
|
||||
function semanticSummary(business: SseFrame[]) {
|
||||
const semanticCounts = Object.fromEntries(eventSemantics.map((semantic) => [semantic, 0])) as Record<EventSemantic, number>;
|
||||
const terminalStatuses: string[] = [];
|
||||
for (const frame of business) {
|
||||
@@ -173,15 +175,105 @@ function matchesSemantic(event: Record<string, any>, semantic: EventSemantic) {
|
||||
if (semantic === "terminal") return event.terminal === true || type === "terminal" || type === "terminal_status" || type === "result" || event.agentRunEventType === "terminal_status";
|
||||
return event.final === true || event.replyAuthority === true || Boolean(record(event.finalResponse)?.text);
|
||||
}
|
||||
function timingSummary(business: SseFrame[], detail: "summary" | "full") {
|
||||
const rawTimeline = business.map((frame, index) => {
|
||||
const event = record(frame.data?.event) ?? {};
|
||||
const sourceAt = timestamp(event.createdAt);
|
||||
const producedAt = timestamp(frame.data?.producedAt);
|
||||
const observedAt = timestamp(frame.observedAt);
|
||||
const semantics = eventSemantics.filter((semantic) => matchesSemantic(event, semantic));
|
||||
return {
|
||||
index,
|
||||
eventType: eventType(frame),
|
||||
label: String(event.label ?? "").trim() || null,
|
||||
semantics,
|
||||
sourceAt,
|
||||
producedAt,
|
||||
observedAt,
|
||||
projectionLagMs: elapsedMs(sourceAt, producedAt),
|
||||
deliveryLagMs: elapsedMs(producedAt, observedAt)
|
||||
};
|
||||
});
|
||||
const seen = new Set<string>();
|
||||
const timeline = rawTimeline.filter((entry) => {
|
||||
const key = [entry.eventType, entry.label, entry.sourceAt].join("\u0000");
|
||||
if (seen.has(key)) return false;
|
||||
seen.add(key);
|
||||
return true;
|
||||
}).sort((left, right) => Date.parse(left.sourceAt ?? left.observedAt ?? "") - Date.parse(right.sourceAt ?? right.observedAt ?? ""))
|
||||
.map((entry, index) => ({ ...entry, index }));
|
||||
const sourceTimeline = timeline.filter((entry) => entry.sourceAt !== null);
|
||||
const firstSourceAt = sourceTimeline[0]?.sourceAt ?? null;
|
||||
const withDeltas = timeline.map((entry, index) => ({
|
||||
...entry,
|
||||
sinceFirstMs: elapsedMs(firstSourceAt, entry.sourceAt),
|
||||
sincePreviousMs: elapsedMs(timeline[index - 1]?.sourceAt ?? null, entry.sourceAt)
|
||||
}));
|
||||
const firstBackendAt = firstSemanticAt(withDeltas, "backend");
|
||||
const firstAssistantAt = firstSemanticAt(withDeltas, "assistant");
|
||||
const terminalAt = firstSemanticAt(withDeltas, "terminal");
|
||||
const finalAt = firstSemanticAt(withDeltas, "final");
|
||||
const runnerDispatchedAt = firstLabelAt(withDeltas, "agentrun:backend:runner-dispatch-completed");
|
||||
const runnerReadyAt = firstLabelAt(withDeltas, "agentrun:backend:run-claimed");
|
||||
const providerStartedAt = firstLabelAt(withDeltas, "agentrun:backend:turn/started");
|
||||
const slowestGap = withDeltas
|
||||
.filter((entry) => entry.sincePreviousMs !== null)
|
||||
.sort((left, right) => (right.sincePreviousMs ?? 0) - (left.sincePreviousMs ?? 0))[0] ?? null;
|
||||
return {
|
||||
clock: "agentrun-source-event-created-at",
|
||||
rawEventCount: rawTimeline.length,
|
||||
analyzedEventCount: timeline.length,
|
||||
duplicateEventCount: rawTimeline.length - timeline.length,
|
||||
milestones: { firstEventAt: firstSourceAt, firstBackendAt, firstAssistantAt, terminalAt, finalAt },
|
||||
phasesMs: {
|
||||
firstEventToBackend: elapsedMs(firstSourceAt, firstBackendAt),
|
||||
admissionAndDispatch: elapsedMs(firstSourceAt, runnerDispatchedAt),
|
||||
runnerProvisioning: elapsedMs(runnerDispatchedAt, runnerReadyAt),
|
||||
runtimePreparation: elapsedMs(runnerReadyAt, providerStartedAt),
|
||||
providerExecution: elapsedMs(providerStartedAt, firstAssistantAt),
|
||||
backendToAssistant: elapsedMs(firstBackendAt, firstAssistantAt),
|
||||
assistantToTerminal: elapsedMs(firstAssistantAt, terminalAt),
|
||||
firstEventToTerminal: elapsedMs(firstSourceAt, terminalAt),
|
||||
firstEventToFinal: elapsedMs(firstSourceAt, finalAt)
|
||||
},
|
||||
slowestGap: slowestGap ? {
|
||||
durationMs: slowestGap.sincePreviousMs,
|
||||
toIndex: slowestGap.index,
|
||||
toEventType: slowestGap.eventType,
|
||||
toLabel: slowestGap.label
|
||||
} : null,
|
||||
...(detail === "full" ? { timeline: withDeltas } : {}),
|
||||
valuesPrinted: false
|
||||
};
|
||||
}
|
||||
function timingDetailValue(value: unknown): "summary" | "full" {
|
||||
const detail = String(value ?? "summary").trim().toLowerCase();
|
||||
if (detail === "summary" || detail === "full") return detail;
|
||||
throw codedError("workbench_timing_detail_unsupported", "--timing-detail must be summary or full");
|
||||
}
|
||||
function firstSemanticAt(timeline: Array<{ semantics: EventSemantic[]; sourceAt: string | null }>, semantic: EventSemantic) {
|
||||
return timeline.find((entry) => entry.semantics.includes(semantic) && entry.sourceAt)?.sourceAt ?? null;
|
||||
}
|
||||
function firstLabelAt(timeline: Array<{ label: string | null; sourceAt: string | null }>, label: string) {
|
||||
return timeline.find((entry) => entry.label === label && entry.sourceAt)?.sourceAt ?? null;
|
||||
}
|
||||
function timestamp(value: unknown): string | null {
|
||||
const text = String(value ?? "").trim();
|
||||
return text && Number.isFinite(Date.parse(text)) ? new Date(Date.parse(text)).toISOString() : null;
|
||||
}
|
||||
function elapsedMs(start: string | null, end: string | null): number | null {
|
||||
if (!start || !end) return null;
|
||||
return Math.max(0, Date.parse(end) - Date.parse(start));
|
||||
}
|
||||
function eventType(frame: { data: Record<string, any> }) { return String(frame.data?.event?.eventType ?? frame.data?.event?.type ?? frame.data?.eventType ?? "unknown"); }
|
||||
function parseSseBlock(block: string): { name: string; data: Record<string, any> } | null {
|
||||
function parseSseBlock(block: string): SseFrame | null {
|
||||
if (!block || block.startsWith(":")) return null;
|
||||
const lines = block.split(/\r?\n/u);
|
||||
const name = lines.find((line) => line.startsWith("event:"))?.slice(6).trim() || "message";
|
||||
const dataText = lines.filter((line) => line.startsWith("data:" )).map((line) => line.slice(5).trimStart()).join("\n");
|
||||
if (!dataText) return null;
|
||||
const data = JSON.parse(dataText);
|
||||
return { name, data: record(data) ?? {} };
|
||||
return { name, data: record(data) ?? {}, observedAt: new Date().toISOString() };
|
||||
}
|
||||
function compact(value: Record<string, unknown>) { return Object.fromEntries(Object.entries(value).filter(([, item]) => item !== undefined)); }
|
||||
function required(parsed: Parsed, key: string) { const value = parsed.values[key]; if (!value) throw codedError("option_required", `--${key} is required`); return value; }
|
||||
|
||||
Reference in New Issue
Block a user