// 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 productEventsPath = String(parsed.productEventsPath || "").trim();
  const debugEventsPath = String(parsed.debugEventsPath || "").trim();
  const completionTimeoutMs = Number(parsed.completionTimeoutMs);
  const requestObservationQuietMs = Number(parsed.requestObservationQuietMs);
  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 (!productEventsPath.startsWith("/") || productEventsPath.includes("?") || productEventsPath.includes("#")) {
    throw new Error("Workbench Kafka debug replay productEventsPath must be an absolute path without query or fragment in the owning YAML");
  }
  if (!debugEventsPath.startsWith("/") || debugEventsPath.includes("?") || debugEventsPath.includes("#")) {
    throw new Error("Workbench Kafka debug replay debugEventsPath must be an absolute path without query or fragment 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");
  }
  if (!Number.isInteger(requestObservationQuietMs) || requestObservationQuietMs < 100 || requestObservationQuietMs > 30000) {
    throw new Error("Workbench Kafka debug replay requestObservationQuietMs must be 100-30000 in the owning YAML");
  }
  return {
    enabled: parsed.enabled === true,
    topic,
    groupPrefix,
    productEventsPath,
    debugEventsPath,
    completionTimeoutMs,
    requestObservationQuietMs,
    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;
}

function workbenchKafkaDebugReplayNamedCount(value, name) {
  const match = String(value || "").match(new RegExp("(?:^|\\s|·)" + name + "\\s*=\\s*(-|\\d+)", "iu"));
  if (!match) return undefined;
  return match[1] === "-" ? null : Number(match[1]);
}

function workbenchKafkaDebugReplayLayerCounts(value) {
  const text = String(value || "").replace(/\s+/gu, " ").trim();
  if (!/\bserver\b/iu.test(text) || !/\bclient\b/iu.test(text)) return null;
  return {
    server: {
      scanned: workbenchKafkaDebugReplayNamedCount(text, "scanned"),
      matched: workbenchKafkaDebugReplayNamedCount(text, "matched"),
      delivered: workbenchKafkaDebugReplayNamedCount(text, "delivered")
    },
    client: {
      received: workbenchKafkaDebugReplayNamedCount(text, "received"),
      decoded: workbenchKafkaDebugReplayNamedCount(text, "decoded"),
      applied: workbenchKafkaDebugReplayNamedCount(text, "applied")
    },
    valuesRedacted: true
  };
}

function workbenchKafkaDebugReplayRawCounts(value) {
  const text = String(value || "").replace(/\s+/gu, " ").trim();
  if (!/\breceived\s*=/iu.test(text) || !/\bretained\s*=/iu.test(text)) return null;
  return {
    received: workbenchKafkaDebugReplayNamedCount(text, "received"),
    retained: workbenchKafkaDebugReplayNamedCount(text, "retained"),
    decoded: workbenchKafkaDebugReplayNamedCount(text, "decoded"),
    rejected: workbenchKafkaDebugReplayNamedCount(text, "rejected"),
    evicted: workbenchKafkaDebugReplayNamedCount(text, "evicted"),
    bytes: workbenchKafkaDebugReplayNamedCount(text, "bytes"),
    valuesRedacted: true
  };
}

