feat: 沉淀 Workbench Kafka 隔离重放探针
This commit is contained in:
Codex
@@ -63,6 +63,7 @@ export function hwlabNodeWebProbeHelp(): Record<string, unknown> {
|
||||
"bun scripts/cli.ts web-probe observe start --node <node> --lane <lane> --origin public --target-path /projects/mdtodo --sample-interval-ms 5000",
|
||||
"bun scripts/cli.ts web-probe observe command webobs-xxxx --type sendPrompt --text 'ping'",
|
||||
"bun scripts/cli.ts web-probe observe command webobs-xxxx --type validateRealtimeFanout --profile pure-kafka-live --provider gpt.pika --text '<first-turn>' --second-text '<second-turn>'",
|
||||
"bun scripts/cli.ts web-probe observe command webobs-xxxx --type validateWorkbenchKafkaDebugReplay",
|
||||
"bun scripts/cli.ts web-probe observe status webobs-xxxx --command-id cmd-xxxx",
|
||||
"bun scripts/cli.ts web-probe observe command webobs-xxxx --type closeMdtodoSourceConfig",
|
||||
"bun scripts/cli.ts web-probe observe command webobs-xxxx --type selectMdtodoFile --filename 20260609_频率判断_用户反馈.md",
|
||||
@@ -113,6 +114,7 @@ export function hwlabNodeWebProbeHelp(): Record<string, unknown> {
|
||||
"After observe start, prefer observe status|command|stop|collect|analyze <id> instead of repeating --node/--lane/--state-dir.",
|
||||
"Typed observe commands inherit the semantic origin selected by observe start and must not embed a replacement URL or IP.",
|
||||
"validateRealtimeFanout is a YAML-profiled background command: it gates submit on two live product subscribers plus three Kafka debug consumers, validates two turns and disconnect/reconnect without replay, and is polled by exact command id.",
|
||||
"validateWorkbenchKafkaDebugReplay 复用 observer 当前 Workbench session，以 owning YAML 中的 topic、group prefix 和超时验证隔离 Kafka 重放，并通过精确 command id 查询结果。",
|
||||
"collect views render bounded summaries from existing artifacts and do not create a second source of truth.",
|
||||
"performanceCapture records an explicit bounded Chrome CPU profile and drains LongTask/LoAF/event-loop-gap artifacts; performance-summary reads existing artifacts only.",
|
||||
"analyze is offline-only: it reads artifact JSONL and writes analysis/report.md plus analysis/report.json.",
|
||||
|
||||
@@ -154,6 +154,14 @@ export interface HwlabRuntimeWebProbeSpec {
|
||||
readonly browserFreezePolicy?: HwlabRuntimeWebProbeBrowserFreezePolicySpec;
|
||||
readonly projectManagement?: HwlabRuntimeWebProbeProjectManagementSpec;
|
||||
readonly realtimeFanoutProfiles?: Readonly<Record<string, HwlabRuntimeWebProbeRealtimeFanoutProfileSpec>>;
|
||||
readonly workbenchKafkaDebugReplay?: HwlabRuntimeWebProbeWorkbenchKafkaDebugReplaySpec;
|
||||
}
|
||||
|
||||
export interface HwlabRuntimeWebProbeWorkbenchKafkaDebugReplaySpec {
|
||||
readonly enabled: boolean;
|
||||
readonly topic: string;
|
||||
readonly groupPrefix: string;
|
||||
readonly completionTimeoutMs: number;
|
||||
}
|
||||
|
||||
export interface HwlabRuntimeWebProbeRealtimeFanoutProfileSpec {
|
||||
@@ -1298,6 +1306,27 @@ function webProbeConfig(value: unknown, path: string): HwlabRuntimeWebProbeSpec
|
||||
...(raw.browserFreezePolicy === undefined ? {} : { browserFreezePolicy: webProbeBrowserFreezePolicyConfig(raw.browserFreezePolicy, `${path}.browserFreezePolicy`) }),
|
||||
...(raw.projectManagement === undefined ? {} : { projectManagement: webProbeProjectManagementConfig(raw.projectManagement, `${path}.projectManagement`) }),
|
||||
...(raw.realtimeFanoutProfiles === undefined ? {} : { realtimeFanoutProfiles: webProbeRealtimeFanoutProfilesConfig(raw.realtimeFanoutProfiles, `${path}.realtimeFanoutProfiles`) }),
|
||||
...(raw.workbenchKafkaDebugReplay === undefined
|
||||
? {}
|
||||
: { workbenchKafkaDebugReplay: webProbeWorkbenchKafkaDebugReplayConfig(raw.workbenchKafkaDebugReplay, `${path}.workbenchKafkaDebugReplay`) }),
|
||||
};
|
||||
}
|
||||
|
||||
function webProbeWorkbenchKafkaDebugReplayConfig(value: unknown, path: string): HwlabRuntimeWebProbeWorkbenchKafkaDebugReplaySpec {
|
||||
const raw = asRecord(value, path);
|
||||
const topic = stringField(raw, "topic", path);
|
||||
const groupPrefix = stringField(raw, "groupPrefix", path);
|
||||
const completionTimeoutMs = numberField(raw, "completionTimeoutMs", path);
|
||||
if (!/^[A-Za-z0-9._-]+$/u.test(topic)) throw new Error(`${path}.topic must use Kafka topic identifier characters`);
|
||||
if (!/^[A-Za-z0-9._-]+$/u.test(groupPrefix)) throw new Error(`${path}.groupPrefix must use Kafka consumer group identifier characters`);
|
||||
if (!Number.isInteger(completionTimeoutMs) || completionTimeoutMs < 1000 || completionTimeoutMs > 120000) {
|
||||
throw new Error(`${path}.completionTimeoutMs must be an integer between 1000 and 120000`);
|
||||
}
|
||||
return {
|
||||
enabled: booleanField(raw, "enabled", path),
|
||||
topic,
|
||||
groupPrefix,
|
||||
completionTimeoutMs,
|
||||
};
|
||||
}
|
||||
|
||||
|
||||
@@ -175,6 +175,19 @@ function renderWebObserveStatusTable(value: Record<string, unknown>): string {
|
||||
]]),
|
||||
"",
|
||||
] : []),
|
||||
...(exactCommand.type === "validateWorkbenchKafkaDebugReplay" && exactResult !== null ? [
|
||||
"Workbench Kafka debug replay:",
|
||||
webObserveTable(["TRACE", "TOPIC", "GROUP", "RECEIVED", "APPLIED", "TERMINAL", "SCREENSHOT_SHA"], [[
|
||||
webObserveShort(webObserveText(exactResult.traceId), 28),
|
||||
webObserveShort(webObserveText(exactResult.topic), 32),
|
||||
webObserveShort(webObserveText(exactResult.groupId), 48),
|
||||
exactResult.receivedCount,
|
||||
exactResult.appliedCount,
|
||||
exactResult.terminalStatus,
|
||||
webObserveShort(webObserveText(record(exactResult.screenshot)?.sha256), 32),
|
||||
]]),
|
||||
"",
|
||||
] : []),
|
||||
"Recent samples:",
|
||||
webObserveTable(["SEQ", "TS", "PATH", "ROUTE_SESSION", "ACTIVE_SESSION", "MSG", "TRACE"], samples.length > 0 ? samples.map((sample) => [
|
||||
sample.seq,
|
||||
@@ -225,6 +238,7 @@ function renderWebObserveCommandTable(value: Record<string, unknown>): string {
|
||||
const control = record(value.control);
|
||||
const asyncTurn = record(value.asyncTurn);
|
||||
const result = record(observer?.result);
|
||||
const resultScreenshot = record(result?.screenshot);
|
||||
const error = record(observer?.error);
|
||||
const details = record(error?.details);
|
||||
const rawReadiness = record(error?.navigationReadiness) ?? record(details?.readiness) ?? record(details?.readinessAfterWait) ?? record(details?.readinessBeforeClick);
|
||||
@@ -245,6 +259,20 @@ function renderWebObserveCommandTable(value: Record<string, unknown>): string {
|
||||
webObserveShort(webObserveText(asyncTurn?.traceId), 24),
|
||||
webObserveShort(webObserveText(result?.mark ?? result?.currentUrl ?? observer?.error ?? observer?.queued), 80),
|
||||
]]),
|
||||
...(observerCommand?.type === "validateWorkbenchKafkaDebugReplay" && result !== null ? [
|
||||
"",
|
||||
"Workbench Kafka debug replay:",
|
||||
webObserveTable(["TRACE", "TOPIC", "GROUP", "RECEIVED", "APPLIED", "TERMINAL", "REPORT_SHA", "SCREENSHOT_SHA"], [[
|
||||
webObserveShort(webObserveText(result.traceId), 28),
|
||||
webObserveShort(webObserveText(result.topic), 32),
|
||||
webObserveShort(webObserveText(result.groupId), 48),
|
||||
result.receivedCount,
|
||||
result.appliedCount,
|
||||
result.terminalStatus,
|
||||
webObserveShort(webObserveText(result.reportSha256), 32),
|
||||
webObserveShort(webObserveText(resultScreenshot?.sha256), 32),
|
||||
]]),
|
||||
] : []),
|
||||
...(readiness !== null ? [
|
||||
"",
|
||||
"Target readiness:",
|
||||
|
||||
@@ -102,6 +102,7 @@ async function processCommand(command) {
|
||||
expectedActionWaitMs: command.expectedActionWaitMs,
|
||||
}), "sendPrompt");
|
||||
case "validateRealtimeFanout": return validateRealtimeFanout(command);
|
||||
case "validateWorkbenchKafkaDebugReplay": return withObserverSync(await validateWorkbenchKafkaDebugReplay(command), "validateWorkbenchKafkaDebugReplay");
|
||||
case "steer": return withObserverSync(await sendPrompt(String(command.text || ""), { expectedAction: "steer", responsePath: "/v1/agent/chat/steer", noActiveReason: "steer-no-active-turn", expectedActionWaitMs: command.expectedActionWaitMs }), "steer");
|
||||
case "cancel": return withObserverSync(await cancelRunningTurn(), "cancel");
|
||||
case "selectProvider": return withObserverSync(await selectProvider(String(command.provider || command.value || command.text || "")), "selectProvider");
|
||||
|
||||
@@ -0,0 +1,41 @@
|
||||
import assert from "node:assert/strict";
|
||||
import { test } from "bun:test";
|
||||
|
||||
import { nodeWebObserveRunnerKafkaDebugReplaySource } from "./hwlab-node-web-observe-runner-kafka-debug-replay-source";
|
||||
import { nodeWebObserveRunnerSource } from "./hwlab-node-web-observe-runner-source";
|
||||
|
||||
test("Workbench Kafka debug replay source parses the YAML contract and reducer counts", () => {
|
||||
const source = nodeWebObserveRunnerKafkaDebugReplaySource();
|
||||
const build = new Function(`${source}\nreturn { parseWorkbenchKafkaDebugReplayConfig, workbenchKafkaDebugReplayCounts };`) as () => {
|
||||
parseWorkbenchKafkaDebugReplayConfig: (raw: string) => Record<string, unknown>;
|
||||
workbenchKafkaDebugReplayCounts: (text: string) => Record<string, number> | null;
|
||||
};
|
||||
const helpers = build();
|
||||
assert.deepEqual(helpers.parseWorkbenchKafkaDebugReplayConfig(JSON.stringify({
|
||||
enabled: true,
|
||||
topic: "hwlab.event.debug.v1",
|
||||
groupPrefix: "hwlab-v03-workbench-isolated-debug",
|
||||
completionTimeoutMs: 30_000,
|
||||
})), {
|
||||
enabled: true,
|
||||
topic: "hwlab.event.debug.v1",
|
||||
groupPrefix: "hwlab-v03-workbench-isolated-debug",
|
||||
completionTimeoutMs: 30_000,
|
||||
valuesRedacted: true,
|
||||
});
|
||||
assert.deepEqual(helpers.workbenchKafkaDebugReplayCounts("35/35 applied"), { appliedCount: 35, receivedCount: 35 });
|
||||
assert.equal(helpers.workbenchKafkaDebugReplayCounts("等待重放"), null);
|
||||
});
|
||||
|
||||
test("generated observer runner contains the typed debug replay command and passes node syntax check", async () => {
|
||||
const source = nodeWebObserveRunnerSource();
|
||||
assert.match(source, /validateWorkbenchKafkaDebugReplay/u);
|
||||
assert.match(source, /workbench-kafka-debug-toggle/u);
|
||||
assert.match(source, /workbench-kafka-debug-replay/u);
|
||||
assert.match(source, /workbench-kafka-debug-counts/u);
|
||||
const child = Bun.spawn(["node", "--input-type=module", "--check", "-"], { stdin: "pipe", stdout: "pipe", stderr: "pipe" });
|
||||
child.stdin.write(source);
|
||||
child.stdin.end();
|
||||
const [stderr, exitCode] = await Promise.all([new Response(child.stderr).text(), child.exited]);
|
||||
assert.equal(exitCode, 0, stderr);
|
||||
});
|
||||
@@ -0,0 +1,240 @@
|
||||
// SPEC: PJ2026-0104010803 Workbench realtime visibility.
|
||||
// Responsibility: Typed, isolated Workbench Kafka debug replay validation inside the persistent Web observer.
|
||||
|
||||
export function nodeWebObserveRunnerKafkaDebugReplaySource(): string {
|
||||
return String.raw`
|
||||
function parseWorkbenchKafkaDebugReplayConfig(raw) {
|
||||
if (!raw) return null;
|
||||
let parsed;
|
||||
try {
|
||||
parsed = JSON.parse(raw);
|
||||
} catch (error) {
|
||||
throw new Error("UNIDESK_WEB_OBSERVE_WORKBENCH_KAFKA_DEBUG_REPLAY_JSON is invalid JSON: " + (error instanceof Error ? error.message : String(error)));
|
||||
}
|
||||
if (!parsed || typeof parsed !== "object" || Array.isArray(parsed)) {
|
||||
throw new Error("UNIDESK_WEB_OBSERVE_WORKBENCH_KAFKA_DEBUG_REPLAY_JSON must be an object");
|
||||
}
|
||||
const topic = String(parsed.topic || "").trim();
|
||||
const groupPrefix = String(parsed.groupPrefix || "").trim();
|
||||
const completionTimeoutMs = Number(parsed.completionTimeoutMs);
|
||||
if (!/^[A-Za-z0-9._-]+$/u.test(topic)) throw new Error("Workbench Kafka debug replay topic is missing or invalid in the owning YAML");
|
||||
if (!/^[A-Za-z0-9._-]+$/u.test(groupPrefix)) throw new Error("Workbench Kafka debug replay groupPrefix is missing or invalid in the owning YAML");
|
||||
if (!Number.isInteger(completionTimeoutMs) || completionTimeoutMs < 1000 || completionTimeoutMs > 120000) {
|
||||
throw new Error("Workbench Kafka debug replay completionTimeoutMs must be 1000-120000 in the owning YAML");
|
||||
}
|
||||
return { enabled: parsed.enabled === true, topic, groupPrefix, completionTimeoutMs, valuesRedacted: true };
|
||||
}
|
||||
|
||||
function workbenchKafkaDebugReplayCounts(value) {
|
||||
const match = String(value || "").match(/(\d+)\s*\/\s*(\d+)\s+applied/iu);
|
||||
return match ? { appliedCount: Number(match[1]), receivedCount: Number(match[2]) } : null;
|
||||
}
|
||||
|
||||
async function validateWorkbenchKafkaDebugReplay(command) {
|
||||
const config = workbenchKafkaDebugReplay;
|
||||
if (!config || config.enabled !== true) throw new Error("validateWorkbenchKafkaDebugReplay is not enabled by the owning YAML");
|
||||
const reportDir = path.join(stateDir, "artifacts", "workbench-kafka-debug-replay", safeId(command.id));
|
||||
const report = {
|
||||
ok: false,
|
||||
contract: "web-probe-workbench-kafka-debug-replay-v1",
|
||||
commandId: command.id,
|
||||
expected: { topic: config.topic, groupPrefix: config.groupPrefix, completionTimeoutMs: config.completionTimeoutMs, valuesRedacted: true },
|
||||
origin,
|
||||
startedAt: new Date().toISOString(),
|
||||
valuesRedacted: true,
|
||||
};
|
||||
const requests = [];
|
||||
let requestPage = null;
|
||||
const requestListener = (request) => {
|
||||
let url;
|
||||
try { url = new URL(request.url()); } catch { return; }
|
||||
if (url.pathname !== "/v1/workbench/debug/kafka-sse/events") return;
|
||||
requests.push({
|
||||
method: request.method(),
|
||||
path: url.pathname,
|
||||
stream: url.searchParams.get("stream"),
|
||||
traceId: url.searchParams.get("traceId"),
|
||||
fromBeginning: url.searchParams.get("fromBeginning"),
|
||||
valuesRedacted: true,
|
||||
});
|
||||
};
|
||||
try {
|
||||
const controlRecovery = await ensureControlPageResponsiveForCommand("validateWorkbenchKafkaDebugReplay");
|
||||
const snapshot = await workbenchSessionSnapshot();
|
||||
const currentPath = safeUrlPath(currentPageUrl()) || "";
|
||||
if (!isWorkbenchPathname(currentPath)) throw new Error("validateWorkbenchKafkaDebugReplay requires the original Workbench page");
|
||||
const sessionId = snapshot?.activeSessionId || snapshot?.routeSessionId || routeSessionIdFromUrl(currentPageUrl());
|
||||
if (!sessionId) throw new Error("validateWorkbenchKafkaDebugReplay requires an existing Workbench session");
|
||||
|
||||
const toggle = page.locator('[data-testid="workbench-kafka-debug-toggle"]').first();
|
||||
const toggleVisible = await toggle.waitFor({ state: "visible", timeout: 10000 }).then(() => true).catch(() => false);
|
||||
if (!toggleVisible) throw new Error("Workbench Kafka debug toggle is not visible; verify the owning YAML capability and admin session");
|
||||
if (await toggle.isDisabled()) throw new Error("Workbench Kafka debug toggle is disabled; verify the YAML capability and admin session");
|
||||
const toggleCount = await page.locator('[data-testid="workbench-kafka-debug-toggle"]').count();
|
||||
if (toggleCount !== 1) throw new Error("Workbench Kafka debug toggle count is " + toggleCount + ", expected 1");
|
||||
const toggleCheckedBefore = await toggle.isChecked();
|
||||
if (!toggleCheckedBefore) await toggle.check({ timeout: 10000 });
|
||||
|
||||
const panel = page.locator('[data-testid="workbench-kafka-debug-panel"]').first();
|
||||
const clearButton = page.locator('[data-testid="workbench-kafka-debug-clear"]').first();
|
||||
const replayButton = page.locator('[data-testid="workbench-kafka-debug-replay"]').first();
|
||||
const countsNode = page.locator('[data-testid="workbench-kafka-debug-counts"]').first();
|
||||
await panel.waitFor({ state: "visible", timeout: 10000 });
|
||||
await clearButton.waitFor({ state: "visible", timeout: 10000 });
|
||||
await replayButton.waitFor({ state: "visible", timeout: 10000 });
|
||||
await clearButton.click({ timeout: 10000 });
|
||||
const cleared = await workbenchKafkaDebugReplayWaitForCleared(panel, countsNode, 5000);
|
||||
const before = await workbenchKafkaDebugReplayPanelSnapshot(panel, countsNode);
|
||||
if (!/^trc_[A-Za-z0-9_-]+$/u.test(before.traceId || "")) throw new Error("Workbench Kafka debug replay requires the current traceId");
|
||||
if (await replayButton.isDisabled()) throw new Error("Workbench Kafka debug replay button is disabled for traceId " + before.traceId);
|
||||
|
||||
requestPage = page;
|
||||
requestPage.on("request", requestListener);
|
||||
await replayButton.click({ timeout: 10000 });
|
||||
const terminalStatus = await workbenchKafkaDebugReplayWaitForTerminal(panel, config.completionTimeoutMs);
|
||||
if (terminalStatus !== "completed") throw new Error("Workbench Kafka debug replay ended with status " + terminalStatus);
|
||||
const after = await workbenchKafkaDebugReplayPanelSnapshot(panel, countsNode);
|
||||
const matchingRequest = requests.find((item) => item.stream === "hwlab-debug" && item.traceId === before.traceId && item.fromBeginning === "true") || null;
|
||||
if (!matchingRequest) throw new Error("Workbench Kafka debug replay did not issue the expected trace-scoped fromBeginning request");
|
||||
if (after.traceId !== before.traceId) throw new Error("Workbench Kafka debug replay traceId changed during replay");
|
||||
if (after.topic !== config.topic) throw new Error("Workbench Kafka debug replay topic mismatch: " + (after.topic || "missing"));
|
||||
if (!after.groupId || !after.groupId.startsWith(config.groupPrefix + "-")) {
|
||||
throw new Error("Workbench Kafka debug replay groupId does not use the YAML group prefix");
|
||||
}
|
||||
if (!after.deliveryText.includes("debug-replay") || !after.deliveryText.includes("fromBeginning=true")) {
|
||||
throw new Error("Workbench Kafka debug replay delivery contract is missing");
|
||||
}
|
||||
if (!after.counts || after.counts.receivedCount <= 0 || after.counts.appliedCount <= 0) {
|
||||
throw new Error("Workbench Kafka debug replay received/applied counts must both be greater than zero");
|
||||
}
|
||||
|
||||
const screenshot = await captureScreenshot("workbench-kafka-debug-replay-" + safeId(command.id));
|
||||
Object.assign(report, {
|
||||
ok: true,
|
||||
status: "passed",
|
||||
completedAt: new Date().toISOString(),
|
||||
sessionId,
|
||||
traceId: after.traceId,
|
||||
topic: after.topic,
|
||||
groupId: after.groupId,
|
||||
groupPrefix: config.groupPrefix,
|
||||
receivedCount: after.counts.receivedCount,
|
||||
appliedCount: after.counts.appliedCount,
|
||||
terminalStatus,
|
||||
toggle: { available: true, enabled: true, checkedBefore: toggleCheckedBefore, checkedAfter: await toggle.isChecked(), valuesRedacted: true },
|
||||
clear: cleared,
|
||||
request: matchingRequest,
|
||||
screenshot: { path: screenshot.path, sha256: screenshot.sha256, valuesRedacted: true },
|
||||
controlRecovery,
|
||||
valuesRedacted: true,
|
||||
});
|
||||
const reportArtifact = await workbenchKafkaDebugReplayWriteReport(reportDir, report);
|
||||
report.reportPath = reportArtifact.path;
|
||||
report.reportSha256 = reportArtifact.sha256;
|
||||
await appendJsonl(files.control, eventRecord("workbench-kafka-debug-replay-validated", {
|
||||
sessionId,
|
||||
traceId: report.traceId,
|
||||
topic: report.topic,
|
||||
groupId: report.groupId,
|
||||
receivedCount: report.receivedCount,
|
||||
appliedCount: report.appliedCount,
|
||||
reportPath: report.reportPath,
|
||||
reportSha256: report.reportSha256,
|
||||
screenshot: report.screenshot,
|
||||
valuesRedacted: true,
|
||||
}));
|
||||
return {
|
||||
ok: true,
|
||||
status: "passed",
|
||||
sessionId,
|
||||
traceId: report.traceId,
|
||||
topic: report.topic,
|
||||
groupId: report.groupId,
|
||||
groupPrefix: report.groupPrefix,
|
||||
receivedCount: report.receivedCount,
|
||||
appliedCount: report.appliedCount,
|
||||
terminalStatus,
|
||||
reportPath: report.reportPath,
|
||||
reportSha256: report.reportSha256,
|
||||
screenshot: report.screenshot,
|
||||
valuesRedacted: true,
|
||||
};
|
||||
} catch (error) {
|
||||
Object.assign(report, {
|
||||
ok: false,
|
||||
status: "failed",
|
||||
completedAt: new Date().toISOString(),
|
||||
failure: errorSummary(error),
|
||||
requests,
|
||||
valuesRedacted: true,
|
||||
});
|
||||
const reportArtifact = await workbenchKafkaDebugReplayWriteReport(reportDir, report).catch(() => null);
|
||||
const wrapped = error instanceof Error ? error : new Error(String(error));
|
||||
wrapped.details = {
|
||||
...(wrapped.details || {}),
|
||||
reportPath: reportArtifact?.path || null,
|
||||
reportSha256: reportArtifact?.sha256 || null,
|
||||
valuesRedacted: true,
|
||||
};
|
||||
throw wrapped;
|
||||
} finally {
|
||||
if (requestPage) requestPage.off("request", requestListener);
|
||||
}
|
||||
}
|
||||
|
||||
async function workbenchKafkaDebugReplayWaitForCleared(panel, countsNode, timeoutMs) {
|
||||
const deadline = Date.now() + timeoutMs;
|
||||
while (Date.now() <= deadline) {
|
||||
const status = await panel.getAttribute("data-replay-status");
|
||||
const counts = workbenchKafkaDebugReplayCounts(await countsNode.textContent());
|
||||
if (status === "idle" && counts?.appliedCount === 0 && counts?.receivedCount === 0) {
|
||||
return { status, ...counts, valuesRedacted: true };
|
||||
}
|
||||
await sleep(100);
|
||||
}
|
||||
throw new Error("Workbench Kafka debug panel did not clear to idle 0/0");
|
||||
}
|
||||
|
||||
async function workbenchKafkaDebugReplayWaitForTerminal(panel, timeoutMs) {
|
||||
const deadline = Date.now() + timeoutMs;
|
||||
while (Date.now() <= deadline) {
|
||||
const status = await panel.getAttribute("data-replay-status");
|
||||
if (status === "completed") return status;
|
||||
if (status === "incomplete" || status === "error" || status === "closed") return status;
|
||||
await sleep(200);
|
||||
}
|
||||
throw new Error("Workbench Kafka debug replay did not reach a terminal panel status within " + timeoutMs + "ms");
|
||||
}
|
||||
|
||||
async function workbenchKafkaDebugReplayPanelSnapshot(panel, countsNode) {
|
||||
const contract = await panel.locator('[data-testid="workbench-isolated-debug-contract"]').evaluate((root) => {
|
||||
const values = {};
|
||||
for (const span of Array.from(root.querySelectorAll("span"))) {
|
||||
const key = span.querySelector("b")?.textContent?.trim() || "";
|
||||
const value = span.querySelector("code")?.textContent?.trim() || "";
|
||||
if (key) values[key] = value;
|
||||
}
|
||||
return { values, text: root.textContent?.replace(/\s+/gu, " ").trim() || "" };
|
||||
});
|
||||
return {
|
||||
status: await panel.getAttribute("data-replay-status"),
|
||||
traceId: contract.values.traceId || null,
|
||||
topic: contract.values.topic || null,
|
||||
groupId: contract.values.group || null,
|
||||
deliveryText: contract.text,
|
||||
counts: workbenchKafkaDebugReplayCounts(await countsNode.textContent()),
|
||||
valuesRedacted: true,
|
||||
};
|
||||
}
|
||||
|
||||
async function workbenchKafkaDebugReplayWriteReport(reportDir, report) {
|
||||
await mkdir(reportDir, { recursive: true, mode: 0o700 });
|
||||
const file = path.join(reportDir, "report.json");
|
||||
await writeFile(file, JSON.stringify(sanitize(report), null, 2) + "\n", { mode: 0o600 });
|
||||
const meta = await fileMeta(file);
|
||||
const artifact = { kind: "workbench-kafka-debug-replay-report", path: file, byteCount: meta.byteCount, sha256: meta.sha256, commandId: activeCommandId, valuesRedacted: true };
|
||||
await appendJsonl(files.artifacts, { ts: new Date().toISOString(), ...artifact });
|
||||
return artifact;
|
||||
}
|
||||
`;
|
||||
}
|
||||
@@ -9,6 +9,7 @@ import { nodeWebObserveRunnerWorkbenchSource } from "./hwlab-node-web-observe-ru
|
||||
import { nodeWebObserveRunnerSamplingSource } from "./hwlab-node-web-observe-runner-sampling-source";
|
||||
import { nodeWebObserveRunnerUtilitySource } from "./hwlab-node-web-observe-runner-utility-source";
|
||||
import { nodeWebObserveRunnerRealtimeSource } from "./hwlab-node-web-observe-runner-realtime-source";
|
||||
import { nodeWebObserveRunnerKafkaDebugReplaySource } from "./hwlab-node-web-observe-runner-kafka-debug-replay-source";
|
||||
|
||||
export function nodeWebObserveRunnerSource(): string {
|
||||
return String.raw`#!/usr/bin/env node
|
||||
@@ -52,6 +53,7 @@ const alertThresholds = parseAlertThresholds(process.env.UNIDESK_WEB_OBSERVE_ALE
|
||||
const browserFreezePolicy = parseBrowserFreezePolicy(process.env.UNIDESK_WEB_OBSERVE_BROWSER_FREEZE_POLICY_JSON);
|
||||
const projectManagement = parseProjectManagementConfig(process.env.UNIDESK_WEB_OBSERVE_PROJECT_MANAGEMENT_JSON);
|
||||
const realtimeFanoutProfiles = parseRealtimeFanoutProfiles(process.env.UNIDESK_WEB_OBSERVE_REALTIME_FANOUT_PROFILES_JSON);
|
||||
const workbenchKafkaDebugReplay = parseWorkbenchKafkaDebugReplayConfig(process.env.UNIDESK_WEB_OBSERVE_WORKBENCH_KAFKA_DEBUG_REPLAY_JSON);
|
||||
const playwrightProxy = proxyConfigFromEnv(baseUrl);
|
||||
const chromiumLaunchOptions = chromiumLaunchOptionsForProxy(playwrightProxy);
|
||||
const pageId = "control-" + randomBytes(4).toString("hex");
|
||||
@@ -197,6 +199,8 @@ ${nodeWebObserveRunnerWorkbenchSource()}
|
||||
|
||||
${nodeWebObserveRunnerRealtimeSource()}
|
||||
|
||||
${nodeWebObserveRunnerKafkaDebugReplaySource()}
|
||||
|
||||
${nodeWebObserveRunnerSamplingSource()}
|
||||
|
||||
${nodeWebObserveRunnerUtilitySource()}
|
||||
|
||||
@@ -139,6 +139,7 @@ export type NodeWebProbeObserveCommandType =
|
||||
| "newSession"
|
||||
| "sendPrompt"
|
||||
| "validateRealtimeFanout"
|
||||
| "validateWorkbenchKafkaDebugReplay"
|
||||
| "steer"
|
||||
| "cancel"
|
||||
| "selectProvider"
|
||||
|
||||
@@ -66,6 +66,48 @@ test("observe status renderer exposes the exact command phase and failed report
|
||||
assert.match(text, /terminal event arrived before subscriber disconnect/u);
|
||||
});
|
||||
|
||||
test("observe status renderer exposes bounded Workbench Kafka debug replay evidence", () => {
|
||||
const rendered = withWebObserveStatusRendered({
|
||||
ok: true,
|
||||
status: "running",
|
||||
command: "web-probe observe status",
|
||||
id: "webobs-fixture",
|
||||
node: "NC01",
|
||||
lane: "v03",
|
||||
observer: {
|
||||
processAlive: true,
|
||||
manifest: { baseUrl: "http://internal.example.test", targetPath: "/workbench" },
|
||||
heartbeat: { status: "running", sampleSeq: 12, commandSeq: 3, updatedAt: "2026-07-10T01:00:01.000Z" },
|
||||
diagnostics: { effectiveLiveness: "alive", heartbeatStale: false },
|
||||
commands: { pendingCount: 0, processingCount: 0, abandonedCount: 0, failedCount: 0 },
|
||||
exactCommand: {
|
||||
commandId: "cmd-debug-replay",
|
||||
type: "validateWorkbenchKafkaDebugReplay",
|
||||
phase: "done",
|
||||
ok: true,
|
||||
result: {
|
||||
status: "passed",
|
||||
traceId: "trc_fixture",
|
||||
topic: "hwlab.event.debug.v1",
|
||||
groupId: "hwlab-v03-workbench-isolated-debug-123-fixture",
|
||||
receivedCount: 35,
|
||||
appliedCount: 35,
|
||||
terminalStatus: "completed",
|
||||
reportSha256: "sha256:report",
|
||||
screenshot: { sha256: "sha256:image" },
|
||||
},
|
||||
},
|
||||
tails: { samples: [], control: [], network: [] },
|
||||
},
|
||||
next: {},
|
||||
});
|
||||
|
||||
const text = String(rendered.renderedText);
|
||||
assert.match(text, /Workbench Kafka debug replay:/u);
|
||||
assert.match(text, /trc_fixture.*hwlab\.event\.debug\.v1/u);
|
||||
assert.match(text, /35.*35.*completed.*sha256:image/u);
|
||||
});
|
||||
|
||||
test("web-probe script render warns to promote repeated scripts into commands", () => {
|
||||
const commandPromotionHint = {
|
||||
schemaVersion: "unidesk.web-probe.command-promotion.v1",
|
||||
|
||||
@@ -68,3 +68,39 @@ test("observe status keeps exact command not-found structured", async () => {
|
||||
valuesRedacted: true,
|
||||
});
|
||||
});
|
||||
|
||||
test("observe status preserves bounded Workbench Kafka debug replay evidence", async () => {
|
||||
const stateDir = await mkdtemp(join(tmpdir(), "unidesk-web-observe-debug-replay-status-"));
|
||||
const commandId = "cmd-debug-replay";
|
||||
await mkdir(join(stateDir, "commands", "done"), { recursive: true });
|
||||
await writeFile(join(stateDir, "commands", "done", `${commandId}.json`), JSON.stringify({
|
||||
ok: true,
|
||||
commandId,
|
||||
type: "validateWorkbenchKafkaDebugReplay",
|
||||
completedAt: "2026-07-10T12:00:00.000Z",
|
||||
result: {
|
||||
ok: true,
|
||||
status: "passed",
|
||||
sessionId: "ses_fixture",
|
||||
traceId: "trc_fixture",
|
||||
topic: "hwlab.event.debug.v1",
|
||||
groupId: "hwlab-v03-workbench-isolated-debug-123-fixture",
|
||||
groupPrefix: "hwlab-v03-workbench-isolated-debug",
|
||||
receivedCount: 35,
|
||||
appliedCount: 35,
|
||||
terminalStatus: "completed",
|
||||
reportPath: "/tmp/report.json",
|
||||
reportSha256: "sha256:report",
|
||||
screenshot: { path: "/tmp/replay.png", sha256: "sha256:image" },
|
||||
},
|
||||
}) + "\n");
|
||||
|
||||
const status = await runStatusScript(stateDir, commandId);
|
||||
assert.equal(status.exactCommand.type, "validateWorkbenchKafkaDebugReplay");
|
||||
assert.equal(status.exactCommand.result.traceId, "trc_fixture");
|
||||
assert.equal(status.exactCommand.result.topic, "hwlab.event.debug.v1");
|
||||
assert.equal(status.exactCommand.result.groupPrefix, "hwlab-v03-workbench-isolated-debug");
|
||||
assert.equal(status.exactCommand.result.receivedCount, 35);
|
||||
assert.equal(status.exactCommand.result.appliedCount, 35);
|
||||
assert.equal(status.exactCommand.result.screenshot.sha256, "sha256:image");
|
||||
});
|
||||
|
||||
@@ -74,7 +74,7 @@ export function nodeWebObserveStatusNodeScript(tailLines: number, node: string,
|
||||
if(!parsed)continue;
|
||||
const result=parsed.result&&typeof parsed.result==='object'?parsed.result:null;
|
||||
const error=parsed.error&&typeof parsed.error==='object'?parsed.error:null;
|
||||
return {commandId:exactCommandId,phase:bucket,ok:parsed.ok??null,type:parsed.type||null,createdAt:parsed.createdAt||null,completedAt:parsed.completedAt||null,failedAt:parsed.failedAt||null,result:result?{ok:result.ok??null,status:result.status||null,profile:result.profile||null,sessionId:result.sessionId||null,traceIds:Array.isArray(result.traceIds)?result.traceIds.slice(0,4):[],runIds:Array.isArray(result.runIds)?result.runIds.slice(0,4):[],commandIds:Array.isArray(result.commandIds)?result.commandIds.slice(0,4):[],warmRunnerReused:result.warmRunnerReused??null,forbiddenRequestCount:result.forbiddenRequestCount??null,replayedEventCount:result.replayedEventCount??null,reportPath:result.reportPath||null,reportSha256:result.reportSha256||null,valuesRedacted:true}:null,error:error?{name:error.name||null,message:short(error.message,200),details:error.details&&typeof error.details==='object'?{profile:error.details.profile||null,reportPath:error.details.reportPath||null,reportSha256:error.details.reportSha256||null,valuesRedacted:true}:null}:null,valuesRedacted:true};
|
||||
return {commandId:exactCommandId,phase:bucket,ok:parsed.ok??null,type:parsed.type||null,createdAt:parsed.createdAt||null,completedAt:parsed.completedAt||null,failedAt:parsed.failedAt||null,result:result?{ok:result.ok??null,status:result.status||null,profile:result.profile||null,sessionId:result.sessionId||null,traceId:result.traceId||null,traceIds:Array.isArray(result.traceIds)?result.traceIds.slice(0,4):[],runIds:Array.isArray(result.runIds)?result.runIds.slice(0,4):[],commandIds:Array.isArray(result.commandIds)?result.commandIds.slice(0,4):[],topic:result.topic||null,groupId:result.groupId||null,groupPrefix:result.groupPrefix||null,receivedCount:result.receivedCount??null,appliedCount:result.appliedCount??null,terminalStatus:result.terminalStatus||null,warmRunnerReused:result.warmRunnerReused??null,forbiddenRequestCount:result.forbiddenRequestCount??null,replayedEventCount:result.replayedEventCount??null,reportPath:result.reportPath||null,reportSha256:result.reportSha256||null,screenshot:result.screenshot&&typeof result.screenshot==='object'?{path:result.screenshot.path||null,sha256:result.screenshot.sha256||null,valuesRedacted:true}:null,valuesRedacted:true}:null,error:error?{name:error.name||null,message:short(error.message,200),details:error.details&&typeof error.details==='object'?{profile:error.details.profile||null,reportPath:error.details.reportPath||null,reportSha256:error.details.reportSha256||null,valuesRedacted:true}:null}:null,valuesRedacted:true};
|
||||
}
|
||||
return {commandId:exactCommandId,phase:'not-found',ok:false,valuesRedacted:true};
|
||||
};
|
||||
|
||||
@@ -5,7 +5,7 @@ import { join } from "node:path";
|
||||
import { test } from "bun:test";
|
||||
|
||||
import { hwlabRuntimeLaneSpecForNode } from "../hwlab-node-lanes";
|
||||
import { buildWebObserveCommandVisibility, nodeWebProbeRealtimeFanoutProfiles, resolveWebObserveActionJson } from "./web-probe-observe-actions";
|
||||
import { buildWebObserveCommandVisibility, nodeWebProbeRealtimeFanoutProfiles, nodeWebProbeWorkbenchKafkaDebugReplay, resolveWebObserveActionJson } from "./web-probe-observe-actions";
|
||||
|
||||
test("realtime fanout runner contract derives topics, groups, and independent capabilities from owning YAML", () => {
|
||||
const profiles = nodeWebProbeRealtimeFanoutProfiles(hwlabRuntimeLaneSpecForNode("v03", "NC01"));
|
||||
@@ -29,6 +29,16 @@ test("realtime fanout runner contract derives topics, groups, and independent ca
|
||||
});
|
||||
});
|
||||
|
||||
test("Workbench Kafka debug replay runner contract is derived from owning YAML", () => {
|
||||
assert.deepEqual(nodeWebProbeWorkbenchKafkaDebugReplay(hwlabRuntimeLaneSpecForNode("v03", "NC01")), {
|
||||
enabled: true,
|
||||
topic: "hwlab.event.debug.v1",
|
||||
groupPrefix: "hwlab-v03-workbench-isolated-debug",
|
||||
completionTimeoutMs: 30_000,
|
||||
valuesRedacted: true,
|
||||
});
|
||||
});
|
||||
|
||||
test("web observe command visibility does not equate control completion with async turn completion", () => {
|
||||
const visibility = buildWebObserveCommandVisibility({
|
||||
commandType: "sendPrompt",
|
||||
|
||||
@@ -48,6 +48,12 @@ export function nodeWebProbeRealtimeFanoutProfiles(spec: HwlabRuntimeLaneSpec):
|
||||
}));
|
||||
}
|
||||
|
||||
export function nodeWebProbeWorkbenchKafkaDebugReplay(spec: HwlabRuntimeLaneSpec): Record<string, unknown> | null {
|
||||
const config = spec.webProbe?.workbenchKafkaDebugReplay;
|
||||
if (config === undefined) return null;
|
||||
return { ...config, valuesRedacted: true };
|
||||
}
|
||||
|
||||
export function resolveWebObserveActionJson(
|
||||
result: { stdout: string; stderr?: string; exitCode?: number | null; timedOut?: boolean },
|
||||
contract: WebObserveActionJsonContract,
|
||||
@@ -475,6 +481,7 @@ export function runNodeWebProbeObserveStart(
|
||||
`UNIDESK_WEB_OBSERVE_BROWSER_FREEZE_POLICY_JSON=${shellQuote(JSON.stringify(browserFreezePolicy))}`,
|
||||
`UNIDESK_WEB_OBSERVE_PROJECT_MANAGEMENT_JSON=${shellQuote(JSON.stringify(projectManagement))}`,
|
||||
`UNIDESK_WEB_OBSERVE_REALTIME_FANOUT_PROFILES_JSON=${shellQuote(JSON.stringify(nodeWebProbeRealtimeFanoutProfiles(spec)))}`,
|
||||
`UNIDESK_WEB_OBSERVE_WORKBENCH_KAFKA_DEBUG_REPLAY_JSON=${shellQuote(JSON.stringify(nodeWebProbeWorkbenchKafkaDebugReplay(spec)))}`,
|
||||
...(authLogin === null
|
||||
? []
|
||||
: [
|
||||
@@ -726,6 +733,9 @@ export function runNodeWebProbeObserveCommand(options: NodeWebProbeObserveOption
|
||||
if (!options.commandText?.trim()) throw new Error("validateRealtimeFanout requires --text or --text-stdin for the first turn");
|
||||
if (!options.commandSecondText?.trim()) throw new Error("validateRealtimeFanout requires --second-text for the follow-up turn");
|
||||
}
|
||||
if (type === "validateWorkbenchKafkaDebugReplay" && spec.webProbe?.workbenchKafkaDebugReplay?.enabled !== true) {
|
||||
throw new Error("validateWorkbenchKafkaDebugReplay is not enabled by the owning YAML");
|
||||
}
|
||||
const commandId = `cmd-${Date.now().toString(36)}-${randomBytes(3).toString("hex")}`;
|
||||
const payload = {
|
||||
id: commandId,
|
||||
|
||||
@@ -58,6 +58,25 @@ test("validateRealtimeFanout rejects profiles absent from the owning YAML", () =
|
||||
);
|
||||
});
|
||||
|
||||
test("validateWorkbenchKafkaDebugReplay is a YAML-enabled argument-free typed command", () => {
|
||||
const options = parseNodeWebProbeObserveOptions(
|
||||
["command", "--type", "validateWorkbenchKafkaDebugReplay"],
|
||||
"NC01",
|
||||
"v03",
|
||||
spec,
|
||||
"webobs-fixture",
|
||||
null,
|
||||
);
|
||||
|
||||
assert.equal(options.commandType, "validateWorkbenchKafkaDebugReplay");
|
||||
assert.deepEqual(spec.webProbe?.workbenchKafkaDebugReplay, {
|
||||
enabled: true,
|
||||
topic: "hwlab.event.debug.v1",
|
||||
groupPrefix: "hwlab-v03-workbench-isolated-debug",
|
||||
completionTimeoutMs: 30_000,
|
||||
});
|
||||
});
|
||||
|
||||
test("observe start selects the YAML public origin semantically", () => {
|
||||
const options = parseNodeWebProbeObserveOptions(
|
||||
["start", "--origin", "public"],
|
||||
|
||||
@@ -600,6 +600,11 @@ export function parseNodeWebProbeObserveOptions(
|
||||
if (!commandText?.trim()) throw new Error("validateRealtimeFanout requires --text or --text-stdin for the first turn");
|
||||
if (!commandSecondText?.trim()) throw new Error("validateRealtimeFanout requires --second-text for the follow-up turn");
|
||||
}
|
||||
if (observeActionRaw === "command" && commandType === "validateWorkbenchKafkaDebugReplay") {
|
||||
if (spec.webProbe?.workbenchKafkaDebugReplay?.enabled !== true) {
|
||||
throw new Error("validateWorkbenchKafkaDebugReplay is not enabled by the owning YAML");
|
||||
}
|
||||
}
|
||||
return {
|
||||
action: "observe",
|
||||
observeAction: observeActionRaw,
|
||||
@@ -695,6 +700,7 @@ export function parseNodeWebProbeObserveCommandType(value: string): NodeWebProbe
|
||||
|| value === "newSession"
|
||||
|| value === "sendPrompt"
|
||||
|| value === "validateRealtimeFanout"
|
||||
|| value === "validateWorkbenchKafkaDebugReplay"
|
||||
|| value === "steer"
|
||||
|| value === "cancel"
|
||||
|| value === "selectProvider"
|
||||
@@ -730,7 +736,7 @@ export function parseNodeWebProbeObserveCommandType(value: string): NodeWebProbe
|
||||
|| value === "mark"
|
||||
|| value === "stop"
|
||||
) return value;
|
||||
throw new Error(`web-probe observe command --type must be login, loginAccount, logout, listSessions, switchSessions, preflight, goto, gotoProjectMdtodo, newSession, sendPrompt, validateRealtimeFanout, steer, cancel, selectProvider, clickSession, refreshCurrentSession, switchAwayAndBack, assertSessionInvariant, selectProjectSource, selectMdtodoSource, selectMdtodoFile, selectMdtodoTask, expandMdtodoTask, openMdtodoReportPreview, toggleMdtodoReportFullscreen, openMdtodoSourceConfig, closeMdtodoSourceConfig, configureMdtodoHwpodSource, probeMdtodoSource, reindexMdtodoSource, editMdtodoTaskInline, editMdtodoTaskTitle, editMdtodoTaskBody, toggleMdtodoTaskStatus, addMdtodoRootTask, addMdtodoSubTask, continueMdtodoTask, deleteMdtodoTask, launchWorkbenchFromTask, launchWorkbenchFromMdtodo, performanceCapture, screenshot, mark, or stop; got ${value}`);
|
||||
throw new Error(`web-probe observe command --type must be login, loginAccount, logout, listSessions, switchSessions, preflight, goto, gotoProjectMdtodo, newSession, sendPrompt, validateRealtimeFanout, validateWorkbenchKafkaDebugReplay, steer, cancel, selectProvider, clickSession, refreshCurrentSession, switchAwayAndBack, assertSessionInvariant, selectProjectSource, selectMdtodoSource, selectMdtodoFile, selectMdtodoTask, expandMdtodoTask, openMdtodoReportPreview, toggleMdtodoReportFullscreen, openMdtodoSourceConfig, closeMdtodoSourceConfig, configureMdtodoHwpodSource, probeMdtodoSource, reindexMdtodoSource, editMdtodoTaskInline, editMdtodoTaskTitle, editMdtodoTaskBody, toggleMdtodoTaskStatus, addMdtodoRootTask, addMdtodoSubTask, continueMdtodoTask, deleteMdtodoTask, launchWorkbenchFromTask, launchWorkbenchFromMdtodo, performanceCapture, screenshot, mark, or stop; got ${value}`);
|
||||
}
|
||||
|
||||
export function parseWebProbeBrowserProxyMode(value: string | undefined): WebProbeBrowserProxyMode {
|
||||
|
||||
Reference in New Issue
Block a user