From 36d54554325cf02e528db1dbdfcc89a033245c23 Mon Sep 17 00:00:00 2001 From: root Date: Thu, 9 Jul 2026 20:36:21 +0200 Subject: [PATCH] feat: stream HWLAB Kafka events to debug SSE --- internal/cloud/kafka-event-bridge.ts | 46 ++++ internal/cloud/server.ts | 8 +- .../cloud/workbench-kafka-sse-debug.test.ts | 94 +++++++ internal/cloud/workbench-kafka-sse-debug.ts | 176 +++++++++++++ web/hwlab-cloud-web/src/api/index.ts | 2 +- .../src/api/workbench-debug.ts | 83 ++++++ .../views/workbench/WorkbenchDebugView.vue | 240 +++++++++++++++++- 7 files changed, 641 insertions(+), 8 deletions(-) create mode 100644 internal/cloud/workbench-kafka-sse-debug.test.ts create mode 100644 internal/cloud/workbench-kafka-sse-debug.ts diff --git a/internal/cloud/kafka-event-bridge.ts b/internal/cloud/kafka-event-bridge.ts index e166bb4f..213921d4 100644 --- a/internal/cloud/kafka-event-bridge.ts +++ b/internal/cloud/kafka-event-bridge.ts @@ -206,6 +206,52 @@ export async function queryKafkaEventStream({ env = process.env, stream = "hwlab }; } +export async function openKafkaEventStream({ env = process.env, stream = "hwlab", topic = null, traceId = null, sessionId = null, runId = null, commandId = null, fromBeginning = false, onEvent, onError = null, kafkaFactory = defaultKafkaFactory } = {}) { + if (typeof onEvent !== "function") throw new Error("onEvent callback is required for Kafka event streaming."); + const brokers = csv(env.HWLAB_KAFKA_BOOTSTRAP_SERVERS); + if (brokers.length === 0) throw new Error("HWLAB_KAFKA_BOOTSTRAP_SERVERS is required for Kafka event streams."); + const clientId = stringValue(env.HWLAB_KAFKA_CLIENT_ID) || DEFAULT_CLIENT_ID; + const resolvedTopic = stringValue(topic) || kafkaTopicForStream(stream, env); + const filters = compactObject({ traceId, sessionId, runId, commandId }); + const kafka = kafkaFactory({ brokers, clientId: `${clientId}-sse` }); + const consumer = kafka.consumer({ groupId: `${clientId}-sse-${Date.now()}-${randomUUID().slice(0, 8)}`, allowAutoTopicCreation: false }); + let running = false; + let stopped = false; + await consumer.connect(); + await consumer.subscribe({ topic: resolvedTopic, fromBeginning: fromBeginning === true }); + running = true; + consumer.run({ + eachMessage: async ({ topic: messageTopic, partition, message }) => { + if (stopped) return; + const valueText = message.value ? Buffer.from(message.value).toString("utf8") : ""; + const value = parseJson(valueText); + if (!value || !eventMatchesFilters(value, filters)) return; + await onEvent({ + topic: messageTopic, + partition, + offset: message.offset, + key: message.key ? Buffer.from(message.key).toString("utf8") : null, + timestamp: message.timestamp ?? null, + value + }); + } + }).catch((error) => { + if (!stopped && typeof onError === "function") onError(error); + }); + return { + ok: true, + stream, + topic: resolvedTopic, + filters, + async stop() { + stopped = true; + if (running) await consumer.stop().catch(() => undefined); + await boundedDisconnect(consumer); + }, + valuesPrinted: false + }; +} + function mapAgentRunSourceEventToHwlabEvent({ input, sourceEvent, payload, run, command, traceId, sessionId, sourceSeq, runId, commandId }) { const base = { traceId, diff --git a/internal/cloud/server.ts b/internal/cloud/server.ts index 7012aefb..a2ffde74 100644 --- a/internal/cloud/server.ts +++ b/internal/cloud/server.ts @@ -71,6 +71,7 @@ import { } from "./server-code-agent-http.ts"; import { drainWorkbenchRealtimeConnections, handleWorkbenchReadModelHttp, handleWorkbenchRealtimeHttp } from "./server-workbench-http.ts"; import { handleWorkbenchDebugFakeSseHttp } from "./workbench-debug-fake-sse.ts"; +import { handleWorkbenchKafkaSseDebugHttp } from "./workbench-kafka-sse-debug.ts"; import { handleWorkbenchLaunchHttp } from "./server-workbench-launch-http.ts"; import { startWorkbenchEmptySessionGc } from "./workbench-empty-session-gc.ts"; import { startHwlabKafkaEventBridge } from "./kafka-event-bridge.ts"; @@ -742,6 +743,11 @@ async function handleRestAdapter(request, response, url, options) { return; } + if (url.pathname === "/v1/workbench/debug/kafka-sse" || url.pathname.startsWith("/v1/workbench/debug/kafka-sse/")) { + await handleWorkbenchKafkaSseDebugHttp(request, response, url, options); + return; + } + if (url.pathname === "/v1/workbench/sync" || url.pathname === "/v1/workbench/sessions" || url.pathname.startsWith("/v1/workbench/sessions/") || url.pathname.startsWith("/v1/workbench/turns/") || url.pathname.startsWith("/v1/workbench/traces/")) { await handleWorkbenchReadModelHttp(request, response, url, options); return; @@ -973,7 +979,7 @@ function navIdForRestPath(pathname, method = "GET") { if (pathname === "/v1/project-management" || pathname.startsWith("/v1/project-management/")) return "project.mdtodo"; if (pathname === "/v1/api-keys" || pathname === "/v1/api-keys/default" || pathname.startsWith("/v1/api-keys/")) return "user.apiKeys"; if (pathname === "/v1/users/me/profile" || pathname === "/v1/users/me/password") return "system.settings"; - if (pathname === "/v1/workbench/debug/fake-sse" || pathname.startsWith("/v1/workbench/debug/fake-sse/")) return "workbench.debug"; + if (pathname === "/v1/workbench/debug/fake-sse" || pathname.startsWith("/v1/workbench/debug/fake-sse/") || pathname === "/v1/workbench/debug/kafka-sse" || pathname.startsWith("/v1/workbench/debug/kafka-sse/")) return "workbench.debug"; if (pathname === "/v1/workbench/events" || pathname === "/v1/workbench/sync" || pathname === "/v1/workbench/launches" || pathname === "/v1/workbench/sessions" || pathname.startsWith("/v1/workbench/sessions/") || pathname.startsWith("/v1/workbench/turns/") || pathname.startsWith("/v1/workbench/traces/")) return "workbench.code"; if (pathname === "/v1/agent/chat" || pathname === "/v1/agent/sessions" || pathname.startsWith("/v1/agent/sessions/") || pathname === "/v1/agent/chat/inspect" || pathname.startsWith("/v1/agent/chat/result/") || pathname.startsWith("/v1/agent/turns/") || pathname.startsWith("/v1/agent/traces/") || pathname === "/v1/agent/chat/cancel" || pathname === "/v1/agent/chat/steer") return "workbench.code"; if (pathname === "/v1/admin/provider-profiles" || pathname.startsWith("/v1/admin/provider-profiles/")) return "admin.providerProfiles"; diff --git a/internal/cloud/workbench-kafka-sse-debug.test.ts b/internal/cloud/workbench-kafka-sse-debug.test.ts new file mode 100644 index 00000000..3d300605 --- /dev/null +++ b/internal/cloud/workbench-kafka-sse-debug.test.ts @@ -0,0 +1,94 @@ +import assert from "node:assert/strict"; +import { createServer } from "node:http"; +import { test } from "bun:test"; + +import { handleWorkbenchKafkaSseDebugHttp } from "./workbench-kafka-sse-debug.ts"; + +test("workbench Kafka SSE debug endpoint streams filtered raw HWLAB Kafka events", async () => { + const fakeKafka = createFakeKafkaFactory(); + const env = { + HWLAB_KAFKA_BOOTSTRAP_SERVERS: "127.0.0.1:9092", + HWLAB_KAFKA_CLIENT_ID: "test-hwlab-cloud-api", + HWLAB_KAFKA_EVENT_TOPIC: "hwlab.event.v1" + }; + const server = createServer((request, response) => { + const url = new URL(request.url ?? "/", "http://127.0.0.1"); + void handleWorkbenchKafkaSseDebugHttp(request, response, url, { env, kafkaFactory: fakeKafka.factory, logger: null }); + }); + await listen(server); + const abort = new AbortController(); + try { + const response = await fetch(`${serverUrl(server)}/v1/workbench/debug/kafka-sse/events?stream=hwlab&sessionId=ses_kafka_sse_debug`, { signal: abort.signal }); + assert.equal(response.status, 200); + assert.match(response.headers.get("content-type") ?? "", /text\/event-stream/u); + const reader = response.body?.getReader(); + assert.ok(reader, "SSE response must expose a readable stream"); + const connected = await readUntil(reader, "hwlab.kafka.connected"); + assert.match(connected, /hwlab\.event\.v1/u); + + await fakeKafka.emit({ eventType: "hwlab.trace.event.projected", sessionId: "ses_other", event: { label: "ignored" } }); + await fakeKafka.emit({ eventType: "hwlab.trace.event.projected", sessionId: "ses_kafka_sse_debug", traceId: "trc_kafka_sse_debug", context: { runId: "run_kafka_sse_debug", commandId: "cmd_kafka_sse_debug" }, event: { label: "agentrun:event:debug", status: "running" } }); + const streamed = await readUntil(reader, "trc_kafka_sse_debug"); + assert.match(streamed, /hwlab\.kafka\.event/u); + assert.match(streamed, /ses_kafka_sse_debug/u); + assert.doesNotMatch(streamed, /ses_other/u); + } finally { + abort.abort(); + await close(server); + } +}); + +function createFakeKafkaFactory() { + let eachMessage: ((input: any) => Promise) | null = null; + const consumer = { + connect: async () => undefined, + subscribe: async () => undefined, + run: async (input: any) => { eachMessage = input.eachMessage; }, + stop: async () => undefined, + disconnect: async () => undefined + }; + return { + factory: () => ({ consumer: () => consumer }), + emit: async (value: Record) => { + assert.ok(eachMessage, "consumer.run must be called before emitting fake Kafka events"); + await eachMessage({ + topic: "hwlab.event.v1", + partition: 0, + message: { + offset: String(value.sessionId === "ses_other" ? 1 : 2), + key: Buffer.from(String(value.sessionId ?? "unknown")), + timestamp: "2026-07-09T18:30:00.000Z", + value: Buffer.from(JSON.stringify(value)) + } + }); + } + }; +} + +async function readUntil(reader: ReadableStreamDefaultReader, pattern: string): Promise { + const decoder = new TextDecoder(); + let text = ""; + const deadline = Date.now() + 3000; + while (!text.includes(pattern)) { + if (Date.now() > deadline) throw new Error(`SSE stream did not include ${pattern}: ${text}`); + const next = await reader.read(); + if (next.done) break; + text += decoder.decode(next.value, { stream: true }); + } + return text; +} + +async function listen(server: ReturnType) { + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); +} + +async function close(server: ReturnType) { + await new Promise((resolve, reject) => server.close((error) => error ? reject(error) : resolve())); +} + +function serverUrl(server: ReturnType): string { + const address = server.address(); + assert.equal(typeof address, "object"); + assert.ok(address && typeof address.port === "number"); + return `http://127.0.0.1:${address.port}`; +} diff --git a/internal/cloud/workbench-kafka-sse-debug.ts b/internal/cloud/workbench-kafka-sse-debug.ts new file mode 100644 index 00000000..39107994 --- /dev/null +++ b/internal/cloud/workbench-kafka-sse-debug.ts @@ -0,0 +1,176 @@ +/* + * SPEC: PJ2026-0104010803 Workbench事件流可见性 draft-2026-07-09-p0-kafka-authority. + * Responsibility: debug-only SSE passthrough for HWLAB Kafka event envelopes. + */ +import { sendJson } from "./server-http-utils.ts"; +import { openKafkaEventStream } from "./kafka-event-bridge.ts"; + +const CONTRACT_VERSION = "workbench-debug-kafka-sse-v1"; +const DEFAULT_STREAM = "hwlab"; +const ALLOWED_STREAMS = new Set(["hwlab", "agentrun"]); + +export async function handleWorkbenchKafkaSseDebugHttp(request, response, url, options = {}) { + const route = routeSuffix(url.pathname); + try { + if (route === "") { + if (request.method !== "GET") return methodNotAllowed(response, "GET"); + return sendJson(response, 200, describeKafkaSseDebug(url, options.env ?? process.env)); + } + if (route === "/events") { + if (request.method !== "GET") return methodNotAllowed(response, "GET"); + return openKafkaDebugSse(request, response, url, options); + } + return sendJson(response, 404, debugError("workbench_debug_kafka_sse_route_not_found", "Workbench debug Kafka SSE route is not implemented.", { route })); + } catch (error) { + options.logger?.warn?.({ + event: "workbench_debug_kafka_sse_failed", + route, + errorName: error?.name ?? "Error", + message: error instanceof Error ? error.message : String(error ?? "unknown"), + valuesRedacted: true + }); + if (response.headersSent) return writeSse(response, "hwlab.kafka.error", { ok: false, error: errorMessagePayload(error) }); + return sendJson(response, 500, debugError("workbench_debug_kafka_sse_failed", "Workbench debug Kafka SSE request failed.")); + } +} + +function describeKafkaSseDebug(url, env) { + const stream = streamFromUrl(url); + return { + ok: true, + contractVersion: CONTRACT_VERSION, + stream, + topic: topicForStream(stream, env), + filters: filtersFromUrl(url), + eventsRoute: `/v1/workbench/debug/kafka-sse/events${url.search || ""}`, + valuesPrinted: false + }; +} + +async function openKafkaDebugSse(request, response, url, options) { + const env = options.env ?? process.env; + const stream = streamFromUrl(url); + const filters = filtersFromUrl(url); + response.writeHead(200, { + "content-type": "text/event-stream; charset=utf-8", + "cache-control": "no-store, no-transform", + connection: "keep-alive", + "x-accel-buffering": "no", + "x-content-type-options": "nosniff" + }); + if (typeof response.flushHeaders === "function") response.flushHeaders(); + writeSse(response, "hwlab.kafka.connected", { + ok: true, + contractVersion: CONTRACT_VERSION, + stream, + topic: topicForStream(stream, env), + filters, + serverSentAt: new Date().toISOString(), + valuesPrinted: false + }); + let kafkaStream = null; + let closed = false; + const close = () => { + if (closed) return; + closed = true; + void kafkaStream?.stop?.(); + }; + request.on("close", close); + try { + kafkaStream = await openKafkaEventStream({ + env, + stream, + fromBeginning: url.searchParams.get("fromBeginning") === "1" || url.searchParams.get("fromBeginning") === "true", + ...filters, + kafkaFactory: options.kafkaFactory, + onEvent: async (record) => { + writeSse(response, "hwlab.kafka.event", { + ok: true, + contractVersion: CONTRACT_VERSION, + stream, + topic: record.topic, + partition: record.partition, + offset: record.offset, + key: record.key, + timestamp: record.timestamp, + value: record.value, + serverSentAt: new Date().toISOString(), + valuesPrinted: false + }); + }, + onError: (error) => writeSse(response, "hwlab.kafka.error", { ok: false, error: errorMessagePayload(error), valuesPrinted: false }) + }); + } catch (error) { + writeSse(response, "hwlab.kafka.error", { ok: false, error: errorMessagePayload(error), valuesPrinted: false }); + } +} + +function writeSse(response, eventName, payload) { + if (response.destroyed || response.writableEnded) return false; + try { + response.write(`event: ${eventName}\n`); + const id = sseEventId(payload); + if (id) response.write(`id: ${id}\n`); + response.write(`data: ${JSON.stringify(payload)}\n\n`); + return true; + } catch { + return false; + } +} + +function routeSuffix(pathname) { + return String(pathname || "").replace(/^\/v1\/workbench\/debug\/kafka-sse/u, ""); +} + +function streamFromUrl(url) { + const stream = String(url.searchParams.get("stream") || DEFAULT_STREAM).trim(); + return ALLOWED_STREAMS.has(stream) ? stream : DEFAULT_STREAM; +} + +function filtersFromUrl(url) { + return compactObject({ + traceId: safeId(url.searchParams.get("traceId") || url.searchParams.get("trace-id")), + sessionId: safeId(url.searchParams.get("sessionId") || url.searchParams.get("session-id")), + runId: safeId(url.searchParams.get("runId") || url.searchParams.get("run-id")), + commandId: safeId(url.searchParams.get("commandId") || url.searchParams.get("command-id")) + }); +} + +function topicForStream(stream, env) { + if (stream === "agentrun") return textValue(env.HWLAB_KAFKA_AGENTRUN_EVENT_TOPIC) || "agentrun.event.v1"; + return textValue(env.HWLAB_KAFKA_EVENT_TOPIC) || "hwlab.event.v1"; +} + +function methodNotAllowed(response, allow) { + response.setHeader("allow", allow); + return sendJson(response, 405, debugError("method_not_allowed", `Method not allowed; expected ${allow}.`)); +} + +function debugError(code, message, extra = {}) { + return { ok: false, error: { code, message, ...extra }, valuesPrinted: false }; +} + +function errorMessagePayload(error) { + return { code: "kafka_sse_error", message: error instanceof Error ? error.message : String(error ?? "unknown") }; +} + +function safeId(value) { + const text = textValue(value); + return text && /^[A-Za-z0-9_.:-]{3,220}$/u.test(text) ? text : null; +} + +function textValue(value) { + const text = typeof value === "string" ? value.trim() : value === null || value === undefined ? "" : String(value).trim(); + return text.length > 0 ? text : null; +} + +function compactObject(value) { + return Object.fromEntries(Object.entries(value).filter(([, entry]) => textValue(entry))); +} + +function sseEventId(payload) { + const topic = textValue(payload?.topic); + const partition = textValue(payload?.partition); + const offset = textValue(payload?.offset); + return topic && partition && offset ? `${topic}:${partition}:${offset}` : null; +} diff --git a/web/hwlab-cloud-web/src/api/index.ts b/web/hwlab-cloud-web/src/api/index.ts index b8b17ed0..c941e47a 100644 --- a/web/hwlab-cloud-web/src/api/index.ts +++ b/web/hwlab-cloud-web/src/api/index.ts @@ -4,7 +4,7 @@ export { fetchJson, fetchText, type ActivityRef, type ActivityRefSource, type ApiRequestOptions } from "./client"; export { authAPI, HWLAB_WEB_SESSION_COOKIE } from "./auth"; export { workbenchAPI } from "./workbench"; -export { workbenchDebugAPI, connectWorkbenchDebugFakeSse, type WorkbenchDebugFakeSseQueue, type WorkbenchDebugFakeSseResponse, type WorkbenchDebugFakeSseSequence, type WorkbenchDebugFakeSseStream } from "./workbench-debug"; +export { workbenchDebugAPI, connectWorkbenchDebugFakeSse, connectWorkbenchKafkaSseDebug, type WorkbenchDebugFakeSseQueue, type WorkbenchDebugFakeSseResponse, type WorkbenchDebugFakeSseSequence, type WorkbenchDebugFakeSseStream, type WorkbenchKafkaSseDebugEvent, type WorkbenchKafkaSseDebugFilters } from "./workbench-debug"; export { connectWorkbenchEvents, type WorkbenchEventStream, type WorkbenchRealtimeEvent } from "./workbench-events"; export { agentAPI } from "./agent"; export { hwpodAPI } from "./hwpod"; diff --git a/web/hwlab-cloud-web/src/api/workbench-debug.ts b/web/hwlab-cloud-web/src/api/workbench-debug.ts index 653f256e..00ea5af8 100644 --- a/web/hwlab-cloud-web/src/api/workbench-debug.ts +++ b/web/hwlab-cloud-web/src/api/workbench-debug.ts @@ -50,6 +50,37 @@ export interface WorkbenchDebugFakeSseStream { close: () => void; } +export interface WorkbenchKafkaSseDebugFilters { + traceId?: string; + sessionId?: string; + runId?: string; + commandId?: string; +} + +export interface WorkbenchKafkaSseDebugEvent { + ok?: boolean; + contractVersion?: string; + stream?: string; + topic?: string; + partition?: number; + offset?: string; + key?: string | null; + timestamp?: string | null; + value?: Record; + filters?: WorkbenchKafkaSseDebugFilters; + serverSentAt?: string; + error?: { code?: string; message?: string; [key: string]: unknown }; + valuesPrinted?: boolean; +} + +export interface WorkbenchKafkaSseDebugStreamOptions extends WorkbenchKafkaSseDebugFilters { + stream?: "hwlab" | "agentrun"; + fromBeginning?: boolean; + onEvent: (event: WorkbenchKafkaSseDebugEvent, eventName: string) => void; + onOpen?: () => void; + onError?: (event: Event) => void; +} + const DEBUG_FAKE_SSE_EVENTS = [ "workbench.connected", "workbench.trace.snapshot", @@ -93,6 +124,39 @@ export function connectWorkbenchDebugFakeSse(options: WorkbenchDebugFakeSseStrea }; } +export function connectWorkbenchKafkaSseDebug(options: WorkbenchKafkaSseDebugStreamOptions): WorkbenchDebugFakeSseStream | null { + if (typeof EventSource === "undefined") return null; + const source = new EventSource(kafkaDebugPath("/events", { + stream: options.stream || "hwlab", + traceId: options.traceId, + sessionId: options.sessionId, + runId: options.runId, + commandId: options.commandId, + fromBeginning: options.fromBeginning ? "true" : undefined + }), { withCredentials: true }); + source.onopen = () => options.onOpen?.(); + source.onerror = (event) => options.onError?.(event); + const names = ["hwlab.kafka.connected", "hwlab.kafka.event", "hwlab.kafka.error"]; + const listeners = names.map((name) => { + const listener = (event: MessageEvent) => { + const payload = parseKafkaEvent(event.data); + if (payload) options.onEvent(payload, name); + }; + source.addEventListener(name, listener); + return { name, listener }; + }); + source.onmessage = (event) => { + const payload = parseKafkaEvent(event.data); + if (payload) options.onEvent(payload, "message"); + }; + return { + close() { + for (const { name, listener } of listeners) source.removeEventListener(name, listener); + source.close(); + } + }; +} + function debugPath(suffix: string, params: Record = {}): string { const query = new URLSearchParams(); for (const [key, value] of Object.entries(params)) { @@ -103,6 +167,16 @@ function debugPath(suffix: string, params: Record = {}): string { + const query = new URLSearchParams(); + for (const [key, value] of Object.entries(params)) { + const text = typeof value === "string" ? value.trim() : ""; + if (text) query.set(key, text); + } + const qs = query.toString(); + return `/v1/workbench/debug/kafka-sse${suffix}${qs ? `?${qs}` : ""}`; +} + function parseRealtimeEvent(raw: string): WorkbenchRealtimeEvent | null { try { const value = JSON.parse(raw); @@ -111,3 +185,12 @@ function parseRealtimeEvent(raw: string): WorkbenchRealtimeEvent | null { return null; } } + +function parseKafkaEvent(raw: string): WorkbenchKafkaSseDebugEvent | null { + try { + const value = JSON.parse(raw); + return value && typeof value === "object" ? value as WorkbenchKafkaSseDebugEvent : null; + } catch { + return null; + } +} diff --git a/web/hwlab-cloud-web/src/views/workbench/WorkbenchDebugView.vue b/web/hwlab-cloud-web/src/views/workbench/WorkbenchDebugView.vue index f3b0756f..7f4c0ebf 100644 --- a/web/hwlab-cloud-web/src/views/workbench/WorkbenchDebugView.vue +++ b/web/hwlab-cloud-web/src/views/workbench/WorkbenchDebugView.vue @@ -3,27 +3,42 @@ // Responsibility: Workbench debug route for backend-driven fake SSE single-step Trace card inspection. import { computed, onBeforeUnmount, onMounted, ref } from "vue"; -import { connectWorkbenchDebugFakeSse, workbenchDebugAPI, type WorkbenchDebugFakeSseQueue, type WorkbenchDebugFakeSseSequence, type WorkbenchDebugFakeSseStream } from "@/api"; +import { connectWorkbenchDebugFakeSse, connectWorkbenchKafkaSseDebug, workbenchDebugAPI, type WorkbenchDebugFakeSseQueue, type WorkbenchDebugFakeSseSequence, type WorkbenchDebugFakeSseStream, type WorkbenchKafkaSseDebugEvent } from "@/api"; import StatusBadge from "@/components/common/StatusBadge.vue"; import WorkbenchMessageCard from "@/components/workbench/WorkbenchMessageCard.vue"; import type { WorkbenchRealtimeEvent } from "@/api/workbench-events"; import { applyWorkbenchDebugFakeSseEvent, createWorkbenchDebugFakeSseState } from "@/stores/workbench-debug-fake-sse"; const queueId = "trace-card"; +const activeTab = ref<"fake" | "kafka">("fake"); const selectedSequenceId = ref("trace-card-basic"); const sequences = ref([]); const queue = ref(null); const projection = ref(createWorkbenchDebugFakeSseState()); const streamStatus = ref<"connecting" | "open" | "error" | "closed">("connecting"); +const kafkaStatus = ref<"idle" | "connecting" | "open" | "error" | "closed">("idle"); +const kafkaStreamName = ref<"hwlab" | "agentrun">("hwlab"); +const kafkaFromBeginning = ref(false); +const kafkaFilters = ref({ traceId: "", sessionId: "", runId: "", commandId: "" }); +const kafkaEvents = ref([]); const busy = ref(false); const error = ref(null); const appendDraft = ref(defaultAppendDraft()); let stream: WorkbenchDebugFakeSseStream | null = null; +let kafkaStream: WorkbenchDebugFakeSseStream | null = null; + +interface KafkaRawRow { + id: string; + eventName: string; + receivedAt: string; + payload: WorkbenchKafkaSseDebugEvent; +} const message = computed(() => projection.value.message); const logs = computed(() => projection.value.logs); const queueSummary = computed(() => queue.value ? `${queue.value.cursor}/${queue.value.eventCount}` : "-"); const streamBadgeStatus = computed(() => streamStatus.value === "open" ? "completed" : streamStatus.value === "error" ? "failed" : "running"); +const kafkaBadgeStatus = computed(() => kafkaStatus.value === "open" ? "completed" : kafkaStatus.value === "error" ? "failed" : "running"); const canStep = computed(() => !busy.value && (queue.value?.remaining ?? 0) > 0); onMounted(async () => { @@ -35,7 +50,10 @@ onMounted(async () => { onBeforeUnmount(() => { stream?.close(); stream = null; + kafkaStream?.close(); + kafkaStream = null; streamStatus.value = "closed"; + kafkaStatus.value = "closed"; }); async function refreshDescription(): Promise { @@ -65,6 +83,39 @@ function openStream(): void { if (!stream) streamStatus.value = "error"; } +function openKafkaStream(): void { + kafkaStream?.close(); + kafkaStatus.value = "connecting"; + kafkaStream = connectWorkbenchKafkaSseDebug({ + stream: kafkaStreamName.value, + fromBeginning: kafkaFromBeginning.value, + traceId: cleanFilter(kafkaFilters.value.traceId), + sessionId: cleanFilter(kafkaFilters.value.sessionId), + runId: cleanFilter(kafkaFilters.value.runId), + commandId: cleanFilter(kafkaFilters.value.commandId), + onOpen: () => { kafkaStatus.value = "open"; }, + onError: () => { kafkaStatus.value = "error"; }, + onEvent: (event, eventName) => { + if (eventName === "hwlab.kafka.error") { + kafkaStatus.value = "error"; + error.value = event.error?.message || "Kafka SSE 事件流错误"; + } + kafkaEvents.value = [{ id: kafkaRowId(event, eventName), eventName, receivedAt: new Date().toISOString(), payload: event }, ...kafkaEvents.value].slice(0, 120); + } + }); + if (!kafkaStream) kafkaStatus.value = "error"; +} + +function closeKafkaStream(): void { + kafkaStream?.close(); + kafkaStream = null; + kafkaStatus.value = "closed"; +} + +function clearKafkaEvents(): void { + kafkaEvents.value = []; +} + async function resetQueue(): Promise { busy.value = true; error.value = null; @@ -123,6 +174,38 @@ function parseAppendDraft(raw: string): { ok: true; events: WorkbenchRealtimeEve } } +function cleanFilter(value: string): string | undefined { + const text = value.trim(); + return text ? text : undefined; +} + +function kafkaRowId(event: WorkbenchKafkaSseDebugEvent, eventName: string): string { + const topic = event.topic || "topic"; + const partition = event.partition ?? "p"; + const offset = event.offset || Date.now(); + return `${eventName}:${topic}:${partition}:${offset}:${Math.random().toString(16).slice(2)}`; +} + +function kafkaEventSummary(event: WorkbenchKafkaSseDebugEvent): string { + const value = recordValue(event.value); + const nestedEvent = recordValue(value.event); + return String(value.eventType || value.type || nestedEvent.type || event.error?.code || "event"); +} + +function kafkaEventIds(event: WorkbenchKafkaSseDebugEvent): string { + const value = recordValue(event.value); + const context = recordValue(value.context); + return [value.traceId, value.sessionId, context.runId || value.runId, context.commandId || value.commandId].filter(Boolean).join(" · ") || "-"; +} + +function kafkaEventJson(event: WorkbenchKafkaSseDebugEvent): string { + return JSON.stringify(event.value ?? event, null, 2); +} + +function recordValue(value: unknown): Record { + return value && typeof value === "object" && !Array.isArray(value) ? value as Record : {}; +} + function defaultAppendDraft(): string { return JSON.stringify({ type: "trace.event", @@ -151,16 +234,18 @@ function defaultAppendDraft(): string {

实时调试台

- - Queue {{ queueSummary }} + + + {{ activeTab === 'fake' ? `Queue ${queueSummary}` : `${kafkaEvents.length} events` }}
-
+
+ + + + + + + + +
+

{{ error }}

-
+
等待第一条 fake SSE message.snapshot
@@ -204,6 +322,28 @@ function defaultAppendDraft(): string {
+ +
+
+ Raw SSE Events + {{ kafkaEvents.length }} events +
+
等待后端从 Kafka 订阅并透传 hwlab event。
+
    +
  1. +
    + {{ row.eventName }} + {{ kafkaEventSummary(row.payload) }} + {{ row.payload.topic || '-' }} #{{ row.payload.offset || '-' }} +
    +
    + {{ kafkaEventIds(row.payload) }} + +
    + 
    {{ kafkaEventJson(row.payload) }}
    +
    2. +
+
@@ -286,6 +426,24 @@ function defaultAppendDraft(): string { font-weight: 750; } +.workbench-debug-toolbar .compact-label { + width: min(220px, 100%); +} + +.kafka-toolbar { + align-items: end; +} + +.debug-check { + display: flex !important; + width: auto !important; + grid-auto-flow: column; + align-items: center; + gap: 7px !important; + padding-bottom: 8px; + color: #334155 !important; +} + .workbench-debug-grid { display: grid; min-height: 0; @@ -408,6 +566,76 @@ function defaultAppendDraft(): string { white-space: nowrap; } +.kafka-debug-panel { + display: grid; + min-height: 0; + grid-template-rows: auto minmax(0, 1fr); + gap: 10px; + border: 1px solid #d8e1eb; + border-radius: 8px; + background: white; + padding: 10px; +} + +.kafka-event-list { + display: grid; + align-content: start; + gap: 8px; + min-height: 0; + margin: 0; + overflow: auto; + padding: 0; + list-style: none; +} + +.kafka-event-list li { + display: grid; + gap: 6px; + border: 1px solid #d8e1eb; + border-left: 3px solid #0e7490; + border-radius: 8px; + background: #f8fafc; + padding: 9px; +} + +.kafka-event-head, +.kafka-event-meta { + display: grid; + grid-template-columns: minmax(150px, 0.5fr) minmax(160px, 0.7fr) minmax(0, 1fr); + gap: 10px; + align-items: center; + color: #334155; + font-size: 12px; +} + +.kafka-event-head code { + color: #0f766e; +} + +.kafka-event-head span, +.kafka-event-meta { + color: #64748b; +} + +.kafka-event-meta { + grid-template-columns: minmax(0, 1fr) auto; +} + +.kafka-event-list pre { + max-height: 220px; + margin: 0; + overflow: auto; + border-radius: 7px; + background: #0f172a; + padding: 9px; + color: #d1fae5; + font-family: ui-monospace, SFMono-Regular, Menlo, Monaco, Consolas, monospace; + font-size: 11px; + line-height: 1.45; + white-space: pre-wrap; + overflow-wrap: anywhere; +} + @media (max-width: 980px) { .workbench-debug-grid { grid-template-columns: minmax(0, 1fr);