function workbenchKafkaDebugReplayTypedOutcome(input) {
  const status = String(input?.status || "unknown").trim() || "unknown";
  const errorText = String(input?.errorText || "").trim();
  const source = input?.layerCounts ? "typed-dom-v2" : "legacy-dom";
  const codeMatch = errorText.match(/\b(consumer_not_ready|source_trace_missing|producer_not_invoked|topic_no_append_for_replay|record_parse_rejected|records_scanned_but_filter_mismatch|sse_write_failed|transport_failed|decoder_rejected|reducer_rejected|terminal_missing)\b/u);
  const phaseByCode = {
    consumer_not_ready: "consumer",
    source_trace_missing: "producer",
    producer_not_invoked: "producer",
    topic_no_append_for_replay: "kafka",
    record_parse_rejected: "server",
    records_scanned_but_filter_mismatch: "server",
    sse_write_failed: "server",
    transport_failed: "client",
    decoder_rejected: "client",
    reducer_rejected: "client",
    terminal_missing: "terminal"
  };
  if (codeMatch) return { phase: phaseByCode[codeMatch[1]], code: codeMatch[1], source, valuesRedacted: true };
  if (status === "completed") return { phase: "terminal", code: "terminal_complete", source, valuesRedacted: true };
  if (status === "incomplete" && /terminal|终态|未观测/iu.test(errorText)) {
    return { phase: "terminal", code: "terminal_missing", source, valuesRedacted: true };
  }
  if (status === "incomplete") return { phase: "ui", code: "replay_incomplete", source, valuesRedacted: true };
  if (status === "error") return { phase: "ui", code: "replay_error", source, valuesRedacted: true };
  if (status === "closed") return { phase: "client", code: "stream_closed", source, valuesRedacted: true };
  return { phase: "ui", code: "replay_" + status.replace(/[^a-z0-9_-]+/giu, "_"), source, valuesRedacted: true };
}

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-v2",
    commandId: command.id,
    expected: {
      topic: config.topic,
      groupPrefix: config.groupPrefix,
      productEventsPath: config.productEventsPath,
      debugEventsPath: config.debugEventsPath,
      completionTimeoutMs: config.completionTimeoutMs,
      requestObservationQuietMs: config.requestObservationQuietMs,
      valuesRedacted: true
    },
    origin,
    startedAt: new Date().toISOString(),
    valuesRedacted: true
  };
  const requests = [];
  const productEventSourceRequests = [];
  let requestPage = null;
  let requestPhase = "setup";
  const requestListener = (request) => {
    let url;
    try { url = new URL(request.url()); } catch { return; }
    if (url.pathname === config.productEventsPath) {
      productEventSourceRequests.push({ method: request.method(), path: url.pathname, phase: requestPhase, valuesRedacted: true });
      return;
    }
    if (url.pathname !== config.debugEventsPath) return;
    requests.push({
      method: request.method(),
      path: url.pathname,
      stream: url.searchParams.get("stream"),
      traceId: url.searchParams.get("traceId"),
      fromBeginning: url.searchParams.get("fromBeginning"),
      phase: requestPhase,
      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");

    requestPage = page;
    requestPage.on("request", requestListener);
    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);

    requestPhase = "debug-replay";
    await replayButton.click({ timeout: 10000 });
    const terminalStatus = await workbenchKafkaDebugReplayWaitForTerminal(panel, config.completionTimeoutMs);
    const after = await workbenchKafkaDebugReplayWaitForStablePanel(
      panel,
      countsNode,
      config.requestObservationQuietMs
    );
    Object.assign(report, {
      sessionId,
      traceId: after.traceId,
      topic: after.topic,
      groupId: after.groupId,
      terminalStatus,
      phase: after.typedOutcome.phase,
      code: after.typedOutcome.code,
      typedOutcome: after.typedOutcome,
      layerCounts: after.layerCounts,
      layerDom: after.layerDom,
      layerDomStabilization: after.layerDomStabilization,
      legacyCounts: after.counts,
      valuesRedacted: true
    });
    const traceTimeline = await workbenchKafkaDebugReplayTraceTimelineSnapshot(panel);
    report.traceTimeline = traceTimeline;
    const rawHwlabEventWindow = await workbenchKafkaDebugReplayInspectRawWindow({
      panel,
      config,
      pathBefore: currentPath,
      productEventSourceRequests,
      setRequestPhase: (value) => { requestPhase = value; }
    });
    report.rawHwlabEventWindow = rawHwlabEventWindow;
    const matchingRequest = requests.find((item) => item.stream === "hwlab-debug" && item.traceId === before.traceId && item.fromBeginning === "true") || null;
    const replayEvidence = {
      sessionId,
      traceId: after.traceId,
      topic: after.topic,
      groupId: after.groupId,
      terminalStatus,
      phase: after.typedOutcome.phase,
      code: after.typedOutcome.code,
      typedOutcome: after.typedOutcome,
      layerCounts: after.layerCounts,
      layerDom: after.layerDom,
      layerDomStabilization: after.layerDomStabilization,
      legacyCounts: after.counts,
      traceTimeline,
      rawHwlabEventWindow,
      productEventSource: {
        path: config.productEventsPath,
        additionalRequestCountDuringTabSwitch: rawHwlabEventWindow.productEventSourceRequestCount,
        requests: productEventSourceRequests.slice(-10),
        valuesRedacted: true
      },
      valuesRedacted: true
    };
    Object.assign(report, replayEvidence);

    if (terminalStatus !== "completed") throw workbenchKafkaDebugReplayEvidenceError("Workbench Kafka debug replay ended with status " + terminalStatus, replayEvidence);
    if (!matchingRequest) throw workbenchKafkaDebugReplayEvidenceError("Workbench Kafka debug replay did not issue the expected trace-scoped fromBeginning request", replayEvidence);
    if (after.traceId !== before.traceId) throw workbenchKafkaDebugReplayEvidenceError("Workbench Kafka debug replay traceId changed during replay", replayEvidence);
    if (after.topic !== config.topic) throw workbenchKafkaDebugReplayEvidenceError("Workbench Kafka debug replay topic mismatch: " + (after.topic || "missing"), replayEvidence);
    if (!after.groupId || !after.groupId.startsWith(config.groupPrefix + "-")) {
      throw workbenchKafkaDebugReplayEvidenceError("Workbench Kafka debug replay groupId does not use the YAML group prefix", replayEvidence);
    }
    if (!after.deliveryText.includes("debug-replay") || !after.deliveryText.includes("fromBeginning=true")) {
      throw workbenchKafkaDebugReplayEvidenceError("Workbench Kafka debug replay delivery contract is missing", replayEvidence);
    }
    if (!after.counts || after.counts.receivedCount <= 0 || after.counts.appliedCount <= 0) {
      throw workbenchKafkaDebugReplayEvidenceError("Workbench Kafka debug replay received/applied counts must both be greater than zero", replayEvidence);
    }
    if (after.typedOutcome.phase !== "terminal" || after.typedOutcome.code !== "terminal_complete") {
      throw workbenchKafkaDebugReplayEvidenceError("Workbench Kafka debug replay typed terminal contract is not terminal_complete", replayEvidence);
    }
    if (after.layerDomStabilization?.stable !== true) {
      throw workbenchKafkaDebugReplayEvidenceError("Workbench Kafka debug replay layer DOM did not stabilize after terminal status", replayEvidence);
    }
    workbenchKafkaDebugReplayValidateLayerCounts(after.layerCounts, replayEvidence);
    workbenchKafkaDebugReplayValidateTraceTimeline(traceTimeline, replayEvidence);
    workbenchKafkaDebugReplayValidateRawWindow(rawHwlabEventWindow, replayEvidence);

    const screenshot = await captureScreenshot("workbench-kafka-debug-replay-" + safeId(command.id));
    Object.assign(report, {
      ok: true,
      status: "passed",
      completedAt: new Date().toISOString(),
      groupPrefix: config.groupPrefix,
      receivedCount: after.counts.receivedCount,
      appliedCount: after.counts.appliedCount,
      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,
      phase: report.phase,
      code: report.code,
      receivedCount: report.receivedCount,
      appliedCount: report.appliedCount,
      layerCounts: report.layerCounts,
      layerDom: report.layerDom,
      layerDomStabilization: report.layerDomStabilization,
      traceTimeline: report.traceTimeline,
      rawHwlabEventWindow: report.rawHwlabEventWindow,
      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,
      phase: report.phase,
      code: report.code,
      receivedCount: report.receivedCount,
      appliedCount: report.appliedCount,
      layerCounts: report.layerCounts,
      layerDom: report.layerDom,
      layerDomStabilization: report.layerDomStabilization,
      traceTimeline: report.traceTimeline,
      rawHwlabEventWindow: report.rawHwlabEventWindow,
      productEventSource: report.productEventSource,
      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
    });
    if (!report.screenshot) {
      report.screenshot = await captureScreenshot("workbench-kafka-debug-replay-failed-" + safeId(command.id))
        .then((screenshot) => ({ path: screenshot.path, sha256: screenshot.sha256, valuesRedacted: true }))
        .catch((screenshotError) => ({ available: false, error: errorSummary(screenshotError), valuesRedacted: true }));
    }
    const reportArtifact = await workbenchKafkaDebugReplayWriteReport(reportDir, report).catch(() => null);
    const wrapped = error instanceof Error ? error : new Error(String(error));
    wrapped.details = {
      ...(wrapped.details || {}),
      traceId: report.traceId || wrapped.details?.traceId || null,
      topic: report.topic || wrapped.details?.topic || null,
      groupId: report.groupId || wrapped.details?.groupId || null,
      terminalStatus: report.terminalStatus || wrapped.details?.terminalStatus || null,
      receivedCount: report.receivedCount ?? report.legacyCounts?.receivedCount ?? wrapped.details?.receivedCount ?? null,
      appliedCount: report.appliedCount ?? report.legacyCounts?.appliedCount ?? wrapped.details?.appliedCount ?? null,
      phase: report.phase || wrapped.details?.phase || null,
      code: report.code || wrapped.details?.code || null,
      layerCounts: report.layerCounts || wrapped.details?.layerCounts || null,
      layerDom: report.layerDom || wrapped.details?.layerDom || null,
      layerDomStabilization: report.layerDomStabilization || wrapped.details?.layerDomStabilization || null,
      traceTimeline: report.traceTimeline || wrapped.details?.traceTimeline || null,
      rawHwlabEventWindow: report.rawHwlabEventWindow || wrapped.details?.rawHwlabEventWindow || null,
      screenshot: report.screenshot || wrapped.details?.screenshot || null,
      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 workbenchKafkaDebugReplayWaitForTabSelected(tab, timeoutMs) {
  const deadline = Date.now() + timeoutMs;
  while (Date.now() <= deadline) {
    if (await tab.getAttribute("aria-selected") === "true") return true;
    await sleep(50);
  }
  return false;
}

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() || "" };
  });
  const status = await panel.getAttribute("data-replay-status");
  const layerDom = await workbenchKafkaDebugReplayLayerDomSnapshot(panel);
  const layerText = layerDom.text || "";
  const errorNode = panel.locator(".isolated-debug-error").first();
  const errorText = await errorNode.count() > 0 ? String(await errorNode.textContent() || "").trim() : "";
  const layerCounts = workbenchKafkaDebugReplayLayerCounts(layerText);
  return {
    status,
    traceId: contract.values.traceId || null,
    topic: contract.values.topic || null,
    groupId: contract.values.group || null,
    deliveryText: contract.text,
    counts: workbenchKafkaDebugReplayCounts(await countsNode.textContent()),
    layerCounts,
    layerDom,
    error: errorText ? { textHash: sha256Text(errorText), textPreview: truncate(errorText, 160), textBytes: Buffer.byteLength(errorText), valuesRedacted: true } : null,
    typedOutcome: workbenchKafkaDebugReplayTypedOutcome({ status, errorText, layerCounts }),
    valuesRedacted: true
  };
}

