import type { WorkbenchCommand } from "../../internal/workbench/contracts.ts"; import { workbenchRuntime } from "../../internal/workbench/runtime.ts"; import { workbenchNativeServiceCommand } from "./workbench-native-service.ts"; export async function runWorkbenchCli(argv: string[], env: Record = process.env) { const parsed = parse(argv); if (parsed.help || parsed.positionals.length === 0) return help(); if (parsed.positionals[0] === "service") return workbenchNativeServiceCommand({ service: parsed.positionals[1], action: parsed.positionals[2] ?? "status", cwd: process.cwd(), env }); if (parsed.positionals[0] === "events" && parsed.positionals[1] === "inspect") return inspectEvents(parsed, env); const command = commandFrom(parsed, env); if (parsed.overApi) return overApi(command, parsed, env); const runtime = workbenchRuntime(env); try { const result = await runtime.dispatch(command); return { ...result, transport: "local", route: "application-dispatcher", identity: identity(result.data) }; } finally { await runtime.close(); } } function help() { return { ok: true, operation: "help", mode: "local-dispatcher-default", usage: [ "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] [--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" ], transportContract: "--over-api only changes transport; --overapi is unsupported", localConfig: ["WORKBENCH_MODE", "WORKBENCH_NATIVE_STATE_FILE"], nativeServiceConfig: ["WORKBENCH_PUBLIC_BASE_URL", "WORKBENCH_API_BIND_HOST", "WORKBENCH_API_PROBE_HOST", "WORKBENCH_API_PORT", "WORKBENCH_API_IDLE_TIMEOUT_SECONDS", "WORKBENCH_WORKER_BIND_HOST", "WORKBENCH_WORKER_PROBE_HOST", "WORKBENCH_WORKER_HEALTH_PORT", "WORKBENCH_WEB_BIND_HOST", "WORKBENCH_WEB_PROBE_HOST", "WORKBENCH_WEB_PORT", "WORKBENCH_NATIVE_API_URL", "WORKBENCH_WEB_RUNTIME_CONFIG"], apiConfig: ["WORKBENCH_API_URL", "HWLAB_API_KEY"] }; } async function inspectEvents(parsed: Parsed, env: Record) { if (!parsed.overApi) throw codedError("workbench_events_over_api_required", "workbench events inspect requires --over-api so it exercises the product SSE route"); const baseUrl = parsed.values["api-url"] || env.WORKBENCH_API_URL; if (!baseUrl) throw codedError("workbench_api_url_required", "WORKBENCH_API_URL or --api-url is required with --over-api"); const sessionId = parsed.values["session-id"] ?? ""; 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(); if (sessionId) params.set("sessionId", sessionId); if (traceId) params.set("traceId", traceId); const path = `/v1/workbench/events?${params.toString()}`; const controller = new AbortController(); const timer = setTimeout(() => controller.abort(), timeoutMs); const frames: SseFrame[] = []; let buffer = ""; try { const response = await fetch(`${baseUrl.replace(/\/$/u, "")}${path}`, { signal: controller.signal }); if (!response.ok || !response.body) throw codedError("workbench_sse_open_failed", `Workbench SSE returned HTTP ${response.status}`); const reader = response.body.getReader(); const decoder = new TextDecoder(); readStream: while (true) { const item = await reader.read(); if (item.done) break; buffer += decoder.decode(item.value, { stream: true }); let boundary = buffer.indexOf("\n\n"); while (boundary >= 0) { const block = buffer.slice(0, boundary); buffer = buffer.slice(boundary + 2); const frame = parseSseBlock(block); if (frame) frames.push(frame); if (framesInspectionSatisfied(frames, minEvents, waitFor)) { await reader.cancel().catch(() => undefined); break readStream; } boundary = buffer.indexOf("\n\n"); } } } catch (error: any) { const business = businessFrames(frames); if (error?.name !== "AbortError" && !(error?.code === "ECONNRESET" && framesInspectionSatisfied(frames, minEvents, waitFor))) { const observed = semanticSummary(business); throw codedError("workbench_sse_read_failed", `Workbench SSE read failed: ${error?.code ?? error?.name ?? "unknown"}; frames=${frames.length}; businessEvents=${business.length}; observed=${observed.observedSemantics.join(",")}; connected=${frames.some((entry) => entry.name === "workbench.connected")}`); } } finally { clearTimeout(timer); } const business = businessFrames(frames); const connectedFrame = frames.find((entry) => entry.name === "workbench.connected") ?? null; const connected = connectedFrame !== null; const connectedContract = connectedFrame ? workbenchConnectedContract(connectedFrame.data) : null; const semantics = semanticSummary(business); const missingSemantics = waitFor.filter((semantic) => !semantics.observedSemantics.includes(semantic)); const ok = connected && inspectionSatisfied(business, minEvents, waitFor); const allEventTypes = business.map(eventType); 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, connectedContract, eventCount: business.length, eventTypes: timingDetail === "full" ? allEventTypes : [...new Set(allEventTypes)], eventTypeCounts: countValues(allEventTypes), 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 { const [group, action] = parsed.positionals; if (group === "health") return { operation: "health" }; const actor = { id: required(parsed, "actor-id"), role: parsed.values["actor-role"] ?? "user" }; if (group === "session" && action === "create") return { operation: "session.create", actor, params: compact({ sessionId: parsed.values["session-id"], conversationId: parsed.values["conversation-id"], projectId: parsed.values["project-id"], providerProfile: parsed.values["provider-profile"] }) }; if (group === "turn" && action === "submit") return { operation: "turn.submit", actor, traceId: parsed.values["trace-id"], params: compact({ sessionId: required(parsed, "session-id"), message: required(parsed, "message"), projectId: parsed.values["project-id"], providerProfile: parsed.values["provider-profile"], shortConnection: true }) }; if (group === "turn" && action === "cancel") return { operation: "turn.cancel", actor, traceId: required(parsed, "trace-id"), params: compact({ sessionId: parsed.values["session-id"] }) }; throw codedError("unsupported_workbench_command", `unsupported Workbench command: ${group} ${action ?? ""}`.trim()); } async function overApi(command: WorkbenchCommand, parsed: Parsed, env: Record) { const baseUrl = parsed.values["api-url"] || env.WORKBENCH_API_URL; if (!baseUrl) throw codedError("workbench_api_url_required", "WORKBENCH_API_URL or --api-url is required with --over-api"); const headers: Record = { "content-type": "application/json" }; const apiKey = env.HWLAB_API_KEY; if (apiKey) headers.authorization = `Bearer ${apiKey}`; const route = "/v1/workbench/commands"; const response = await fetch(`${baseUrl.replace(/\/$/u, "")}${route}`, { method: "POST", headers, body: JSON.stringify(command) }); const body = await response.json().catch(() => null) as Record | null; if (!body) throw codedError("workbench_api_invalid_json", `Workbench API returned invalid JSON (${response.status})`); return { ...body, transport: "api", baseUrl, route: `POST ${route}`, httpStatus: response.status, identity: identity(body.data) }; } type Parsed = { positionals: string[]; values: Record; overApi: boolean; help: boolean }; function parse(argv: string[]): Parsed { const parsed: Parsed = { positionals: [], values: {}, overApi: false, help: false }; for (let index = 0; index < argv.length; index += 1) { const token = argv[index]; if (token === "--overapi") throw codedError("unsupported_option", "--overapi was retired; use --over-api"); if (token === "--over-api") parsed.overApi = true; else if (token === "--help" || token === "-h") parsed.help = true; else if (token.startsWith("--")) { const value = argv[index + 1]; if (!value || value.startsWith("--")) throw codedError("option_value_required", `${token} requires a value`); parsed.values[token.slice(2)] = value; index += 1; } else parsed.positionals.push(token); } return parsed; } function identity(data: any) { return { sessionId: data?.sessionId ?? data?.session?.sessionId ?? null, traceId: data?.traceId ?? null, workflowId: data?.workflowId ?? null, workflowRunId: data?.workflowRunId ?? null }; } function record(value: unknown): Record | null { return value && typeof value === "object" && !Array.isArray(value) ? value as Record : null; } function positiveInteger(value: unknown, fallback: number): number { const parsed = Number.parseInt(String(value ?? ""), 10); return Number.isSafeInteger(parsed) && parsed > 0 ? parsed : fallback; } const eventSemantics = ["user", "backend", "assistant", "terminal", "final"] as const; type EventSemantic = typeof eventSemantics[number]; function semanticWait(value: unknown): EventSemantic[] { const requested = String(value ?? "").split(",").map((item) => item.trim().toLowerCase()).filter(Boolean); const unsupported = requested.filter((item) => !eventSemantics.includes(item as 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[]; } type SseFrame = { name: string; data: Record; observedAt: string }; function workbenchConnectedContract(data: Record) { const refreshReplay = record(data.refreshReplay); const counts = record(refreshReplay?.counts); const query = record(refreshReplay?.query); const queryTiming = record(query?.timing); const queryIndex = record(query?.index); return { deliverySemantics: String(data.deliverySemantics ?? "").trim() || null, realtimeSource: String(data.realtimeSource ?? "").trim() || null, replay: data.replay === true, liveOnly: data.liveOnly === true, lossPossible: data.lossPossible === true, refreshReplay: refreshReplay ? { phase: String(refreshReplay.phase ?? "").trim() || null, topic: String(refreshReplay.topic ?? "").trim() || null, barrier: Array.isArray(refreshReplay.barrier) ? refreshReplay.barrier : [], counts: counts ? { matched: Number(counts.matched ?? 0), replayed: Number(counts.replayed ?? 0), buffered: Number(counts.buffered ?? 0), bufferedDelivered: Number(counts.bufferedDelivered ?? 0), liveDelivered: Number(counts.liveDelivered ?? 0), deduplicated: Number(counts.deduplicated ?? 0) } : null, query: query ? { completionReason: String(query.completionReason ?? "").trim() || null, complete: query.complete === true, scannedCount: nullableNumber(query.scannedCount), parsedCount: nullableNumber(query.parsedCount), matchedCount: nullableNumber(query.matchedCount), filterRejectedCount: nullableNumber(query.filterRejectedCount), keyRejectedCount: nullableNumber(query.keyRejectedCount), partitionKeyScoped: query.partitionKeyScoped === true, targetPartition: nullableNumber(query.targetPartition), index: queryIndex ? { hit: queryIndex.hit === true, source: String(queryIndex.source ?? "").trim() || null, fallbackReason: String(queryIndex.fallbackReason ?? "").trim() || null, indexedEventCount: nullableNumber(queryIndex.indexedEventCount), indexedSessionCount: nullableNumber(queryIndex.indexedSessionCount), incompleteSessionCount: nullableNumber(queryIndex.incompleteSessionCount), blocking: false, valuesPrinted: false } : null, timing: queryTiming ? { endOffsetSnapshotMs: nullableNumber(queryTiming.endOffsetSnapshotMs), groupOffsetsMs: nullableNumber(queryTiming.groupOffsetsMs), consumerConnectMs: nullableNumber(queryTiming.consumerConnectMs), consumerSubscribeMs: nullableNumber(queryTiming.consumerSubscribeMs), scanMs: nullableNumber(queryTiming.scanMs), cleanupMs: nullableNumber(queryTiming.cleanupMs), indexLookupMs: nullableNumber(queryTiming.indexLookupMs), totalMs: nullableNumber(queryTiming.totalMs), valuesPrinted: false } : null, valuesPrinted: false } : null } : null, valuesPrinted: false }; } function businessFrames(frames: SseFrame[]) { return frames.filter((entry) => entry.name === "hwlab.event.v1"); } function countValues(values: string[]) { const counts: Record = {}; for (const value of values) counts[value] = (counts[value] ?? 0) + 1; return counts; } function nullableNumber(value: unknown) { const parsed = Number(value); return Number.isFinite(parsed) && parsed >= 0 ? parsed : null; } function framesInspectionSatisfied(frames: SseFrame[], minEvents: number, waitFor: EventSemantic[]) { return frames.some((entry) => entry.name === "workbench.connected") && inspectionSatisfied(businessFrames(frames), minEvents, waitFor); } 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: SseFrame[]) { const semanticCounts = Object.fromEntries(eventSemantics.map((semantic) => [semantic, 0])) as Record; const terminalStatuses: string[] = []; for (const frame of business) { const event = record(frame.data?.event) ?? {}; for (const semantic of eventSemantics) if (matchesSemantic(event, semantic)) semanticCounts[semantic] += 1; if (matchesSemantic(event, "terminal")) { const status = String(event.terminalStatus ?? event.status ?? "").trim(); if (status && !terminalStatuses.includes(status)) terminalStatuses.push(status); } } return { observedSemantics: eventSemantics.filter((semantic) => semanticCounts[semantic] > 0), semanticCounts, terminalStatuses }; } function matchesSemantic(event: Record, semantic: EventSemantic) { const type = String(event.eventType ?? event.type ?? "").trim().toLowerCase(); if (semantic === "user") return type === "user" || type === "user_message"; if (semantic === "backend") return type === "backend" || type === "backend_status"; if (semantic === "assistant") return type === "assistant" || type === "assistant_message"; 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): 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) ?? {}, 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; } function codedError(code: string, message: string) { return Object.assign(new Error(message), { code }); }