From 23c11b5849caaa45ab445feaaea276dbf58ae946 Mon Sep 17 00:00:00 2001 From: root Date: Fri, 10 Jul 2026 13:48:15 +0200 Subject: [PATCH] =?UTF-8?q?fix:=20=E5=BC=BA=E5=88=B6=E9=9A=94=E7=A6=BB=20K?= =?UTF-8?q?afka=20=E8=B0=83=E8=AF=95=E9=85=8D=E7=BD=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../cloud/workbench-kafka-debug-capability.ts | 23 +++++++++++++-- .../cloud/workbench-kafka-sse-debug.test.ts | 24 +++++++++++++++ tools/hwlab-cli/kafka-regenerate.test.ts | 22 ++++++++++++++ tools/src/hwlab-cli/kafka-regenerate.ts | 29 +++++++++++++++---- .../WorkbenchKafkaDebugPanel.test.ts | 2 ++ 5 files changed, 93 insertions(+), 7 deletions(-) diff --git a/internal/cloud/workbench-kafka-debug-capability.ts b/internal/cloud/workbench-kafka-debug-capability.ts index e8a311a0..1c1eebd5 100644 --- a/internal/cloud/workbench-kafka-debug-capability.ts +++ b/internal/cloud/workbench-kafka-debug-capability.ts @@ -11,8 +11,18 @@ export const WORKBENCH_KAFKA_DEBUG_ENVS = Object.freeze({ export function workbenchKafkaDebugCapability(env = process.env) { const enabled = requiredBoolean(env?.[WORKBENCH_KAFKA_DEBUG_ENVS.enabled], WORKBENCH_KAFKA_DEBUG_ENVS.enabled); - const topic = requiredKafkaName(env?.[WORKBENCH_KAFKA_DEBUG_ENVS.topic], WORKBENCH_KAFKA_DEBUG_ENVS.topic, 249); - const groupIdPrefix = requiredKafkaName(env?.[WORKBENCH_KAFKA_DEBUG_ENVS.groupIdPrefix], WORKBENCH_KAFKA_DEBUG_ENVS.groupIdPrefix, 180); + const topic = requiredIsolatedKafkaName( + env?.[WORKBENCH_KAFKA_DEBUG_ENVS.topic], + WORKBENCH_KAFKA_DEBUG_ENVS.topic, + 249, + ["hwlab.event.v1", env?.HWLAB_KAFKA_HWLAB_EVENT_TOPIC, env?.HWLAB_KAFKA_EVENT_TOPIC] + ); + const groupIdPrefix = requiredIsolatedKafkaName( + env?.[WORKBENCH_KAFKA_DEBUG_ENVS.groupIdPrefix], + WORKBENCH_KAFKA_DEBUG_ENVS.groupIdPrefix, + 180, + [env?.HWLAB_KAFKA_HWLAB_EVENT_GROUP_ID] + ); const replayLimit = requiredPositiveInteger(env?.[WORKBENCH_KAFKA_DEBUG_ENVS.replayLimit], WORKBENCH_KAFKA_DEBUG_ENVS.replayLimit); const replayTimeoutMs = requiredPositiveInteger(env?.[WORKBENCH_KAFKA_DEBUG_ENVS.replayTimeoutMs], WORKBENCH_KAFKA_DEBUG_ENVS.replayTimeoutMs); return { enabled, topic, groupIdPrefix, replayLimit, replayTimeoutMs, valuesPrinted: false }; @@ -33,6 +43,15 @@ function requiredKafkaName(value, name, maxLength) { return text; } +function requiredIsolatedKafkaName(value, name, maxLength, productValues) { + const text = requiredKafkaName(value, name, maxLength); + const products = new Set(productValues.map((candidate) => String(candidate ?? "").trim()).filter(Boolean)); + if (products.has(text) || !/(?:^|[.-])debug(?:[.-]|$)/u.test(text)) { + throw capabilityError(`${name} must identify an isolated debug Kafka resource and must not reuse a product topic or group.`, name); + } + return text; +} + function requiredPositiveInteger(value, name) { const number = Number(value); if (!Number.isInteger(number) || number <= 0) throw capabilityError(`${name} is required and must be a positive integer.`, name); diff --git a/internal/cloud/workbench-kafka-sse-debug.test.ts b/internal/cloud/workbench-kafka-sse-debug.test.ts index 6d7ba0d1..a01eaf2f 100644 --- a/internal/cloud/workbench-kafka-sse-debug.test.ts +++ b/internal/cloud/workbench-kafka-sse-debug.test.ts @@ -2,6 +2,7 @@ import assert from "node:assert/strict"; import { createServer } from "node:http"; import { test } from "bun:test"; +import { workbenchKafkaDebugCapability } from "./workbench-kafka-debug-capability.ts"; import { handleWorkbenchKafkaSseDebugHttp } from "./workbench-kafka-sse-debug.ts"; test("workbench Kafka SSE debug endpoint streams filtered raw HWLAB Kafka events", async () => { @@ -87,6 +88,29 @@ test("isolated Workbench Kafka replay uses its YAML-owned topic, unique group, a } }); +test("isolated Workbench Kafka capability rejects product topic and group identities", () => { + const env = { + HWLAB_WORKBENCH_KAFKA_DEBUG_REPLAY_ENABLED: "true", + HWLAB_KAFKA_HWLAB_DEBUG_EVENT_TOPIC: "hwlab.event.debug.v1", + HWLAB_KAFKA_HWLAB_DEBUG_GROUP_PREFIX: "hwlab-test-workbench-isolated-debug", + HWLAB_WORKBENCH_KAFKA_DEBUG_REPLAY_LIMIT: "10", + HWLAB_WORKBENCH_KAFKA_DEBUG_REPLAY_TIMEOUT_MS: "3000", + HWLAB_KAFKA_HWLAB_EVENT_TOPIC: "hwlab.event.v1", + HWLAB_KAFKA_HWLAB_EVENT_GROUP_ID: "hwlab-v03-workbench-live-sse" + }; + + assert.throws( + () => workbenchKafkaDebugCapability({ ...env, HWLAB_KAFKA_HWLAB_DEBUG_EVENT_TOPIC: "hwlab.event.v1" }), + (error) => error?.code === "workbench_kafka_debug_capability_invalid" + && error?.envName === "HWLAB_KAFKA_HWLAB_DEBUG_EVENT_TOPIC" + ); + assert.throws( + () => workbenchKafkaDebugCapability({ ...env, HWLAB_KAFKA_HWLAB_DEBUG_GROUP_PREFIX: "hwlab-v03-workbench-live-sse" }), + (error) => error?.code === "workbench_kafka_debug_capability_invalid" + && error?.envName === "HWLAB_KAFKA_HWLAB_DEBUG_GROUP_PREFIX" + ); +}); + test("isolated Workbench Kafka debug fails closed when disabled and rejects replay", async () => { const fakeKafka = createFakeKafkaFactory(); const env = { diff --git a/tools/hwlab-cli/kafka-regenerate.test.ts b/tools/hwlab-cli/kafka-regenerate.test.ts index 06a00f52..ac287665 100644 --- a/tools/hwlab-cli/kafka-regenerate.test.ts +++ b/tools/hwlab-cli/kafka-regenerate.test.ts @@ -154,6 +154,28 @@ test("reconstruction identity fabrication and product output topic fail closed", ], { cwd, env: {}, now: () => "2026-07-10T12:00:00.000Z" }); assert.equal(productResult.exitCode, 1); assert.equal(productResult.payload.error.code, "product_topic_forbidden"); + + const productGroupResult = await runKafkaCli([ + "regenerate", "hwlab", "--from", "jsonl", "--session-id", SESSION_ID, + "--jsonl-file", inputFile, "--group-prefix", "hwlab-v03-workbench-live-sse", "--no-publish" + ], { cwd, env: {}, now: () => "2026-07-10T12:00:00.000Z" }); + assert.equal(productGroupResult.exitCode, 1); + assert.equal(productGroupResult.payload.error.code, "debug_group_required"); +}); + +test("Kafka mode requires a CLI or YAML-owned isolated group prefix", async () => { + let readCalled = false; + const result = await runKafkaCli([ + "regenerate", "hwlab", "--from", "kafka", "--session-id", SESSION_ID, "--no-publish" + ], { + env: {}, + now: () => "2026-07-10T12:00:00.000Z", + async readKafka() { readCalled = true; return { events: [] }; } + }); + + assert.equal(result.exitCode, 1); + assert.equal(result.payload.error.code, "kafka_debug_group_missing"); + assert.equal(readCalled, false); }); test("Kafka query exposes the isolated generated group and original value hash", async () => { diff --git a/tools/src/hwlab-cli/kafka-regenerate.ts b/tools/src/hwlab-cli/kafka-regenerate.ts index bc5b9db4..d0a4a18d 100644 --- a/tools/src/hwlab-cli/kafka-regenerate.ts +++ b/tools/src/hwlab-cli/kafka-regenerate.ts @@ -16,7 +16,6 @@ import { const CLI_NAME = "hwlab-cli"; const VERSION = "0.3.0-kafka-debug"; -const DEFAULT_GROUP_PREFIX = "hwlab-v03-workbench-isolated-debug"; const DEFAULT_LIMIT = 500; const DEFAULT_TIMEOUT_MS = 5000; @@ -70,7 +69,6 @@ export async function regenerateHwlabDebugEvents(parsed: ParsedArgs, dependencie const sourceMode = enumValue(parsed.from ?? "kafka", "from", ["kafka", "jsonl"]); const inputTopicResolution = resolveConfig(parsed.inputTopic, undefined, DEFAULT_AGENTRUN_DEBUG_EVENT_TOPIC, "--input-topic", "contract-default"); const outputTopicResolution = resolveConfig(parsed.outputTopic, dependencies.env.HWLAB_KAFKA_HWLAB_DEBUG_EVENT_TOPIC, DEFAULT_HWLAB_DEBUG_EVENT_TOPIC, "--output-topic", "env:HWLAB_KAFKA_HWLAB_DEBUG_EVENT_TOPIC"); - const groupPrefixResolution = resolveConfig(parsed.groupPrefix, dependencies.env.HWLAB_KAFKA_HWLAB_DEBUG_GROUP_PREFIX, DEFAULT_GROUP_PREFIX, "--group-prefix", "env:HWLAB_KAFKA_HWLAB_DEBUG_GROUP_PREFIX"); const inputTopic = inputTopicResolution.value; const outputTopic = outputTopicResolution.value; assertDebugTopic(outputTopic); @@ -78,9 +76,14 @@ export async function regenerateHwlabDebugEvents(parsed: ParsedArgs, dependencie const timeoutMs = boundedInteger(parsed.timeoutMs, DEFAULT_TIMEOUT_MS, 250, 60000, "timeoutMs"); const expectedCount = optionalPositiveInteger(parsed.expectCount, "expectCount"); const readLimit = expectedCount === undefined ? limit : Math.min(5000, expectedCount + 1); - const groupPrefix = groupPrefixResolution.value; - assertDebugGroup(groupPrefix); const shouldPublish = parsed.noPublish === true ? false : parsed.publish === true ? true : sourceMode === "kafka"; + const groupPrefixResolution = resolveKafkaGroupConfig( + parsed.groupPrefix, + dependencies.env.HWLAB_KAFKA_HWLAB_DEBUG_GROUP_PREFIX, + sourceMode === "kafka" || shouldPublish + ); + const groupPrefix = text(groupPrefixResolution.value); + if (groupPrefix) assertDebugGroup(groupPrefix); const readResult = sourceMode === "jsonl" ? await readJsonlInput(requiredText(parsed.jsonlFile, "jsonlFile"), dependencies.cwd, inputTopic) @@ -275,7 +278,8 @@ function kafkaHelp() { defaults: { inputTopic: DEFAULT_AGENTRUN_DEBUG_EVENT_TOPIC, outputTopic: DEFAULT_HWLAB_DEBUG_EVENT_TOPIC, - groupPrefix: DEFAULT_GROUP_PREFIX, + groupPrefix: null, + groupPrefixAuthority: "--group-prefix or env:HWLAB_KAFKA_HWLAB_DEBUG_GROUP_PREFIX when Kafka is used", kafkaInputPublishes: true, jsonlInputPublishes: false }, @@ -472,6 +476,21 @@ function resolveConfig(cliValue: unknown, envValue: unknown, fallback: string, c return { value: fallback, source: "contract-default" }; } +function resolveKafkaGroupConfig(cliValue: unknown, envValue: unknown, required: boolean) { + const explicit = text(cliValue); + if (explicit) return { value: explicit, source: "--group-prefix" }; + const injected = text(envValue); + if (injected) return { value: injected, source: "env:HWLAB_KAFKA_HWLAB_DEBUG_GROUP_PREFIX" }; + if (required) { + throw cliError( + "kafka_debug_group_missing", + "Kafka debug operations require --group-prefix or YAML-injected HWLAB_KAFKA_HWLAB_DEBUG_GROUP_PREFIX", + { field: "HWLAB_KAFKA_HWLAB_DEBUG_GROUP_PREFIX" } + ); + } + return { value: null, source: "offline-no-kafka" }; +} + function optionalPositiveInteger(value: unknown, field: string) { if (value === undefined) return undefined; return boundedInteger(value, 0, 1, 5000, field); diff --git a/web/hwlab-cloud-web/src/components/workbench/WorkbenchKafkaDebugPanel.test.ts b/web/hwlab-cloud-web/src/components/workbench/WorkbenchKafkaDebugPanel.test.ts index 2d897d7e..cf36cb9e 100644 --- a/web/hwlab-cloud-web/src/components/workbench/WorkbenchKafkaDebugPanel.test.ts +++ b/web/hwlab-cloud-web/src/components/workbench/WorkbenchKafkaDebugPanel.test.ts @@ -1,3 +1,5 @@ +// @vitest-environment jsdom + import { afterEach, beforeEach, describe, expect, it } from "vitest"; import { mount } from "@vue/test-utils"; import { nextTick } from "vue";