async function workbenchKafkaDebugReplayLayerDomSnapshot(panel) {
  const selector = '[data-testid="workbench-kafka-debug-layer-counts"]';
  const nodes = await panel.locator(selector).evaluateAll((elements) => elements.map((element) => {
    const style = getComputedStyle(element);
    const rect = element.getBoundingClientRect();
    const text = element.textContent?.replace(/\s+/gu, " ").trim() || "";
    return {
      connected: element.isConnected,
      visible: style.display !== "none" && style.visibility !== "hidden" && rect.width > 0 && rect.height > 0,
      text
    };
  }));
  const selectedIndex = nodes.findIndex((node) => node.visible);
  const selected = nodes[selectedIndex >= 0 ? selectedIndex : 0] || null;
  const text = selected?.text || "";
  return {
    selector,
    matchCount: nodes.length,
    visibleCount: nodes.filter((node) => node.visible).length,
    connectedCount: nodes.filter((node) => node.connected).length,
    selectedIndex: selectedIndex >= 0 ? selectedIndex : nodes.length > 0 ? 0 : null,
    text,
    textPresent: text.length > 0,
    textBytes: Buffer.byteLength(text),
    textHash: sha256Text(text),
    textPreview: truncate(text, 240),
    valuesRedacted: true
  };
}

