mirrors/pikasTech-HWLAB
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: wait for Workbench event semantics

Browse Source 
Co-Authored-By: Codex <noreply@openai.com>
This commit is contained in:
root
2026-07-18 09:54:06 +02:00
parent faef792fab
commit 28c6a21d27
3 changed files with 166 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/src/workbench-cli.ts
+50 -10
View File
@@ -27,13 +27,13 @@ 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] [--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]",
"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_API_BIND_HOST", "WORKBENCH_API_PROBE_HOST", "WORKBENCH_API_PUBLIC_HOST", "WORKBENCH_API_PORT", "WORKBENCH_WORKER_BIND_HOST", "WORKBENCH_WORKER_PROBE_HOST", "WORKBENCH_WORKER_HEALTH_PORT", "WORKBENCH_WEB_BIND_HOST", "WORKBENCH_WEB_PROBE_HOST", "WORKBENCH_WEB_PUBLIC_HOST", "WORKBENCH_WEB_PORT", "WORKBENCH_NATIVE_API_URL", "WORKBENCH_WEB_RUNTIME_CONFIG"],
nativeServiceConfig: ["WORKBENCH_API_BIND_HOST", "WORKBENCH_API_PROBE_HOST", "WORKBENCH_API_PUBLIC_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_PUBLIC_HOST", "WORKBENCH_WEB_PORT", "WORKBENCH_NATIVE_API_URL", "WORKBENCH_WEB_RUNTIME_CONFIG"],
apiConfig: ["WORKBENCH_API_URL", "HWLAB_API_KEY"]
};
}
@@ -46,7 +46,8 @@ 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 minEvents = positiveInteger(parsed.values["min-events"], 1);
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);
@@ -70,7 +71,7 @@ async function inspectEvents(parsed: Parsed, env: Record<string, string | undefi
buffer = buffer.slice(boundary + 2);
const frame = parseSseBlock(block);
if (frame) frames.push(frame);
if (frames.filter((entry) => entry.name === "hwlab.event.v1").length >= minEvents) {
if (inspectionSatisfied(businessFrames(frames), minEvents, waitFor)) {
await reader.cancel().catch(() => undefined);
break readStream;
}
@@ -78,17 +79,20 @@ async function inspectEvents(parsed: Parsed, env: Record<string, string | undefi
}
}
} catch (error: any) {
const businessCount = frames.filter((entry) => entry.name === "hwlab.event.v1").length;
if (error?.name !== "AbortError" && !(error?.code === "ECONNRESET" && businessCount >= minEvents)) {
throw codedError("workbench_sse_read_failed", `Workbench SSE read failed: ${error?.code ?? error?.name ?? "unknown"}; frames=${frames.length}; businessEvents=${businessCount}; connected=${frames.some((entry) => entry.name === "workbench.connected")}`);
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 = frames.filter((entry) => entry.name === "hwlab.event.v1");
const business = businessFrames(frames);
const connected = frames.some((entry) => entry.name === "workbench.connected");
const ok = connected && business.length >= minEvents;
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((entry) => String(entry.data?.event?.eventType ?? entry.data?.event?.type ?? entry.data?.eventType ?? "unknown")), timeoutMs, minEvents, error: ok ? undefined : { code: "workbench_kafka_sse_event_missing", message: "Workbench SSE did not deliver the required Kafka business events within the bounded window" }, valuesPrinted: false };
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 };
}
function commandFrom(parsed: Parsed, env: Record<string, string | undefined>): WorkbenchCommand {
@@ -134,6 +138,42 @@ function parse(argv: string[]): 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[];
}
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[]) {
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> }>) {
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 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 {
if (!block || block.startsWith(":")) return null;
const lines = block.split(/\r?\n/u);