pikasTech-unidesk/scripts/src/hwlab-node-web-observe-runner-kafka-debug-replay-source.ts

// 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;
}
`;
}