async function workbenchKafkaDebugReplayWaitForStablePanel(panel, countsNode, timeoutMs) {
  const startedAtMs = Date.now();
  const deadline = startedAtMs + timeoutMs;
  const stableQuietMs = Math.min(500, Math.max(100, Math.floor(timeoutMs / 4)));
  let latest = await workbenchKafkaDebugReplayPanelSnapshot(panel, countsNode);
  let previousSignature = null;
  let stableSinceMs = null;
  let sampleCount = 0;
  while (Date.now() <= deadline) {
    latest = await workbenchKafkaDebugReplayPanelSnapshot(panel, countsNode);
    sampleCount += 1;
    const signature = JSON.stringify({
      status: latest.status,
      counts: latest.counts,
      layerCounts: latest.layerCounts,
      layerTextHash: latest.layerDom?.textHash,
      matchCount: latest.layerDom?.matchCount,
      visibleCount: latest.layerDom?.visibleCount
    });
    const domReady = latest.layerDom?.matchCount === 1
      && latest.layerDom?.visibleCount === 1
      && latest.layerDom?.textPresent === true
      && latest.layerCounts !== null;
    if (domReady && signature === previousSignature) {
      stableSinceMs ??= Date.now();
      if (Date.now() - stableSinceMs >= stableQuietMs) {
        return {
          ...latest,
          layerDomStabilization: {
            stable: true,
            timedOut: false,
            sampleCount,
            elapsedMs: Date.now() - startedAtMs,
            stableQuietMs,
            valuesRedacted: true
          }
        };
      }
    } else {
      stableSinceMs = null;
    }
    previousSignature = signature;
    await sleep(100);
  }
  return {
    ...latest,
    layerDomStabilization: {
      stable: false,
      timedOut: true,
      sampleCount,
      elapsedMs: Date.now() - startedAtMs,
      stableQuietMs,
      valuesRedacted: true
    }
  };
}

