// SPEC: PJ2026-0104010803 Workbench realtime visibility.
// Responsibility: Long-running, typed realtime fanout validation inside the persistent Web observer.

export function nodeWebObserveRunnerRealtimeSource(): string {
  return String.raw`
function parseRealtimeFanoutProfiles(raw) {
  if (!raw) return {};
  try {
    const parsed = JSON.parse(raw);
    return parsed && typeof parsed === "object" && !Array.isArray(parsed) ? parsed : {};
  } catch {
    return {};
  }
}

async function validateRealtimeFanout(command) {
  const profileId = String(command.profile || "").trim();
  const profile = realtimeFanoutProfiles[profileId];
  if (!profile || typeof profile !== "object") throw new Error("validateRealtimeFanout profile is not declared by the owning YAML: " + profileId);
  const provider = String(command.provider || "").trim();
  const firstPrompt = String(command.text || "").trim();
  const secondPrompt = String(command.secondText || "").trim();
  if (!provider) throw new Error("validateRealtimeFanout requires provider");
  if (!firstPrompt || !secondPrompt) throw new Error("validateRealtimeFanout requires first and second prompt");
  const effective = realtimeFanoutEffectiveProfile(profile, command.durationMs);
  const reportDir = path.join(stateDir, "artifacts", "realtime-fanout", safeId(command.id));
  await mkdir(reportDir, { recursive: true, mode: 0o700 });
  const network = [];
  const requestListener = (request) => {
    let url;
    try { url = new URL(request.url()); } catch { return; }
    if (url.origin !== new URL(baseUrl).origin) return;
    network.push({
      at: Date.now(),
      method: request.method(),
      path: url.pathname,
      sessionId: url.searchParams.get("sessionId"),
      traceId: url.searchParams.get("traceId"),
      afterSeq: url.searchParams.get("afterSeq"),
      lastEventId: request.headers()["last-event-id"] || null,
      valuesRedacted: true,
    });
    if (network.length > 1000) network.splice(0, network.length - 1000);
  };
  page.on("request", requestListener);
  let subscriberA = null;
  let subscriberB = null;
  let subscriberB2 = null;
  let debug = null;
  let createdSessionId = null;
  const report = {
    ok: false,
    contract: "web-probe-realtime-fanout-v1",
    commandId: command.id,
    profile: profileId,
    provider,
    startedAt: new Date().toISOString(),
    promptEvidence: {
      first: { hash: sha256Text(firstPrompt), bytes: Buffer.byteLength(firstPrompt) },
      second: { hash: sha256Text(secondPrompt), bytes: Buffer.byteLength(secondPrompt) },
      valuesRedacted: true,
    },
    valuesRedacted: true,
  };
  try {
    const providerSelection = await selectProvider(provider);
    const session = await createSessionFromUi();
    const sessionId = session.sessionId;
    createdSessionId = sessionId;
    report.sessionId = sessionId;
    report.providerSelection = sanitize(providerSelection);
    const storageState = await context.storageState();
    subscriberA = await realtimeNewSubscriber(storageState, sessionId, "subscriber-a", effective);
    subscriberB = await realtimeNewSubscriber(storageState, sessionId, "subscriber-b", effective);
    debug = await realtimeNewDebugContext(storageState);
    const connectedA = await realtimeWaitProductConnected(subscriberA, effective.barrierTimeoutMs);
    const connectedB = await realtimeWaitProductConnected(subscriberB, effective.barrierTimeoutMs);
    realtimeAssertLiveConnected(connectedA, sessionId, "subscriber-a", effective);
    realtimeAssertLiveConnected(connectedB, sessionId, "subscriber-b", effective);
    await appendJsonl(files.control, eventRecord("realtime-fanout-product-ready", {
      profile: profileId,
      sessionId,
      subscriberCount: 2,
      liveOnly: true,
      replay: false,
      valuesRedacted: true,
    }));

    const first = await realtimeRunTurn({
      command,
      turn: 1,
      prompt: firstPrompt,
      sessionId,
      subscribers: [subscriberA, subscriberB],
      debug,
      profile: effective,
    });
    const bBeforeDisconnect = await realtimeProductSnapshot(subscriberB);
    realtimeAssertStableProductTransport(bBeforeDisconnect, "subscriber-b-before-disconnect");
    if (bBeforeDisconnect.records.some((item) => item.traceId === first.traceId && item.terminal === true)) {
      throw new Error("first turn reached terminal before subscriber B could be disconnected; use a prompt with an observable live interval");
    }
    const bBeforeShot = await realtimeCaptureSubscriberScreenshot(subscriberB.page, reportDir, "subscriber-b-before-disconnect.png", {
      title: "Subscriber B before disconnect",
      sessionId,
      traceId: first.traceId,
      eventCount: bBeforeDisconnect.records.length,
      terminalSeen: bBeforeDisconnect.records.some((item) => item.terminal === true),
      valuesRedacted: true,
    });
    await subscriberB.context.close();
    subscriberB = null;
    const firstTerminal = await realtimeWaitTraceEvent(subscriberA, first.traceId, true, effective.terminalTimeoutMs);
    await realtimeWaitUiTerminal(first.traceId, effective.terminalTimeoutMs);
    const firstStable = await realtimeWaitTurnStable(debug, first.debugKey, first.traceId, [subscriberA], effective);
    const firstDebug = firstStable.debug;
    realtimeEnrichTurnExecutionIds(first, firstDebug);
    const firstProductA = firstStable.products[0];
    realtimeAssertStableProductTransport(firstProductA, "subscriber-a-first-terminal");
    realtimeValidateTurnEvidence({
      turn: 1,
      traceId: first.traceId,
      sessionId,
      runId: first.runId,
      commandId: first.commandId,
      debug: firstDebug,
      terminalSnapshots: [firstProductA],
      preTerminalSnapshots: [bBeforeDisconnect],
      profile: effective,
    });
    await realtimeCloseDebugStreams(debug, first.debugKey);

    subscriberB2 = await realtimeNewSubscriber(storageState, sessionId, "subscriber-b-reconnected", effective);
    const bReconnected = await realtimeWaitProductConnected(subscriberB2, effective.barrierTimeoutMs);
    realtimeAssertLiveConnected(bReconnected, sessionId, "subscriber-b-reconnected", effective);
    await sleep(effective.reconnectQuietMs);
    const reconnectBeforeSecond = await realtimeProductSnapshot(subscriberB2);
    realtimeAssertStableProductTransport(reconnectBeforeSecond, "subscriber-b-reconnected-before-second-turn");
    if (reconnectBeforeSecond.records.some((item) => item.traceId === first.traceId)) {
      throw new Error("reconnected subscriber replayed first-turn events");
    }

    const second = await realtimeRunTurn({
      command,
      turn: 2,
      prompt: secondPrompt,
      sessionId,
      subscribers: [subscriberA, subscriberB2],
      debug,
      profile: effective,
    });
    if (second.traceId === first.traceId) throw new Error("second turn reused the first business traceId");
    const secondTerminalA = await realtimeWaitTraceEvent(subscriberA, second.traceId, true, effective.terminalTimeoutMs);
    const secondTerminalB = await realtimeWaitTraceEvent(subscriberB2, second.traceId, true, effective.terminalTimeoutMs);
    await realtimeWaitUiTerminal(second.traceId, effective.terminalTimeoutMs);
    const secondStable = await realtimeWaitTurnStable(debug, second.debugKey, second.traceId, [subscriberA, subscriberB2], effective);
    const secondDebug = secondStable.debug;
    realtimeEnrichTurnExecutionIds(second, secondDebug);
    const [secondProductA, secondProductB] = secondStable.products;
    realtimeAssertStableProductTransport(secondProductA, "subscriber-a-second-terminal");
    realtimeAssertStableProductTransport(secondProductB, "subscriber-b-reconnected-second-terminal");
    if (secondProductB.records.some((item) => item.traceId === first.traceId)) {
      throw new Error("reconnected subscriber received first-turn events after the second submit");
    }
    realtimeValidateTurnEvidence({
      turn: 2,
      traceId: second.traceId,
      sessionId,
      runId: second.runId,
      commandId: second.commandId,
      debug: secondDebug,
      terminalSnapshots: [secondProductA, secondProductB],
      preTerminalSnapshots: [],
      profile: effective,
    });
    await realtimeCloseDebugStreams(debug, second.debugKey);

    const productA = await realtimeProductSnapshot(subscriberA);
    const productB2 = await realtimeProductSnapshot(subscriberB2);
    realtimeAssertStableProductTransport(productA, "subscriber-a-final");
    realtimeAssertStableProductTransport(productB2, "subscriber-b-reconnected-final");
    const mainStreams = network.filter((item) => item.path === effective.productEventsPath && item.sessionId === sessionId);
    const forbidden = network.filter((item) => realtimeForbiddenRequest(item, effective.forbiddenRequestPaths));
    if (mainStreams.length !== 1) throw new Error("main Workbench session SSE connection count is " + mainStreams.length + ", expected 1");
    if (mainStreams[0].traceId !== null) throw new Error("main Workbench switched to trace-scoped SSE");
    if (mainStreams[0].afterSeq !== null || mainStreams[0].lastEventId !== null) throw new Error("main Workbench requested replay cursor");
    if (forbidden.length > 0) throw new Error("forbidden recovery or polling requests observed: " + forbidden.map((item) => item.path).join(","));

    const summaryScreenshot = await realtimeCaptureSubscriberScreenshot(debug.page, reportDir, "realtime-fanout-summary.png", {
      title: "Pure Kafka live fanout validated",
      profile: profileId,
      sessionId,
      traceIds: [first.traceId, second.traceId],
      productSubscriberCount: 2,
      debugStreams: effective.debugStreams,
      forbiddenRequestCount: forbidden.length,
      replayedEventCount: 0,
      valuesRedacted: true,
    });
    const subscriberAScreenshot = await realtimeCaptureSubscriberScreenshot(subscriberA.page, reportDir, "subscriber-a-final.png", {
      title: "Subscriber A continuous live connection",
      sessionId,
      traceIds: [first.traceId, second.traceId],
      eventCount: productA.records.length,
      terminalCount: productA.records.filter((item) => item.terminal === true).length,
      valuesRedacted: true,
    });
    const subscriberB2Screenshot = await realtimeCaptureSubscriberScreenshot(subscriberB2.page, reportDir, "subscriber-b-reconnected-final.png", {
      title: "Subscriber B reconnected live-only",
      sessionId,
      traceIds: [second.traceId],
      eventCount: productB2.records.length,
      firstTraceReplayCount: productB2.records.filter((item) => item.traceId === first.traceId).length,
      valuesRedacted: true,
    });
    const turns = [
      realtimeTurnSummary(first, firstDebug, productA, firstTerminal),
      realtimeTurnSummary(second, secondDebug, productB2, secondTerminalB),
    ];
    Object.assign(report, {
      ok: true,
      status: "passed",
      completedAt: new Date().toISOString(),
      turns,
      productSse: {
        mainSessionConnectionCount: mainStreams.length,
        sessionScoped: true,
        afterSeqCount: 0,
        lastEventIdCount: 0,
        subscriberAEventCount: productA.records.length,
        subscriberBReconnectEventCount: productB2.records.length,
        subscriberBFirstTraceReplayCount: 0,
        valuesRedacted: true,
      },
      liveContract: {
        expectedKafka: effective.expectedKafka,
        subscriberA: realtimeConnectedSummary(connectedA),
        subscriberBBeforeDisconnect: realtimeConnectedSummary(connectedB),
        subscriberBAfterReconnect: realtimeConnectedSummary(bReconnected),
        valuesRedacted: true,
      },
      forbiddenRequests: { count: forbidden.length, rows: forbidden, valuesRedacted: true },
      screenshots: [bBeforeShot, summaryScreenshot, subscriberAScreenshot, subscriberB2Screenshot],
      warmRunnerReused: first.runnerId && second.runnerId ? first.runnerId === second.runnerId : null,
      valuesRedacted: true,
    });
    const artifacts = await realtimeWriteEvidenceArtifacts(reportDir, network, [firstDebug, secondDebug]);
    report.artifacts = artifacts;
    const reportArtifact = await realtimeWriteReport(reportDir, report);
    await appendJsonl(files.control, eventRecord("realtime-fanout-validated", {
      profile: profileId,
      sessionId,
      traceIds: turns.map((item) => item.traceId),
      runIds: turns.map((item) => item.runId),
      commandIds: turns.map((item) => item.commandId),
      reportPath: reportArtifact.path,
      reportSha256: reportArtifact.sha256,
      valuesRedacted: true,
    }));
    const observer = await syncObserverPageToControlSession("validateRealtimeFanout", sessionId);
    return {
      ok: true,
      status: "passed",
      profile: profileId,
      sessionId,
      traceIds: turns.map((item) => item.traceId),
      runIds: turns.map((item) => item.runId),
      commandIds: turns.map((item) => item.commandId),
      warmRunnerReused: report.warmRunnerReused,
      forbiddenRequestCount: 0,
      replayedEventCount: 0,
      reportPath: reportArtifact.path,
      reportSha256: reportArtifact.sha256,
      screenshots: report.screenshots,
      observer,
      valuesRedacted: true,
    };
  } catch (error) {
    Object.assign(report, {
      ok: false,
      status: "failed",
      completedAt: new Date().toISOString(),
      failure: errorSummary(error),
      valuesRedacted: true,
    });
    const partialDebug = debug ? await realtimeAllDebugSnapshots(debug).catch(() => []) : [];
    const partialProducts = await realtimePartialProductEvidence([subscriberA, subscriberB, subscriberB2]);
    const partialArtifacts = await realtimeWriteEvidenceArtifacts(reportDir, network, partialDebug).catch(() => null);
    report.partialEvidence = { products: partialProducts, debug: partialDebug.map(realtimeDebugSnapshotSummary), valuesRedacted: true };
    if (partialArtifacts) report.artifacts = partialArtifacts;
    const reportArtifact = await realtimeWriteReport(reportDir, report).catch(() => null);
    if (createdSessionId) {
      await syncObserverPageToControlSession("validateRealtimeFanout-failed", createdSessionId)
        .catch((syncError) => appendJsonl(files.errors, eventRecord("realtime-fanout-observer-sync-error", { commandId: command.id, error: errorSummary(syncError), valuesRedacted: true })));
    }
    const wrapped = error instanceof Error ? error : new Error(String(error));
    wrapped.details = {
      ...(wrapped.details || {}),
      profile: profileId,
      reportPath: reportArtifact?.path || null,
      reportSha256: reportArtifact?.sha256 || null,
      valuesRedacted: true,
    };
    throw wrapped;
  } finally {
    page.off("request", requestListener);
    if (debug) await realtimeCloseDebugStreams(debug, null).catch(() => {});
    if (debug) await debug.context.close().catch(() => {});
    if (subscriberA) await subscriberA.context.close().catch(() => {});
    if (subscriberB) await subscriberB.context.close().catch(() => {});
    if (subscriberB2) await subscriberB2.context.close().catch(() => {});
  }
}

function realtimeFanoutEffectiveProfile(profile, durationMs) {
  const debugStreams = Array.isArray(profile.debugStreams) ? profile.debugStreams.map(String) : [];
  if (Number(profile.subscriberCount) !== 2) throw new Error("realtime fanout profile subscriberCount must be 2");
  if (debugStreams.length !== 3 || new Set(debugStreams).size !== 3 || !["stdio", "agentrun", "hwlab"].every((item) => debugStreams.includes(item))) {
    throw new Error("realtime fanout profile debugStreams must be stdio, agentrun, and hwlab");
  }
  if (profile.fromBeginning !== false) throw new Error("realtime fanout profile fromBeginning must be false");
  const expectedKafka = profile.expectedKafka && typeof profile.expectedKafka === "object" ? profile.expectedKafka : null;
  if (!expectedKafka?.topics || !expectedKafka?.groups || !expectedKafka?.capabilities) throw new Error("realtime fanout profile lacks YAML-derived Kafka expectations");
  if (expectedKafka.capabilities.directPublish !== true || expectedKafka.capabilities.liveKafkaSse !== true) {
    throw new Error("realtime fanout profile requires YAML-enabled directPublish and liveKafkaSse capabilities");
  }
  const terminalTimeoutMs = Number.isFinite(Number(durationMs))
    ? Math.min(Number(profile.terminalTimeoutMs), Math.max(1000, Number(durationMs)))
    : Number(profile.terminalTimeoutMs);
  return {
    ...profile,
    debugStreams,
    barrierTimeoutMs: boundedInteger(profile.barrierTimeoutMs, 45000, 1000, 120000),
    terminalTimeoutMs: boundedInteger(terminalTimeoutMs, 480000, 1000, 600000),
    terminalQuietMs: boundedInteger(profile.terminalQuietMs, 2000, 250, 10000),
    reconnectQuietMs: boundedInteger(profile.reconnectQuietMs, 3000, 250, 30000),
    forbiddenRequestPaths: Array.isArray(profile.forbiddenRequestPaths) ? profile.forbiddenRequestPaths.map(String) : [],
    requiredEventTypes: profile.requiredEventTypes && typeof profile.requiredEventTypes === "object" ? profile.requiredEventTypes : {},
    expectedKafka: sanitize(expectedKafka),
  };
}

async function realtimeNewSubscriber(storageState, sessionId, label, profile) {
  const subscriberContext = await browser.newContext({ storageState, viewport });
  const subscriberPage = await subscriberContext.newPage();
  await subscriberPage.goto(new URL("/health/live", baseUrl).toString(), { waitUntil: "domcontentloaded", timeout: profile.barrierTimeoutMs });
  const key = label + "-" + sessionId;
  await subscriberPage.evaluate(({ key: streamKey, sessionId: scopedSessionId, profile: inputProfile }) => {
    const root = window.__unideskRealtimeFanoutProducts ??= {};
    const state = { connected: null, connectedEventCount: 0, openCount: 0, records: [], errors: [], source: null };
    root[streamKey] = state;
    const firstText = (...values) => {
      for (const value of values) if (typeof value === "string" && value.trim()) return value.trim();
      return null;
    };
    const fingerprint = (text) => {
      const bytes = new TextEncoder().encode(String(text));
      let hash = 2166136261;
      for (const byte of bytes) hash = Math.imul((hash ^ byte) >>> 0, 16777619) >>> 0;
      return "fnv1a32:" + hash.toString(16).padStart(8, "0") + ":" + bytes.length;
    };
    const source = new EventSource(inputProfile.productEventsPath + "?sessionId=" + encodeURIComponent(scopedSessionId), { withCredentials: true });
    state.source = source;
    source.onopen = () => { state.openCount += 1; };
    const record = (eventName, raw) => {
      let value;
      try { value = JSON.parse(raw); } catch { return; }
      const nested = value?.event && typeof value.event === "object" ? value.event : {};
      const context = value?.context && typeof value.context === "object" ? value.context : {};
      const eventType = firstText(nested.eventType, nested.type, value.eventType, value.type);
      state.records.push({
        eventName,
        at: Date.now(),
        schema: firstText(value.schema),
        eventId: firstText(value.eventId, nested.id),
        sourceEventId: firstText(value.sourceEventId, nested.sourceEventId),
        eventType,
        sourceSeq: Number.isFinite(Number(value.sourceSeq)) ? Number(value.sourceSeq) : Number.isFinite(Number(nested.sourceSeq)) ? Number(nested.sourceSeq) : null,
        traceId: firstText(value.traceId, nested.traceId, context.traceId),
        hwlabSessionId: firstText(value.hwlabSessionId, value.sessionId, nested.hwlabSessionId, nested.sessionId, context.sessionId),
        runId: firstText(value.runId, nested.runId, context.runId),
        commandId: firstText(value.commandId, nested.commandId, context.commandId),
        terminal: value.terminal === true || nested.terminal === true || eventType === "terminal" || eventType === "terminal_status",
        status: firstText(nested.status, value.status),
        envelopeFingerprint: fingerprint(raw),
        valuesRedacted: true,
      });
      if (state.records.length > 800) state.records.splice(0, state.records.length - 800);
    };
    source.addEventListener(inputProfile.productReadyEvent, (event) => {
      state.connectedEventCount += 1;
      try { state.connected = JSON.parse(event.data); } catch { state.connected = { parseError: true }; }
    });
    source.addEventListener(inputProfile.businessEvent, (event) => record(inputProfile.businessEvent, event.data));
    source.addEventListener("workbench.error", (event) => state.errors.push({ at: Date.now(), event: "workbench.error", dataBytes: String(event.data || "").length }));
    source.onerror = () => state.errors.push({ at: Date.now(), event: "transport.error" });
  }, { key, sessionId, profile: { productEventsPath: profile.productEventsPath, productReadyEvent: profile.productReadyEvent, businessEvent: profile.businessEvent } });
  return { label, key, context: subscriberContext, page: subscriberPage };
}

async function realtimeNewDebugContext(storageState) {
  const debugContext = await browser.newContext({ storageState, viewport });
  const debugPage = await debugContext.newPage();
  await debugPage.goto(new URL("/health/live", baseUrl).toString(), { waitUntil: "domcontentloaded", timeout: 30000 });
  return { context: debugContext, page: debugPage };
}

async function realtimeRunTurn(input) {
  const turn = input.turn;
  const prompt = input.prompt;
  await page.locator("#command-input").fill(prompt);
  await page.locator("#command-send").waitFor({ state: "visible", timeout: input.profile.barrierTimeoutMs });
  await page.waitForFunction(() => {
    const button = document.querySelector("#command-send");
    return Boolean(button && !button.disabled && button.getAttribute("data-action") === "turn");
  }, null, { timeout: input.profile.barrierTimeoutMs });
  let capturedResolve;
  let releaseResolve;
  let releasedToServer = false;
  const capturedPromise = new Promise((resolve) => { capturedResolve = resolve; });
  const releasePromise = new Promise((resolve) => { releaseResolve = resolve; });
  const handler = async (route) => {
    const request = route.request();
    let body = {};
    try { body = request.postDataJSON(); } catch {}
    capturedResolve({ traceId: String(body?.traceId || ""), sessionId: String(body?.sessionId || ""), provider: String(body?.providerProfile || "") });
    const releaseAction = await releasePromise;
    if (releaseAction === "continue") await route.continue();
    else await route.abort("aborted");
  };
  await page.route("**/v1/agent/chat", handler);
  try {
    const admissionPromise = page.waitForResponse((response) => {
      let url;
      try { url = new URL(response.url()); } catch { return false; }
      return response.request().method() === "POST" && url.pathname === "/v1/agent/chat";
    }, { timeout: Math.min(input.profile.terminalTimeoutMs, input.profile.barrierTimeoutMs + 30000) })
      .then((response) => ({ response, error: null }))
      .catch((error) => ({ response: null, error }));
    await page.locator("#command-send").click();
    const captured = await withHardTimeout(capturedPromise, input.profile.barrierTimeoutMs, "realtime turn request capture timed out");
    if (!/^trc_[A-Za-z0-9_-]+$/u.test(captured.traceId)) throw new Error("realtime turn did not pre-generate a traceId");
    if (captured.sessionId !== input.sessionId) throw new Error("realtime turn sessionId mismatch");
    if (captured.provider !== String(input.command.provider || "")) throw new Error("realtime turn provider mismatch");
    const debugKey = "turn-" + turn + "-" + captured.traceId;
    await realtimeOpenDebugStreams(input.debug, debugKey, captured.traceId, input.profile);
    const debugConnected = await realtimeWaitDebugConnected(input.debug, debugKey, input.profile);
    for (const stream of input.profile.debugStreams) {
      const connected = debugConnected[stream];
      if (connected?.consumerReady !== true) throw new Error(stream + " debug consumer is not ready");
      if (!String(connected?.groupId || "").includes("-debug-sse-")) throw new Error(stream + " debug group is not isolated");
      if (connected?.topic !== input.profile.expectedKafka.topics[stream]) throw new Error(stream + " debug topic does not match the owning YAML");
    }
    for (const subscriber of input.subscribers) {
      const connected = await realtimeWaitProductConnected(subscriber, input.profile.barrierTimeoutMs);
      realtimeAssertLiveConnected(connected, input.sessionId, subscriber.label, input.profile);
    }
    await appendJsonl(files.control, eventRecord("realtime-fanout-ready-barrier", {
      turn,
      traceId: captured.traceId,
      productSubscriberCount: input.subscribers.length,
      debugStreams: input.profile.debugStreams,
      topics: Object.fromEntries(input.profile.debugStreams.map((stream) => [stream, debugConnected[stream]?.topic || null])),
      groups: Object.fromEntries(input.profile.debugStreams.map((stream) => [stream, debugConnected[stream]?.groupId || null])),
      valuesRedacted: true,
    }));
    releasedToServer = true;
    releaseResolve("continue");
    const admissionOutcome = await admissionPromise;
    if (!admissionOutcome.response) throw new Error("realtime turn admission response failed: " + String(admissionOutcome.error?.message || admissionOutcome.error || "unknown"));
    const admission = admissionOutcome.response;
    if (admission.status() < 200 || admission.status() >= 300) throw new Error("realtime turn admission returned HTTP " + admission.status());
    let admissionBody = {};
    try { admissionBody = await admission.json(); } catch {}
    const canonicalTraceId = realtimeFirstText(admissionBody?.traceId, admissionBody?.result?.traceId, captured.traceId);
    if (canonicalTraceId !== captured.traceId) throw new Error("canonical trace changed after ready barrier");
    const nonTerminal = [];
    for (const subscriber of input.subscribers) {
      nonTerminal.push(await realtimeWaitTraceEvent(subscriber, canonicalTraceId, false, input.profile.terminalTimeoutMs));
    }
    const ids = await realtimeWaitTurnIds(input.debug, debugKey, canonicalTraceId, input.profile.terminalTimeoutMs);
    await appendJsonl(files.control, eventRecord("realtime-fanout-pre-terminal", {
      turn,
      traceId: canonicalTraceId,
      runId: ids.runId,
      commandId: ids.commandId,
      subscriberCount: nonTerminal.length,
      valuesRedacted: true,
    }));
    return { turn, traceId: canonicalTraceId, sessionId: input.sessionId, runId: ids.runId, commandId: ids.commandId, runnerId: ids.runnerId, attemptId: ids.attemptId, debugKey, admissionStatus: admission.status(), valuesRedacted: true };
  } finally {
    if (!releasedToServer) releaseResolve("abort");
    await page.unroute("**/v1/agent/chat", handler).catch(() => {});
  }
}

async function realtimeOpenDebugStreams(debug, key, traceId, profile) {
  await debug.page.evaluate(({ key: streamKey, traceId: scopedTraceId, profile: inputProfile }) => {
    const root = window.__unideskRealtimeFanoutDebug ??= {};
    const state = { connected: {}, connectedEventCount: {}, openCount: {}, records: {}, errors: {}, sources: {} };
    root[streamKey] = state;
    const firstText = (...values) => {
      for (const value of values) if (typeof value === "string" && value.trim()) return value.trim();
      return null;
    };
    const finite = (...values) => {
      for (const value of values) {
        const number = Number(value);
        if (Number.isFinite(number)) return number;
      }
      return null;
    };
    const fingerprint = (text) => {
      const bytes = new TextEncoder().encode(String(text));
      let hash = 2166136261;
      for (const byte of bytes) hash = Math.imul((hash ^ byte) >>> 0, 16777619) >>> 0;
      return "fnv1a32:" + hash.toString(16).padStart(8, "0") + ":" + bytes.length;
    };
    for (const stream of inputProfile.debugStreams) {
      state.records[stream] = [];
      state.errors[stream] = [];
      state.connectedEventCount[stream] = 0;
      state.openCount[stream] = 0;
      const params = new URLSearchParams({ stream, traceId: scopedTraceId });
      if (inputProfile.fromBeginning === true) params.set("fromBeginning", "true");
      const source = new EventSource(inputProfile.debugEventsPath + "?" + params.toString(), { withCredentials: true });
      state.sources[stream] = source;
      source.onopen = () => { state.openCount[stream] += 1; };
      source.addEventListener(inputProfile.debugReadyEvent, (event) => {
        state.connectedEventCount[stream] += 1;
        try { state.connected[stream] = JSON.parse(event.data); } catch { state.connected[stream] = { parseError: true }; }
      });
      source.addEventListener("hwlab.kafka.event", (event) => {
        let outer;
        try { outer = JSON.parse(event.data); } catch { return; }
        const value = outer?.value && typeof outer.value === "object" ? outer.value : {};
        const nested = value?.event && typeof value.event === "object" ? value.event : {};
        const payload = nested?.payload && typeof nested.payload === "object" ? nested.payload : {};
        const context = value?.context && typeof value.context === "object" ? value.context : {};
        const run = value?.run && typeof value.run === "object" ? value.run : {};
        const command = value?.command && typeof value.command === "object" ? value.command : {};
        const eventType = stream === "agentrun"
          ? firstText(nested.type, value.eventType)
          : stream === "hwlab"
            ? firstText(nested.eventType, nested.type, value.eventType)
            : firstText(value.eventType, value.stdio?.method);
        state.records[stream].push({
          stream,
          at: Date.now(),
          topic: firstText(outer.topic),
          partition: finite(outer.partition),
          offset: firstText(outer.offset),
          schema: firstText(value.schema),
          eventId: firstText(value.eventId, nested.id),
          sourceEventId: firstText(value.sourceEventId, nested.sourceEventId),
          eventType,
          sourceSeq: finite(value.sourceSeq, nested.seq, nested.sourceSeq),
          frameSeq: finite(value.frameSeq, value.stdio?.frameSeq),
          traceId: firstText(value.traceId, context.traceId, payload.traceId, nested.traceId),
          hwlabSessionId: firstText(value.hwlabSessionId, value.sessionId, run.hwlabSessionId, run.sessionId, context.sessionId, payload.hwlabSessionId, payload.sessionId, nested.hwlabSessionId, nested.sessionId),
          runId: firstText(value.runId, run.runId, context.runId, nested.runId, payload.runId),
          commandId: firstText(value.commandId, command.commandId, context.commandId, payload.commandId, nested.commandId),
          runnerId: firstText(value.runnerId, run.runnerId, command.runnerId, context.runnerId, payload.runnerId, nested.runnerId),
          attemptId: firstText(value.attemptId, run.attemptId, command.attemptId, context.attemptId, payload.attemptId, nested.attemptId),
          terminal: value.terminal === true || nested.terminal === true || eventType === "terminal" || eventType === "terminal_status",
          direction: firstText(value.stdio?.direction),
          method: firstText(value.stdio?.method),
          envelopeFingerprint: stream === "hwlab" ? fingerprint(JSON.stringify(value)) : null,
          valuesRedacted: true,
        });
        if (state.records[stream].length > 1000) state.records[stream].splice(0, state.records[stream].length - 1000);
      });
      source.addEventListener("hwlab.kafka.error", (event) => state.errors[stream].push({ at: Date.now(), dataBytes: String(event.data || "").length }));
      source.onerror = () => state.errors[stream].push({ at: Date.now(), transport: true });
    }
  }, { key, traceId, profile: {
    debugStreams: profile.debugStreams,
    debugEventsPath: profile.debugEventsPath,
    debugReadyEvent: profile.debugReadyEvent,
    fromBeginning: profile.fromBeginning,
  } });
}

async function realtimeWaitProductConnected(subscriber, timeoutMs) {
  return realtimePoll(async () => (await realtimeProductSnapshot(subscriber)).connected, timeoutMs, "product connected");
}

async function realtimeWaitDebugConnected(debug, key, profile) {
  return realtimePoll(async () => {
    const snapshot = await realtimeDebugSnapshot(debug, key);
    return profile.debugStreams.every((stream) => snapshot.connected[stream]?.consumerReady === true) ? snapshot.connected : null;
  }, profile.barrierTimeoutMs, "debug consumers ready");
}

async function realtimeWaitTurnIds(debug, key, traceId, timeoutMs) {
  return realtimePoll(async () => {
    const snapshot = await realtimeDebugSnapshot(debug, key);
    const rows = Object.values(snapshot.records).flat().filter((item) => item.traceId === traceId);
    const runId = rows.map((item) => item.runId).find(Boolean) || null;
    const commandId = rows.map((item) => item.commandId).find(Boolean) || null;
    const runnerId = rows.map((item) => item.runnerId).find(Boolean) || null;
    const attemptId = rows.map((item) => item.attemptId).find(Boolean) || null;
    return runId && commandId ? { runId, commandId, runnerId, attemptId } : null;
  }, timeoutMs, "realtime run and command ids");
}

async function realtimeWaitDebugTurnComplete(debug, key, traceId, profile) {
  return realtimePoll(async () => {
    const snapshot = await realtimeDebugSnapshot(debug, key);
    if (!profile.debugStreams.every((stream) => (snapshot.records[stream] || []).some((item) => item.traceId === traceId))) return null;
    const agentrunTypes = new Set((snapshot.records.agentrun || []).filter((item) => item.traceId === traceId).map((item) => item.eventType));
    const hwlabTypes = new Set((snapshot.records.hwlab || []).filter((item) => item.traceId === traceId).map((item) => item.eventType));
    if (!(profile.requiredEventTypes.agentrun || []).every((type) => agentrunTypes.has(type))) return null;
    if (!(profile.requiredEventTypes.hwlab || []).every((type) => hwlabTypes.has(type))) return null;
    return snapshot;
  }, profile.terminalTimeoutMs, "realtime debug terminal event families");
}

async function realtimeWaitTurnStable(debug, key, traceId, subscribers, profile) {
  await realtimeWaitDebugTurnComplete(debug, key, traceId, profile);
  const deadline = Date.now() + profile.terminalTimeoutMs;
  let previousSignature = null;
  let stableSince = Date.now();
  while (Date.now() < deadline) {
    const debugSnapshot = await realtimeDebugSnapshot(debug, key);
    const productSnapshots = await Promise.all(subscribers.map((subscriber) => realtimeProductSnapshot(subscriber)));
    const signature = JSON.stringify({
      debug: Object.fromEntries(profile.debugStreams.map((stream) => {
        const rows = (debugSnapshot.records[stream] || []).filter((item) => item.traceId === traceId);
        return [stream, { count: rows.length, lastOffset: rows.at(-1)?.offset || null, lastFingerprint: rows.at(-1)?.envelopeFingerprint || null }];
      })),
      products: productSnapshots.map((snapshot) => {
        const rows = snapshot.records.filter((item) => item.traceId === traceId);
        return { count: rows.length, lastSourceSeq: rows.at(-1)?.sourceSeq || null, lastFingerprint: rows.at(-1)?.envelopeFingerprint || null };
      }),
    });
    if (signature !== previousSignature) {
      previousSignature = signature;
      stableSince = Date.now();
    } else if (Date.now() - stableSince >= profile.terminalQuietMs) {
      return { debug: debugSnapshot, products: productSnapshots };
    }
    await sleep(250);
  }
  throw new Error("realtime debug/product streams did not become quiet after terminal");
}

async function realtimeWaitTraceEvent(subscriber, traceId, terminal, timeoutMs) {
  return realtimePoll(async () => {
    const snapshot = await realtimeProductSnapshot(subscriber);
    return snapshot.records.find((item) => item.traceId === traceId && item.terminal === terminal) || null;
  }, timeoutMs, terminal ? "terminal product event" : "pre-terminal product event");
}

async function realtimeWaitUiTerminal(traceId, timeoutMs) {
  await page.waitForFunction((expectedTraceId) => {
    const card = document.querySelector(".message-card[data-role='agent'][data-trace-id='" + expectedTraceId + "']");
    if (!card) return false;
    const status = card.getAttribute("data-status");
    const body = card.querySelector(".message-text")?.textContent?.trim() || "";
    return status === "completed" && body.length > 0;
  }, traceId, { timeout: timeoutMs });
}

async function realtimeProductSnapshot(subscriber) {
  return subscriber.page.evaluate((key) => {
    const state = window.__unideskRealtimeFanoutProducts?.[key];
    if (!state) return { connected: null, connectedEventCount: 0, openCount: 0, records: [], errors: [{ missing: true }] };
    return { connected: state.connected ? JSON.parse(JSON.stringify(state.connected)) : null, connectedEventCount: state.connectedEventCount, openCount: state.openCount, records: JSON.parse(JSON.stringify(state.records)), errors: JSON.parse(JSON.stringify(state.errors)) };
  }, subscriber.key);
}

async function realtimeDebugSnapshot(debug, key) {
  return debug.page.evaluate((streamKey) => {
    const state = window.__unideskRealtimeFanoutDebug?.[streamKey];
    if (!state) return { connected: {}, connectedEventCount: {}, openCount: {}, records: { stdio: [], agentrun: [], hwlab: [] }, errors: { stdio: [], agentrun: [], hwlab: [] } };
    return { connected: JSON.parse(JSON.stringify(state.connected)), connectedEventCount: JSON.parse(JSON.stringify(state.connectedEventCount)), openCount: JSON.parse(JSON.stringify(state.openCount)), records: JSON.parse(JSON.stringify(state.records)), errors: JSON.parse(JSON.stringify(state.errors)) };
  }, key);
}

async function realtimeAllDebugSnapshots(debug) {
  return debug.page.evaluate(() => {
    const root = window.__unideskRealtimeFanoutDebug || {};
    return Object.entries(root).map(([key, state]) => ({
      key,
      connected: JSON.parse(JSON.stringify(state.connected || {})),
      connectedEventCount: JSON.parse(JSON.stringify(state.connectedEventCount || {})),
      openCount: JSON.parse(JSON.stringify(state.openCount || {})),
      records: JSON.parse(JSON.stringify(state.records || {})),
      errors: JSON.parse(JSON.stringify(state.errors || {})),
    }));
  });
}

async function realtimeCloseDebugStreams(debug, key) {
  if (!debug?.page || debug.page.isClosed()) return;
  await debug.page.evaluate((streamKey) => {
    const root = window.__unideskRealtimeFanoutDebug || {};
    const keys = streamKey ? [streamKey] : Object.keys(root);
    for (const current of keys) {
      for (const source of Object.values(root[current]?.sources || {})) source?.close?.();
    }
  }, key);
}

function realtimeAssertLiveConnected(connected, sessionId, label, profile) {
  if (connected?.realtimeSource !== profile.expectedKafka.topics.hwlab) throw new Error(label + " realtime source mismatch");
  if (connected?.deliverySemantics !== "live-only" || connected?.liveOnly !== true) throw new Error(label + " is not live-only");
  if (connected?.replay !== false || connected?.replaySupported !== false || connected?.lossPossible !== true) throw new Error(label + " replay/loss disclosure mismatch");
  if (connected?.filters?.sessionId !== sessionId) throw new Error(label + " session filter mismatch");
  const capabilities = connected?.capabilities || {};
  for (const [name, expected] of Object.entries(profile.expectedKafka.capabilities)) {
    if (capabilities[name] !== expected) throw new Error(label + " capability " + name + " differs from the owning YAML");
  }
}

function realtimeConnectedSummary(connected) {
  return {
    realtimeSource: connected?.realtimeSource || null,
    deliverySemantics: connected?.deliverySemantics || null,
    liveOnly: connected?.liveOnly === true,
    replay: connected?.replay === true,
    replaySupported: connected?.replaySupported === true,
    lossPossible: connected?.lossPossible === true,
    filters: { sessionId: connected?.filters?.sessionId || null, traceId: connected?.filters?.traceId || null },
    capabilities: sanitize(connected?.capabilities || {}),
    valuesRedacted: true,
  };
}

function realtimeAssertStableProductTransport(snapshot, label) {
  if (snapshot.openCount !== 1) throw new Error(label + " product SSE open count is " + snapshot.openCount + ", expected 1");
  if (snapshot.connectedEventCount !== 1) throw new Error(label + " product connected event count is " + snapshot.connectedEventCount + ", expected 1");
  if ((snapshot.errors || []).length !== 0) throw new Error(label + " product SSE observed a transport or business error");
}

function realtimeEnrichTurnExecutionIds(turn, debug) {
  const rows = Object.values(debug.records || {}).flat().filter((item) => item.traceId === turn.traceId);
  turn.runnerId ||= rows.map((item) => item.runnerId).find(Boolean) || null;
  turn.attemptId ||= rows.map((item) => item.attemptId).find(Boolean) || null;
}

function realtimeValidateTurnEvidence(input) {
  const profile = input.profile;
  for (const stream of profile.debugStreams) {
    if (input.debug.connected[stream]?.consumerReady !== true) throw new Error("turn " + input.turn + " " + stream + " consumer not ready");
    if (input.debug.connected[stream]?.topic !== profile.expectedKafka.topics[stream]) throw new Error("turn " + input.turn + " " + stream + " topic mismatch");
    if (input.debug.openCount[stream] !== 1 || input.debug.connectedEventCount[stream] !== 1) throw new Error("turn " + input.turn + " " + stream + " debug SSE reconnected");
    const rows = (input.debug.records[stream] || []).filter((item) => item.traceId === input.traceId);
    if (rows.length === 0) throw new Error("turn " + input.turn + " " + stream + " has no trace records");
    if (!realtimeMonotonicOffsets(rows)) throw new Error("turn " + input.turn + " " + stream + " offsets are not monotonic");
    if (rows.some((item) => item.hwlabSessionId !== input.sessionId)) throw new Error("turn " + input.turn + " " + stream + " session missing or mismatched");
    if (rows.some((item) => item.runId !== input.runId)) throw new Error("turn " + input.turn + " " + stream + " runId missing or mismatched");
    if (rows.some((item) => item.commandId && item.commandId !== input.commandId)) throw new Error("turn " + input.turn + " " + stream + " commandId mismatched");
    const commandBoundRows = stream === "stdio"
      ? rows
      : rows.filter((item) => (profile.requiredEventTypes[stream] || []).includes(item.eventType));
    if (commandBoundRows.length === 0 || commandBoundRows.some((item) => item.commandId !== input.commandId)) {
      throw new Error("turn " + input.turn + " " + stream + " command-bound event lacks the current commandId");
    }
    if ((input.debug.errors[stream] || []).length !== 0) throw new Error("turn " + input.turn + " " + stream + " debug SSE error");
  }
  const stdio = input.debug.records.stdio.filter((item) => item.traceId === input.traceId);
  const frameSeqs = stdio.map((item) => item.frameSeq).filter(Number.isFinite);
  if (frameSeqs.length === 0 || !realtimeStrictlyIncreasing(frameSeqs)) throw new Error("turn " + input.turn + " stdio frameSeq is not strictly increasing");
  const agentrunTypes = new Set(input.debug.records.agentrun.filter((item) => item.traceId === input.traceId).map((item) => item.eventType));
  const hwlabTypes = new Set(input.debug.records.hwlab.filter((item) => item.traceId === input.traceId).map((item) => item.eventType));
  for (const type of profile.requiredEventTypes.agentrun || []) if (!agentrunTypes.has(type)) throw new Error("turn " + input.turn + " AgentRun lacks " + type);
  for (const type of profile.requiredEventTypes.hwlab || []) if (!hwlabTypes.has(type)) throw new Error("turn " + input.turn + " HWLAB lacks " + type);
  const identityRows = Object.values(input.debug.records).flat().filter((item) => item.traceId === input.traceId);
  if (!input.runId || new Set(identityRows.map((item) => item.runId)).size !== 1) throw new Error("turn " + input.turn + " runId missing or mismatched across streams");
  const commandIds = new Set(identityRows.map((item) => item.commandId).filter(Boolean));
  if (!input.commandId || commandIds.size !== 1 || !commandIds.has(input.commandId)) throw new Error("turn " + input.turn + " commandId missing or mismatched across command-bound events");
  const hwlabRows = input.debug.records.hwlab.filter((item) => item.traceId === input.traceId);
  for (const snapshot of input.terminalSnapshots) {
    const rows = snapshot.records.filter((item) => item.traceId === input.traceId);
    if (!rows.some((item) => item.terminal !== true) || !rows.some((item) => item.terminal === true)) throw new Error("turn " + input.turn + " product subscriber lacks pre-terminal or terminal event");
    realtimeAssertEnvelopePassthrough(hwlabRows, rows, true, "turn " + input.turn + " terminal subscriber");
  }
  for (const snapshot of input.preTerminalSnapshots) {
    const rows = snapshot.records.filter((item) => item.traceId === input.traceId);
    if (!rows.some((item) => item.terminal !== true)) throw new Error("turn " + input.turn + " disconnected subscriber lacks pre-terminal event");
    realtimeAssertEnvelopePassthrough(hwlabRows, rows, false, "turn " + input.turn + " pre-terminal subscriber");
  }
}

function realtimeAssertEnvelopePassthrough(kafkaRows, productRows, requireAllKafkaRows, label) {
  const sameEnvelope = (left, right) => left.envelopeFingerprint && left.envelopeFingerprint === right.envelopeFingerprint
    && left.eventId === right.eventId
    && left.sourceEventId === right.sourceEventId
    && left.sourceSeq === right.sourceSeq
    && left.eventType === right.eventType
    && left.traceId === right.traceId
    && left.hwlabSessionId === right.hwlabSessionId
    && left.runId === right.runId
    && left.commandId === right.commandId
    && left.terminal === right.terminal;
  for (const productRow of productRows) {
    if (!kafkaRows.some((kafkaRow) => sameEnvelope(kafkaRow, productRow))) throw new Error(label + " received a product SSE envelope absent from hwlab.event.v1");
  }
  if (requireAllKafkaRows) {
    if (productRows.length !== kafkaRows.length) throw new Error(label + " envelope count differs from hwlab.event.v1");
    for (const kafkaRow of kafkaRows) {
      if (!productRows.some((productRow) => sameEnvelope(kafkaRow, productRow))) throw new Error(label + " did not receive an unchanged hwlab.event.v1 envelope");
    }
  }
}

function realtimeMonotonicOffsets(rows) {
  const partitions = new Map();
  for (const row of rows) {
    let offset;
    try { offset = BigInt(String(row.offset)); } catch { return false; }
    if (!Number.isFinite(row.partition)) return false;
    const partition = row.partition;
    const previous = partitions.get(partition);
    if (previous !== undefined && offset <= previous) return false;
    partitions.set(partition, offset);
  }
  return partitions.size > 0;
}

function realtimeStrictlyIncreasing(values) {
  for (let index = 1; index < values.length; index += 1) if (!(values[index] > values[index - 1])) return false;
  return true;
}

function realtimeForbiddenRequest(item, forbiddenPaths) {
  const pathValue = String(item.path || "").toLowerCase();
  return forbiddenPaths.some((entry) => pathValue.includes(String(entry).toLowerCase()));
}

function realtimeTurnSummary(turn, debug, product, terminal) {
  const streams = {};
  for (const [stream, rowsRaw] of Object.entries(debug.records)) {
    const rows = rowsRaw.filter((item) => item.traceId === turn.traceId);
    streams[stream] = {
      topic: debug.connected[stream]?.topic || rows[0]?.topic || null,
      groupId: debug.connected[stream]?.groupId || null,
      count: rows.length,
      partitions: [...new Set(rows.map((item) => item.partition).filter(Number.isFinite))],
      firstOffset: rows[0]?.offset || null,
      lastOffset: rows.at(-1)?.offset || null,
      eventTypes: [...new Set(rows.map((item) => item.eventType).filter(Boolean))],
      firstFrameSeq: rows.map((item) => item.frameSeq).find(Number.isFinite) || null,
      lastFrameSeq: rows.map((item) => item.frameSeq).filter(Number.isFinite).at(-1) || null,
      openCount: debug.openCount?.[stream] || 0,
      connectedEventCount: debug.connectedEventCount?.[stream] || 0,
      errorCount: (debug.errors?.[stream] || []).length,
      valuesRedacted: true,
    };
  }
  const productRows = product.records.filter((item) => item.traceId === turn.traceId);
  return {
    turn: turn.turn,
    traceId: turn.traceId,
    runId: turn.runId,
    commandId: turn.commandId,
    runnerId: turn.runnerId,
    attemptId: turn.attemptId,
    sessionId: turn.sessionId,
    admissionStatus: turn.admissionStatus,
    terminalStatus: terminal?.status || null,
    streams,
    product: { count: productRows.length, openCount: product.openCount, connectedEventCount: product.connectedEventCount, errorCount: product.errors.length, preTerminalSeen: productRows.some((item) => item.terminal !== true), terminalSeen: productRows.some((item) => item.terminal === true), valuesRedacted: true },
    valuesRedacted: true,
  };
}

async function realtimeCaptureSubscriberScreenshot(targetPage, reportDir, name, summary) {
  await targetPage.evaluate((value) => {
    document.documentElement.style.background = "#0b1020";
    document.body.innerHTML = "";
    const pre = document.createElement("pre");
    pre.style.cssText = "color:#d7e3ff;background:#111a30;padding:32px;border-radius:16px;font:18px/1.55 ui-monospace,monospace;white-space:pre-wrap;margin:40px;";
    pre.textContent = JSON.stringify(value, null, 2);
    document.body.appendChild(pre);
  }, summary);
  const file = path.join(reportDir, name);
  await targetPage.screenshot({ path: file, fullPage: true, animations: "disabled", timeout: screenshotCaptureTimeoutMs });
  const meta = await fileMeta(file);
  const artifact = { kind: "realtime-fanout-screenshot", path: file, name, byteCount: meta.byteCount, sha256: meta.sha256, commandId: activeCommandId, valuesRedacted: true };
  await appendJsonl(files.artifacts, { ts: new Date().toISOString(), ...artifact });
  return artifact;
}

async function realtimePartialProductEvidence(subscribers) {
  const out = [];
  for (const subscriber of subscribers.filter(Boolean)) {
    const snapshot = await realtimeProductSnapshot(subscriber).catch(() => null);
    if (!snapshot) continue;
    out.push({
      label: subscriber.label,
      openCount: snapshot.openCount,
      connectedEventCount: snapshot.connectedEventCount,
      errorCount: snapshot.errors.length,
      eventCount: snapshot.records.length,
      traceIds: [...new Set(snapshot.records.map((item) => item.traceId).filter(Boolean))].slice(0, 4),
      terminalCount: snapshot.records.filter((item) => item.terminal === true).length,
      connected: realtimeConnectedSummary(snapshot.connected),
      valuesRedacted: true,
    });
  }
  return out;
}

function realtimeDebugSnapshotSummary(snapshot) {
  return {
    key: snapshot.key || null,
    streams: Object.fromEntries(Object.entries(snapshot.records || {}).map(([stream, rows]) => [stream, {
      topic: snapshot.connected?.[stream]?.topic || null,
      groupId: snapshot.connected?.[stream]?.groupId || null,
      openCount: snapshot.openCount?.[stream] || 0,
      connectedEventCount: snapshot.connectedEventCount?.[stream] || 0,
      errorCount: (snapshot.errors?.[stream] || []).length,
      eventCount: rows.length,
      traceIds: [...new Set(rows.map((item) => item.traceId).filter(Boolean))].slice(0, 4),
      valuesRedacted: true,
    }])),
    valuesRedacted: true,
  };
}

async function realtimeWriteEvidenceArtifacts(reportDir, network, debugSnapshots) {
  const requestsFile = path.join(reportDir, "request-ledger.jsonl");
  const eventsFile = path.join(reportDir, "events.jsonl");
  await writeFile(requestsFile, network.map((item) => JSON.stringify(item)).join("\n") + "\n", { mode: 0o600 });
  const eventRows = [];
  for (const snapshot of debugSnapshots) {
    for (const [stream, rows] of Object.entries(snapshot.records || {})) {
      for (const row of rows) eventRows.push({ ...row, stream, valuesRedacted: true });
    }
  }
  await writeFile(eventsFile, eventRows.map((item) => JSON.stringify(item)).join("\n") + "\n", { mode: 0o600 });
  const requestMeta = await fileMeta(requestsFile);
  const eventMeta = await fileMeta(eventsFile);
  const artifacts = {
    requestLedger: { path: requestsFile, ...requestMeta, valuesRedacted: true },
    events: { path: eventsFile, ...eventMeta, count: eventRows.length, valuesRedacted: true },
    valuesRedacted: true,
  };
  await appendJsonl(files.artifacts, { ts: new Date().toISOString(), kind: "realtime-fanout-request-ledger", commandId: activeCommandId, ...artifacts.requestLedger });
  await appendJsonl(files.artifacts, { ts: new Date().toISOString(), kind: "realtime-fanout-events", commandId: activeCommandId, ...artifacts.events });
  return artifacts;
}

async function realtimeWriteReport(reportDir, report) {
  const reportFile = path.join(reportDir, "report.json");
  await writeFile(reportFile, JSON.stringify(sanitize(report), null, 2) + "\n", { mode: 0o600 });
  const meta = await fileMeta(reportFile);
  const artifact = { kind: "realtime-fanout-report", path: reportFile, byteCount: meta.byteCount, sha256: meta.sha256, commandId: activeCommandId, valuesRedacted: true };
  await appendJsonl(files.artifacts, { ts: new Date().toISOString(), ...artifact });
  return artifact;
}

async function realtimePoll(read, timeoutMs, label) {
  const deadline = Date.now() + timeoutMs;
  while (Date.now() < deadline) {
    const value = await read();
    if (value) return value;
    await sleep(250);
  }
  throw new Error(label + " timed out after " + timeoutMs + "ms");
}

function realtimeFirstText(...values) {
  for (const value of values) if (typeof value === "string" && value.trim()) return value.trim();
  return "";
}
`;
}