root
281 lines
18 KiB
TypeScript
281 lines
18 KiB
TypeScript
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<string, string | undefined> = 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<string, string | undefined>) {
|
|
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 (inspectionSatisfied(businessFrames(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" && inspectionSatisfied(business, 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 connected = frames.some((entry) => entry.name === "workbench.connected");
|
|
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, 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 {
|
|
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<string, string | undefined>) {
|
|
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<string, string> = { "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<string, any> | 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<string, string>; 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<string, any> | null { return value && typeof value === "object" && !Array.isArray(value) ? value as Record<string, any> : 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<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: SseFrame[]) {
|
|
const semanticCounts = Object.fromEntries(eventSemantics.map((semantic) => [semantic, 0])) as Record<EventSemantic, number>;
|
|
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<string, any>, 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<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): 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<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; }
|
|
function codedError(code: string, message: string) { return Object.assign(new Error(message), { code }); }
|