async function workbenchKafkaDebugReplayTraceTimelineSnapshot(panel) {
  return panel.evaluate((root) => {
    const rows = Array.from(root.querySelectorAll('[data-testid="trace-render-row"]'));
    const authorityRows = rows.filter((row) => row.hasAttribute("data-event-seq-authority"));
    const sourceRows = authorityRows.filter((row) => row.getAttribute("data-event-seq-authority") === "source");
    const projectedRows = authorityRows.filter((row) => row.getAttribute("data-event-seq-authority") === "projected");
    const readableSourceRows = sourceRows.filter((row) => Boolean(row.textContent?.replace(/\s+/gu, " ").trim()));
    return {
      contract: authorityRows.length > 0 ? "trace-sequence-authority-v1" : "legacy-ui",
      authorityAvailable: authorityRows.length > 0,
      rowCount: rows.length,
      sourceAuthorityRowCount: sourceRows.length,
      projectedAuthorityRowCount: projectedRows.length,
      readableSourceRowCount: readableSourceRows.length,
      samples: readableSourceRows.slice(0, 8).map((row) => ({
        seq: row.getAttribute("data-event-seq"),
        authority: row.getAttribute("data-event-seq-authority"),
        kind: row.getAttribute("data-event-kind"),
        status: row.getAttribute("data-event-status"),
        terminal: row.getAttribute("data-terminal") === "true",
        textBytes: new TextEncoder().encode(row.textContent?.replace(/\s+/gu, " ").trim() || "").byteLength
      })),
      valuesRedacted: true
    };
  });
}

