diff --git a/internal/workbench/workbench.test.ts b/internal/workbench/workbench.test.ts index 7d800c98..51f4eec6 100644 --- a/internal/workbench/workbench.test.ts +++ b/internal/workbench/workbench.test.ts @@ -165,10 +165,10 @@ describe("Workbench native HTTP adapter", () => { ], {}); for (let index = 0; index < 20 && subscribers.size === 0; index += 1) await Bun.sleep(5); const events = [ - { type: "user", eventType: "user" }, - { type: "backend", eventType: "backend" }, - { type: "assistant", eventType: "assistant", final: true, replyAuthority: true, finalResponse: { text: "done" } }, - { type: "result", eventType: "terminal", terminal: true, terminalStatus: "completed", status: "completed" } + { type: "user", eventType: "user", createdAt: "2026-07-19T16:00:00.000Z" }, + { type: "backend", eventType: "backend", createdAt: "2026-07-19T16:00:02.000Z" }, + { type: "assistant", eventType: "assistant", createdAt: "2026-07-19T16:01:02.000Z", final: true, replyAuthority: true, finalResponse: { text: "done" } }, + { type: "result", eventType: "terminal", createdAt: "2026-07-19T16:01:03.000Z", terminal: true, terminalStatus: "completed", status: "completed" } ]; for (const event of events) { for (const listener of subscribers) listener({ schema: "hwlab.event.v1", sessionId: "ses_l0_semantics", traceId: "trc_l0_semantics", event }); @@ -181,7 +181,27 @@ describe("Workbench native HTTP adapter", () => { observedSemantics: ["user", "backend", "assistant", "terminal", "final"], missingSemantics: [], semanticCounts: { user: 1, backend: 1, assistant: 1, terminal: 1, final: 1 }, - terminalStatuses: ["completed"] + terminalStatuses: ["completed"], + timing: { + clock: "agentrun-source-event-created-at", + phasesMs: { + firstEventToBackend: 2000, + admissionAndDispatch: null, + runnerProvisioning: null, + runtimePreparation: null, + providerExecution: null, + backendToAssistant: 60000, + assistantToTerminal: 1000, + firstEventToTerminal: 63000, + firstEventToFinal: 62000 + }, + slowestGap: { + durationMs: 60000, + toIndex: 2, + toEventType: "assistant" + }, + valuesPrinted: false + } }); } finally { server.stop(true); diff --git a/tools/src/workbench-cli.ts b/tools/src/workbench-cli.ts index 1c3df883..a2a42ace 100644 --- a/tools/src/workbench-cli.ts +++ b/tools/src/workbench-cli.ts @@ -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 0 ? 0 : 1; const params = new URLSearchParams(); @@ -54,7 +55,7 @@ async function inspectEvents(parsed: Parsed, env: Record controller.abort(), timeoutMs); - const frames: Array<{ name: string; data: Record }> = []; + 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 !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): 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 }>) { return frames.filter((entry) => entry.name === "hwlab.event.v1"); } -function inspectionSatisfied(business: Array<{ name: string; data: Record }>, minEvents: number, waitFor: EventSemantic[]) { +type SseFrame = { name: string; data: Record; 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 }>) { +function semanticSummary(business: SseFrame[]) { const semanticCounts = Object.fromEntries(eventSemantics.map((semantic) => [semantic, 0])) as Record; const terminalStatuses: string[] = []; for (const frame of business) { @@ -173,15 +175,105 @@ function matchesSemantic(event: Record, 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(); + 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 }) { return String(frame.data?.event?.eventType ?? frame.data?.event?.type ?? frame.data?.eventType ?? "unknown"); } -function parseSseBlock(block: string): { name: string; data: Record } | 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) { 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; }