async function workbenchKafkaDebugReplayInspectRawWindow(input) {
  const rawTabLocator = input.panel.locator('[data-testid="workbench-kafka-debug-tab-raw"]');
  const rawTabCount = await rawTabLocator.count();
  if (rawTabCount === 0) {
    return {
      available: false,
      contract: "legacy-ui",
      reason: "raw-hwlab-event-tab-not-present",
      productEventsPath: input.config.productEventsPath,
      productEventSourceRequestCount: 0,
      valuesRedacted: true
    };
  }
  if (rawTabCount !== 1) throw new Error("Workbench raw HWLAB Event tab count is " + rawTabCount + ", expected 1");
  const traceTabLocator = input.panel.locator('[data-testid="workbench-kafka-debug-tab-trace"]');
  const traceTabCount = await traceTabLocator.count();
  if (traceTabCount !== 1) throw new Error("Workbench Trace debug tab count is " + traceTabCount + ", expected 1");
  const rawTab = rawTabLocator.first();
  const traceTab = traceTabLocator.first();
  const requestStart = input.productEventSourceRequests.length;
  const pageUrlBefore = currentPageUrl();
  input.setRequestPhase("raw-tab-switch");
  await rawTab.click({ timeout: 10000 });
  if (!await workbenchKafkaDebugReplayWaitForTabSelected(rawTab, 5000)) throw new Error("Workbench raw HWLAB Event tab did not become selected");
  const rawCountsNode = input.panel.locator('[data-testid="workbench-raw-hwlab-counts"]').first();
  const rawTextNode = input.panel.locator('[data-testid="workbench-raw-hwlab-text"]').first();
  const rawContractNode = input.panel.locator('[data-testid="workbench-raw-hwlab-contract"]').first();
  await rawCountsNode.waitFor({ state: "visible", timeout: 5000 });
  await rawTextNode.waitFor({ state: "visible", timeout: 5000 });
  await rawContractNode.waitFor({ state: "visible", timeout: 5000 });
  await sleep(input.config.requestObservationQuietMs);
  const counts = workbenchKafkaDebugReplayRawCounts(await rawCountsNode.textContent());
  const rawText = await rawTextNode.inputValue();
  const contractText = String(await rawContractNode.textContent() || "").replace(/\s+/gu, " ").trim();
  input.setRequestPhase("trace-tab-restore");
  await traceTab.click({ timeout: 10000 });
  if (!await workbenchKafkaDebugReplayWaitForTabSelected(traceTab, 5000)) throw new Error("Workbench Trace debug tab did not restore after raw inspection");
  await sleep(input.config.requestObservationQuietMs);
  input.setRequestPhase("complete");
  const pageUrlAfter = currentPageUrl();
  const productRequests = input.productEventSourceRequests.slice(requestStart);
  return {
    available: true,
    contract: "raw-hwlab-event-window-v1",
    pageId,
    samePage: safeUrlPath(pageUrlBefore) === input.pathBefore && safeUrlPath(pageUrlAfter) === input.pathBefore,
    pathBefore: input.pathBefore,
    pathAfter: safeUrlPath(pageUrlAfter),
    productEventsPath: input.config.productEventsPath,
    productEventSourceRequestCount: productRequests.length,
    productEventSourceRequests: productRequests.slice(-10),
    sourceContractPresent: contractText.includes("EventSource") && contractText.includes("hwlab.event.v1"),
    unfilteredContractPresent: contractText.includes("no client event filter"),
    counts,
    textBytes: Buffer.byteLength(rawText),
    textHash: sha256Text(rawText),
    textPresent: rawText.length > 0,
    valuesRedacted: true
  };
}

function workbenchKafkaDebugReplayValidateLayerCounts(layerCounts, evidence) {
  if (!layerCounts) {
    throw workbenchKafkaDebugReplayEvidenceError("Workbench Kafka debug replay typed server/client count contract is missing", evidence);
  }
  const values = [
    layerCounts.server.scanned,
    layerCounts.server.matched,
    layerCounts.server.delivered,
    layerCounts.client.received,
    layerCounts.client.decoded,
    layerCounts.client.applied
  ];
  if (values.some((value) => !Number.isInteger(value) || value <= 0)) {
    throw workbenchKafkaDebugReplayEvidenceError("Workbench Kafka debug replay typed server/client counts must all be greater than zero", evidence);
  }
  if (layerCounts.server.delivered !== layerCounts.client.received) {
    throw workbenchKafkaDebugReplayEvidenceError("Workbench Kafka debug replay server delivered and client received counts differ", evidence);
  }
  if (layerCounts.server.scanned < layerCounts.server.matched || layerCounts.server.matched !== layerCounts.server.delivered) {
    throw workbenchKafkaDebugReplayEvidenceError("Workbench Kafka debug replay server scanned/matched/delivered counts are inconsistent", evidence);
  }
  if (layerCounts.client.received !== layerCounts.client.decoded || layerCounts.client.decoded !== layerCounts.client.applied) {
    throw workbenchKafkaDebugReplayEvidenceError("Workbench Kafka debug replay client received/decoded/applied counts differ", evidence);
  }
  if (evidence?.legacyCounts && (evidence.legacyCounts.receivedCount !== layerCounts.client.received || evidence.legacyCounts.appliedCount !== layerCounts.client.applied)) {
    throw workbenchKafkaDebugReplayEvidenceError("Workbench Kafka debug replay legacy reducer counts differ from the typed client layer", evidence);
  }
}

function workbenchKafkaDebugReplayValidateTraceTimeline(traceTimeline, evidence) {
  if (!traceTimeline.authorityAvailable) {
    throw workbenchKafkaDebugReplayEvidenceError("Workbench Trace timeline source-authority contract is missing", evidence);
  }
  if (traceTimeline.sourceAuthorityRowCount <= 0 || traceTimeline.readableSourceRowCount <= 0) {
    throw workbenchKafkaDebugReplayEvidenceError("Workbench Trace timeline did not expose a readable source-authority row", evidence);
  }
}

function workbenchKafkaDebugReplayValidateRawWindow(rawWindow, evidence) {
  if (!rawWindow.available) {
    throw workbenchKafkaDebugReplayEvidenceError("Workbench raw HWLAB Event window contract is missing", evidence);
  }
  if (!rawWindow.samePage) throw workbenchKafkaDebugReplayEvidenceError("Workbench raw HWLAB Event inspection navigated away from the original page", evidence);
  if (!rawWindow.sourceContractPresent) throw workbenchKafkaDebugReplayEvidenceError("Workbench raw HWLAB Event window does not declare the existing product EventSource contract", evidence);
  if (!rawWindow.unfilteredContractPresent) throw workbenchKafkaDebugReplayEvidenceError("Workbench raw HWLAB Event window does not declare the unfiltered ingress contract", evidence);
  if (!rawWindow.counts) throw workbenchKafkaDebugReplayEvidenceError("Workbench raw HWLAB Event counts are unreadable", evidence);
  const counts = rawWindow.counts;
  if ([counts.received, counts.retained, counts.decoded, counts.rejected, counts.evicted, counts.bytes].some((value) => !Number.isInteger(value) || value < 0)) {
    throw workbenchKafkaDebugReplayEvidenceError("Workbench raw HWLAB Event counts are incomplete", evidence);
  }
  if (counts.received !== counts.decoded + counts.rejected || counts.received !== counts.retained + counts.evicted) {
    throw workbenchKafkaDebugReplayEvidenceError("Workbench raw HWLAB Event bounded-buffer counts are inconsistent", evidence);
  }
  if (counts.received <= 0 || counts.retained <= 0 || counts.bytes <= 0 || rawWindow.textPresent !== true || rawWindow.textBytes <= 0) {
    throw workbenchKafkaDebugReplayEvidenceError("Workbench raw HWLAB Event window did not retain a readable product SSE frame", evidence);
  }
  if (rawWindow.productEventSourceRequestCount !== 0) {
    throw workbenchKafkaDebugReplayEvidenceError("Workbench raw HWLAB Event tab opened an additional product EventSource", evidence);
  }
}

function workbenchKafkaDebugReplayEvidenceError(message, evidence) {
  const error = new Error(message);
  error.details = {
    traceId: evidence?.traceId || null,
    topic: evidence?.topic || null,
    groupId: evidence?.groupId || null,
    terminalStatus: evidence?.terminalStatus || null,
    receivedCount: evidence?.legacyCounts?.receivedCount ?? null,
    appliedCount: evidence?.legacyCounts?.appliedCount ?? null,
    phase: evidence?.phase || null,
    code: evidence?.code || null,
    layerCounts: evidence?.layerCounts || null,
    layerDom: evidence?.layerDom || null,
    layerDomStabilization: evidence?.layerDomStabilization || null,
    traceTimeline: evidence?.traceTimeline || null,
    rawHwlabEventWindow: evidence?.rawHwlabEventWindow || null,
    valuesRedacted: true
  };
  return error;
}

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