// 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, effective);
    const connectedA = await realtimeWaitProductConnected(subscriberA, effective.barrierTimeoutMs);
    const connectedB = await realtimeWaitProductConnected(subscriberB, effective.barrierTimeoutMs);
    realtimeAssertConnectedContract(connectedA, sessionId, "subscriber-a", effective);
    realtimeAssertConnectedContract(connectedB, sessionId, "subscriber-b", effective);
    await appendJsonl(files.control, eventRecord("realtime-fanout-product-ready", {
      profile: profileId,
      sessionId,
      subscriberCount: 2,
      expectedProductSse: effective.expectedProductSse,
      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);
    let 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,
    });
    subscriberB2 = await realtimeNewSubscriber(storageState, sessionId, "subscriber-b-reconnected", effective);
    const bReconnected = await realtimeWaitProductConnected(subscriberB2, effective.barrierTimeoutMs);
    realtimeAssertConnectedContract(bReconnected, sessionId, "subscriber-b-reconnected", effective, { requireRetainedEvents: effective.expectedProductSse.replayPriorTraceOnReconnect });
    await sleep(effective.reconnectQuietMs);
    let reconnectBeforeSecond;
    if (effective.expectedProductSse.replayPriorTraceOnReconnect) {
      const firstReplayBeforeSecond = await realtimeWaitReplayedTraceJointQuiet(
        debug,
        first.debugKey,
        subscriberB2,
        first.traceId,
        "subscriber-b-reconnected-before-second-turn",
        effective,
      );
      firstDebug = firstReplayBeforeSecond.debug;
      reconnectBeforeSecond = firstReplayBeforeSecond.product;
    } else {
      reconnectBeforeSecond = await realtimeProductSnapshot(subscriberB2);
    }
    realtimeAssertStableProductTransport(reconnectBeforeSecond, "subscriber-b-reconnected-before-second-turn");
    if (effective.expectedProductSse.replayPriorTraceOnReconnect) {
      realtimeAssertReplayedTrace(firstDebug, reconnectBeforeSecond, first.traceId, "subscriber-b-reconnected-before-second-turn");
    } else if (reconnectBeforeSecond.records.some((item) => item.traceId === first.traceId)) {
      throw new Error("reconnected subscriber replayed first-turn events contrary to the owning YAML");
    }

    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 (effective.expectedProductSse.replayPriorTraceOnReconnect) {
      const firstReplayAfterSecond = await realtimeWaitReplayedTraceJointQuiet(
        debug,
        first.debugKey,
        subscriberB2,
        first.traceId,
        "subscriber-b-reconnected-after-second-turn",
        effective,
      );
      firstDebug = firstReplayAfterSecond.debug;
      realtimeAssertStableProductTransport(firstReplayAfterSecond.product, "subscriber-b-reconnected-after-second-turn");
      realtimeAssertReplayedTrace(firstDebug, firstReplayAfterSecond.product, first.traceId, "subscriber-b-reconnected-after-second-turn");
    } else if (secondProductB.records.some((item) => item.traceId === first.traceId)) {
      throw new Error("reconnected subscriber received first-turn events contrary to the owning YAML");
    }
    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);
    let productB2 = await realtimeProductSnapshot(subscriberB2);
    if (effective.expectedProductSse.replayPriorTraceOnReconnect) {
      const firstReplayFinal = await realtimeWaitReplayedTraceJointQuiet(
        debug,
        first.debugKey,
        subscriberB2,
        first.traceId,
        "subscriber-b-reconnected-final",
        effective,
      );
      firstDebug = firstReplayFinal.debug;
      productB2 = firstReplayFinal.product;
    }
    realtimeAssertStableProductTransport(productA, "subscriber-a-final");
    realtimeAssertStableProductTransport(productB2, "subscriber-b-reconnected-final");
    if (effective.expectedProductSse.replayPriorTraceOnReconnect) {
      realtimeAssertReplayedTrace(firstDebug, productB2, first.traceId, "subscriber-b-reconnected-final");
    }
    await realtimeCloseDebugStreams(debug, first.debugKey);
    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 retention and live fanout validated",
      profile: profileId,
      sessionId,
      traceIds: [first.traceId, second.traceId],
      productSubscriberCount: 2,
      debugStreams: effective.debugStreams,
      forbiddenRequestCount: forbidden.length,
      replayedEventCount: bReconnected?.refreshReplay?.counts?.replayed || 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 through Kafka retention then live",
      sessionId,
      traceIds: effective.expectedProductSse.replayPriorTraceOnReconnect ? [first.traceId, second.traceId] : [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: productB2.records.filter((item) => item.traceId === first.traceId).length,
        valuesRedacted: true,
      },
      productUiReducer: {
        mainSessionEventSourceCount: mainStreams.length,
        sessionScoped: mainStreams[0]?.traceId === null,
        firstTraceTerminalRendered: true,
        secondTraceTerminalRendered: true,
        evidenceSource: "same-workbench-page-eventsource-and-ui-terminal",
        valuesRedacted: true,
      },
      realtimeContract: {
        expectedKafka: effective.expectedKafka,
        expectedProductSse: effective.expectedProductSse,
        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: bReconnected?.refreshReplay?.counts?.replayed || 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(() => {});
  }
}

async function validateExistingSessionRefresh(command) {
  const profileId = String(command.profile || "").trim();
  const profile = realtimeFanoutProfiles[profileId];
  if (!profile || typeof profile !== "object") throw new Error("validateExistingSessionRefresh profile is not declared by the owning YAML: " + profileId);
  const effective = realtimeFanoutEffectiveProfile(profile, command.durationMs);
  if (effective.expectedProductSse.refreshHandoff !== true) throw new Error("validateExistingSessionRefresh requires a YAML profile with refresh handoff enabled");
  const reportDir = path.join(stateDir, "artifacts", "existing-session-refresh", 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);
  };
  const report = {
    ok: false,
    contract: "web-probe-existing-session-refresh-v1",
    commandId: command.id,
    profile: profileId,
    origin,
    startedAt: new Date().toISOString(),
    valuesRedacted: true,
  };
  let phase = "session-selection";
  let sessionId = null;
  let refreshReplay = null;
  let screenshot = null;
  let requestPage = null;
  try {
    const controlRecovery = await ensureControlPageResponsiveForCommand("validateExistingSessionRefresh");
    const initial = await workbenchSessionSnapshot();
    sessionId = String(command.sessionId || initial?.activeSessionId || initial?.routeSessionId || "").trim();
    if (!sessionId) throw new Error("validateExistingSessionRefresh requires --session-id or an active Workbench session");
    if (!/^ses_[A-Za-z0-9_-]+$/u.test(sessionId)) throw new Error("validateExistingSessionRefresh session id is invalid");
    Object.assign(report, { sessionId, controlRecovery });
    if (initial?.activeSessionId !== sessionId || initial?.routeSessionId !== sessionId) {
      await gotoTarget("/workbench/sessions/" + encodeURIComponent(sessionId));
      const selected = await waitForWorkbenchSessionHydrated(page, sessionId, { timeoutMs: effective.barrierTimeoutMs });
      if (selected.ok !== true) throw new Error("existing session did not hydrate before refresh validation");
    }
    const before = await realtimeWorkbenchDomIdentity();
    if (before.messageCount < 1) throw new Error("existing session has no rendered messages to validate across refresh");
    report.before = before;

    phase = "refresh-navigation";
    requestPage = page;
    requestPage.on("request", requestListener);
    await realtimeArmExistingSessionRefreshProbe(requestPage, effective);
    await requestPage.reload({ waitUntil: "domcontentloaded", timeout: effective.barrierTimeoutMs });
    const settled = await waitForWorkbenchSessionHydrated(page, sessionId, { timeoutMs: effective.barrierTimeoutMs });
    if (settled.ok !== true) throw new Error("existing session did not hydrate after refresh");
    phase = "retention-barrier";
    const refreshState = await realtimePoll(async () => page.evaluate(() => {
      const state = window.__unideskExistingSessionRefresh;
      const streamFailure = state?.failures?.at(-1) || null;
      if (streamFailure) return { streamFailure };
      const connected = state?.connected?.at(-1) || null;
      return connected ? { connected } : null;
    }), effective.barrierTimeoutMs, "existing session refresh connected evidence");
    if (refreshState.streamFailure) throw realtimeExistingRefreshStreamError(refreshState.streamFailure);
    const connected = refreshState.connected;
    let connectedSummary = null;
    try {
      const connectedEvidence = realtimeExistingRefreshConnectedEvidence(connected, sessionId, effective);
      connectedSummary = connectedEvidence.connected;
      refreshReplay = connectedEvidence.refreshReplay;
    } catch (error) {
      connectedSummary = error?.details?.connected || null;
      refreshReplay = error?.details?.refreshReplay || null;
      if (connectedSummary) report.connected = connectedSummary;
      throw error;
    }
    phase = "live-handoff";
    await realtimeWaitForSameWorkbenchDom(before, effective.barrierTimeoutMs);
    await sleep(effective.reconnectQuietMs);
    const after = await realtimeWorkbenchDomIdentity();
    realtimeAssertSameWorkbenchDom(before, after);
    const browserEvidence = await page.evaluate(() => JSON.parse(JSON.stringify(window.__unideskExistingSessionRefresh || {})));
    if (browserEvidence.sources.length !== 1) throw new Error("main Workbench product EventSource count is " + browserEvidence.sources.length + ", expected 1");
    if (browserEvidence.sources[0]?.opened !== 1) throw new Error("main Workbench product EventSource open count is not 1");
    if (browserEvidence.connected.length !== 1) throw new Error("main Workbench connected event count is " + browserEvidence.connected.length + ", expected 1");
    if (browserEvidence.failures.length !== 0) throw realtimeExistingRefreshStreamError(browserEvidence.failures.at(-1));
    if (browserEvidence.errors.length !== 0) throw new Error("main Workbench product EventSource observed an error");
    realtimeAssertExistingRefreshDelivery(browserEvidence, refreshReplay);
    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 request count is " + mainStreams.length + ", expected 1");
    if (mainStreams[0].traceId !== null || mainStreams[0].afterSeq !== null || mainStreams[0].lastEventId !== null) throw new Error("main Workbench product SSE used a trace or cursor scope");
    if (forbidden.length > 0) throw new Error("forbidden recovery or polling requests observed: " + forbidden.map((item) => item.path).join(","));
    phase = "evidence-write";
    const screenshotFile = path.join(reportDir, "existing-session-refresh.png");
    await page.screenshot({ path: screenshotFile, fullPage: true, animations: "disabled", timeout: screenshotCaptureTimeoutMs });
    const screenshotMeta = await fileMeta(screenshotFile);
    screenshot = { kind: "existing-session-refresh-screenshot", path: screenshotFile, ...screenshotMeta, commandId: activeCommandId, valuesRedacted: true };
    await appendJsonl(files.artifacts, { ts: new Date().toISOString(), ...screenshot });
    const productSse = {
      requestCount: mainStreams.length,
      browserSourceCount: browserEvidence.sources.length,
      openCount: browserEvidence.sources[0]?.opened || 0,
      connectedEventCount: browserEvidence.connected.length,
      businessEventCount: browserEvidence.businessEventCount,
      typedFailureCount: browserEvidence.failures.length,
      livePhase: refreshReplay?.phase === "live",
      valuesRedacted: true,
    };
    Object.assign(report, {
      ok: true,
      status: "passed",
      phase: "live",
      code: "refresh_live_handoff_validated",
      completedAt: new Date().toISOString(),
      after,
      connected: connectedSummary,
      refreshReplay,
      productSse,
      forbiddenRequests: { count: 0, rows: [], valuesRedacted: true },
      screenshot,
      valuesRedacted: true,
    });
    const artifacts = await realtimeWriteEvidenceArtifacts(reportDir, network, [], "existing-session-refresh");
    report.artifacts = artifacts;
    const reportArtifact = await realtimeWriteReport(reportDir, report, "existing-session-refresh-report");
    const observer = await syncObserverPageToControlSession("validateExistingSessionRefresh", sessionId);
    await appendJsonl(files.control, eventRecord("existing-session-refresh-validated", {
      profile: profileId,
      sessionId,
      traceIds: after.traceIds,
      phase: "live",
      code: "refresh_live_handoff_validated",
      reportPath: reportArtifact.path,
      reportSha256: reportArtifact.sha256,
      screenshotSha256: screenshot.sha256,
      valuesRedacted: true,
    }));
    return {
      ok: true,
      status: "passed",
      profile: profileId,
      sessionId,
      traceIds: after.traceIds,
      phase: "live",
      code: "refresh_live_handoff_validated",
      before: realtimeWorkbenchDomSummary(before),
      after: realtimeWorkbenchDomSummary(after),
      refreshReplay,
      productSse,
      forbiddenRequestCount: 0,
      reportPath: reportArtifact.path,
      reportSha256: reportArtifact.sha256,
      screenshot,
      observer,
      valuesRedacted: true,
    };
  } catch (error) {
    const failure = realtimeExistingRefreshFailure(phase, error);
    if (!screenshot) {
      screenshot = await captureScreenshot("existing-session-refresh-failed-" + safeId(command.id))
        .then((artifact) => ({ path: artifact.path, sha256: artifact.sha256, valuesRedacted: true }))
        .catch((screenshotError) => ({ available: false, error: errorSummary(screenshotError), valuesRedacted: true }));
    }
    Object.assign(report, {
      ok: false,
      status: "failed",
      completedAt: new Date().toISOString(),
      phase: failure.phase,
      code: failure.code,
      refreshReplay,
      failure,
      screenshot,
      valuesRedacted: true,
    });
    const artifacts = await realtimeWriteEvidenceArtifacts(reportDir, network, [], "existing-session-refresh").catch(() => null);
    if (artifacts) report.artifacts = artifacts;
    const reportArtifact = await realtimeWriteReport(reportDir, report, "existing-session-refresh-report").catch(() => null);
    const wrapped = error instanceof Error ? error : new Error(String(error));
    wrapped.details = {
      ...(wrapped.details || {}),
      profile: profileId,
      sessionId,
      traceIds: report.after?.traceIds || report.before?.traceIds || [],
      phase: failure.phase,
      code: failure.code,
      failure,
      before: report.before ? realtimeWorkbenchDomSummary(report.before) : null,
      after: report.after ? realtimeWorkbenchDomSummary(report.after) : null,
      refreshReplay,
      screenshot,
      reportPath: reportArtifact?.path || null,
      reportSha256: reportArtifact?.sha256 || null,
      valuesRedacted: true,
    };
    throw wrapped;
  } finally {
    if (requestPage) requestPage.off("request", requestListener);
    await page.evaluate(() => window.sessionStorage.removeItem("__unideskExistingSessionRefreshCapture")).catch(() => {});
  }
}

function realtimeExistingRefreshFailure(phase, error) {
  const codes = {
    "session-selection": "session_selection_failed",
    "refresh-navigation": "session_hydration_failed",
    "retention-barrier": "retention_barrier_failed",
    "live-handoff": "live_handoff_failed",
    "evidence-write": "evidence_write_failed",
  };
  const streamFailure = realtimeExistingRefreshStreamFailure(error?.details?.streamFailure);
  return {
    phase,
    code: codes[phase] || "existing_session_refresh_failed",
    streamPhase: streamFailure?.phase || null,
    streamCode: streamFailure?.code || null,
    streamFailure,
    error: errorSummary(error),
    valuesRedacted: true,
  };
}

function realtimeExistingRefreshConnectedEvidence(connected, sessionId, profile) {
  const connectedSummary = realtimeConnectedSummary(connected);
  const refreshReplay = connectedSummary.refreshReplay;
  try {
    realtimeAssertConnectedContract(connected, sessionId, "existing-session-refresh", profile, { requireRetainedEvents: true });
  } catch (error) {
    const wrapped = error instanceof Error ? error : new Error(String(error));
    wrapped.details = { ...(wrapped.details || {}), connected: connectedSummary, refreshReplay, valuesRedacted: true };
    throw wrapped;
  }
  return { connected: connectedSummary, refreshReplay, valuesRedacted: true };
}

function realtimeAssertExistingRefreshDelivery(browserEvidence, refreshReplay) {
  const businessEventCount = Number(browserEvidence?.businessEventCount);
  const replayed = Number(refreshReplay?.counts?.replayed);
  const bufferedDelivered = Number(refreshReplay?.counts?.bufferedDelivered);
  if (![businessEventCount, replayed, bufferedDelivered].every((value) => Number.isInteger(value) && value >= 0)) {
    throw new Error("existing session refresh delivery counts are invalid");
  }
  const minimumDelivered = replayed + bufferedDelivered;
  if (businessEventCount < minimumDelivered) {
    throw new Error("existing session refresh browser observed " + businessEventCount + " business events, fewer than replayed plus buffered delivery " + minimumDelivered);
  }
}

function realtimeExistingRefreshStreamError(value) {
  const streamFailure = realtimeExistingRefreshStreamFailure(value);
  const error = new Error(streamFailure?.message || "existing session refresh stream reported a typed failure");
  error.details = { streamFailure, valuesRedacted: true };
  return error;
}

function realtimeExistingRefreshStreamFailure(value) {
  const nested = value?.error && typeof value.error === "object" ? value.error : value;
  const phase = typeof value?.phase === "string" ? value.phase : typeof nested?.phase === "string" ? nested.phase : null;
  const code = typeof nested?.code === "string" ? nested.code : typeof value?.code === "string" ? value.code : null;
  const message = typeof nested?.message === "string" ? nested.message.slice(0, 240) : typeof value?.message === "string" ? value.message.slice(0, 240) : null;
  if (!phase && !code && !message) return null;
  return {
    phase,
    code,
    message,
    retryable: nested?.retryable === true,
    fallback: value?.fallback === true,
    valuesRedacted: true,
  };
}

async function realtimeArmExistingSessionRefreshProbe(targetPage, profile) {
  await targetPage.addInitScript(() => {
    const markerKey = "__unideskExistingSessionRefreshCapture";
    const raw = window.sessionStorage.getItem(markerKey);
    if (!raw) return;
    window.sessionStorage.removeItem(markerKey);
    let config;
    try { config = JSON.parse(raw); } catch { return; }
    const NativeEventSource = window.EventSource;
    const state = { sources: [], connected: [], businessEventCount: 0, failures: [], errors: [] };
    window.__unideskExistingSessionRefresh = state;
    if (typeof NativeEventSource !== "function") {
      state.errors.push({ kind: "eventsource-unavailable" });
      return;
    }
    class ObservedEventSource extends NativeEventSource {
      constructor(url, options) {
        super(url, options);
        let pathname = "";
        try { pathname = new URL(String(url), window.location.href).pathname; } catch {}
        if (pathname !== config.productEventsPath) return;
        state.sources.push({ url: String(url), opened: 0 });
        const source = state.sources[state.sources.length - 1];
        this.addEventListener("open", () => { source.opened += 1; });
        this.addEventListener(config.productReadyEvent, (event) => {
          try { state.connected.push(JSON.parse(event.data)); } catch { state.errors.push({ kind: "connected-parse" }); }
        });
        this.addEventListener(config.businessEvent, () => { state.businessEventCount += 1; });
        this.addEventListener("workbench.error", (event) => {
          try { state.failures.push(JSON.parse(event.data)); } catch { state.errors.push({ kind: "workbench-error-parse" }); }
        });
        this.addEventListener("error", () => { state.errors.push({ kind: "eventsource-error" }); });
      }
    }
    Object.defineProperty(ObservedEventSource, "CONNECTING", { value: NativeEventSource.CONNECTING });
    Object.defineProperty(ObservedEventSource, "OPEN", { value: NativeEventSource.OPEN });
    Object.defineProperty(ObservedEventSource, "CLOSED", { value: NativeEventSource.CLOSED });
    window.EventSource = ObservedEventSource;
  });
  await targetPage.evaluate((config) => {
    window.sessionStorage.setItem("__unideskExistingSessionRefreshCapture", JSON.stringify(config));
  }, {
    productEventsPath: profile.productEventsPath,
    productReadyEvent: profile.productReadyEvent,
    businessEvent: profile.businessEvent,
  });
}

async function realtimeWorkbenchDomIdentity() {
  return page.evaluate(() => {
    const cards = Array.from(document.querySelectorAll("#conversation-list > .message-card"));
    const rows = cards.map((element, index) => ({
      index,
      role: element.getAttribute("data-role") || null,
      messageId: element.getAttribute("data-message-id") || null,
      traceId: element.getAttribute("data-trace-id") || null,
      status: element.getAttribute("data-status") || null,
    }));
    const identity = (row) => [row.role || "", row.messageId || "", row.traceId || ""].join(":");
    const identities = rows.map(identity);
    return {
      messageCount: rows.length,
      identities,
      missingIdentityRows: rows.filter((row) => !row.role || (!row.messageId && !row.traceId)).map((row) => row.index),
      duplicateIdentities: [...new Set(identities.filter((value, index, all) => all.indexOf(value) !== index))],
      traceIds: [...new Set(rows.map((row) => row.traceId).filter(Boolean))],
      rows,
      valuesRedacted: true,
    };
  });
}

function realtimeAssertSameWorkbenchDom(before, after) {
  if (before.missingIdentityRows.length > 0 || after.missingIdentityRows.length > 0) throw new Error("Workbench DOM contains rows without a stable message identity");
  if (before.duplicateIdentities.length > 0 || after.duplicateIdentities.length > 0) throw new Error("Workbench DOM contains duplicate message identities");
  if (before.messageCount !== after.messageCount) throw new Error("Workbench message count changed across refresh: " + before.messageCount + " -> " + after.messageCount);
  if (JSON.stringify(before.identities) !== JSON.stringify(after.identities)) throw new Error("Workbench message identity or order changed across refresh");
}

async function realtimeWaitForSameWorkbenchDom(before, timeoutMs) {
  let latest = null;
  return realtimePoll(async () => {
    latest = await realtimeWorkbenchDomIdentity();
    try {
      realtimeAssertSameWorkbenchDom(before, latest);
      return latest;
    } catch {
      return null;
    }
  }, timeoutMs, "existing session Workbench DOM identity restoration").catch((error) => {
    const wrapped = error instanceof Error ? error : new Error(String(error));
    wrapped.details = { ...(wrapped.details || {}), before: realtimeWorkbenchDomSummary(before), after: realtimeWorkbenchDomSummary(latest), valuesRedacted: true };
    throw wrapped;
  });
}

function realtimeWorkbenchDomSummary(value) {
  return value ? {
    messageCount: value.messageCount,
    identities: Array.isArray(value.identities) ? value.identities.slice(0, 80) : [],
    missingIdentityRows: Array.isArray(value.missingIdentityRows) ? value.missingIdentityRows.slice(0, 20) : [],
    duplicateIdentities: Array.isArray(value.duplicateIdentities) ? value.duplicateIdentities.slice(0, 20) : [],
    traceIds: Array.isArray(value.traceIds) ? value.traceIds.slice(0, 20) : [],
    valuesRedacted: true,
  } : null;
}

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 expectedProductSse = profile.expectedProductSse && typeof profile.expectedProductSse === "object" ? profile.expectedProductSse : null;
  if (!expectedProductSse || typeof expectedProductSse.deliverySemantics !== "string") throw new Error("realtime fanout profile lacks YAML-owned product SSE expectations");
  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");
  }
  if (expectedProductSse.refreshHandoff !== (expectedKafka.capabilities.kafkaRefreshReplay === true)) {
    throw new Error("realtime fanout product SSE refreshHandoff differs from the YAML Kafka capability");
  }
  if (expectedProductSse.replayPriorTraceOnReconnect === true && expectedProductSse.replay !== true) {
    throw new Error("realtime fanout replayPriorTraceOnReconnect requires replay=true");
  }
  const hasDurationOverride = durationMs !== null && durationMs !== undefined && durationMs !== "" && Number.isFinite(Number(durationMs));
  const terminalTimeoutMs = hasDurationOverride
    ? Math.min(Number(profile.terminalTimeoutMs), Math.max(1000, Number(durationMs)))
    : Number(profile.terminalTimeoutMs);
  return {
    ...profile,
    debugStreams,
    barrierTimeoutMs: Number(profile.barrierTimeoutMs),
    terminalTimeoutMs: Number(terminalTimeoutMs),
    terminalQuietMs: Number(profile.terminalQuietMs),
    navigationMaxAttempts: Number(profile.navigationMaxAttempts),
    navigationRetryDelayMs: Number(profile.navigationRetryDelayMs),
    reconnectQuietMs: Number(profile.reconnectQuietMs),
    forbiddenRequestPaths: Array.isArray(profile.forbiddenRequestPaths) ? profile.forbiddenRequestPaths.map(String) : [],
    requiredEventTypes: profile.requiredEventTypes && typeof profile.requiredEventTypes === "object" ? profile.requiredEventTypes : {},
    conditionalEventTypePairs: profile.conditionalEventTypePairs && typeof profile.conditionalEventTypePairs === "object" ? sanitize(profile.conditionalEventTypePairs) : {},
    expectedProductSse: sanitize(expectedProductSse),
    expectedKafka: sanitize(expectedKafka),
  };
}

async function realtimeNewSubscriber(storageState, sessionId, label, profile) {
  const subscriberContext = await browser.newContext({ storageState, viewport });
  const subscriberPage = await subscriberContext.newPage();
  await realtimeGotoHealth(subscriberPage, profile.barrierTimeoutMs, profile, label);
  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),
        userMessageId: firstText(value.userMessageId, nested.userMessageId, nested.messageId),
        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, profile) {
  const debugContext = await browser.newContext({ storageState, viewport });
  const debugPage = await debugContext.newPage();
  await realtimeGotoHealth(debugPage, profile.barrierTimeoutMs, profile, "debug-subscriber");
  return { context: debugContext, page: debugPage };
}

async function realtimeGotoHealth(targetPage, timeoutMs, profile, label) {
  const attempts = [];
  const target = new URL("/health/live", baseUrl).toString();
  for (let attempt = 1; attempt <= profile.navigationMaxAttempts; attempt += 1) {
    try {
      const response = await targetPage.goto(target, { waitUntil: "domcontentloaded", timeout: timeoutMs });
      const status = response?.status?.() ?? null;
      if (status !== null && status >= 400) throw new Error(label + " health navigation returned HTTP " + status);
      attempts.push({ attempt, ok: true, status, valuesRedacted: true });
      return { ok: true, attempts, valuesRedacted: true };
    } catch (error) {
      const message = String(error?.message || error || "unknown").slice(0, 240);
      attempts.push({ attempt, ok: false, message, valuesRedacted: true });
      if (attempt >= profile.navigationMaxAttempts || !isRetryableNavigationError(message)) {
        const wrapped = error instanceof Error ? error : new Error(message);
        wrapped.details = { ...(wrapped.details || {}), label, attempts, valuesRedacted: true };
        throw wrapped;
      }
      await sleep(profile.navigationRetryDelayMs * attempt);
    }
  }
  throw new Error(label + " health navigation retry exhausted");
}

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);
      realtimeAssertConnectedContract(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),
          userMessageId: firstText(value.userMessageId, payload.userMessageId, payload.messageId, nested.userMessageId, nested.messageId),
          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) {
  let latest = null;
  try {
    return await realtimePoll(async () => {
      const snapshot = await realtimeDebugSnapshot(debug, key);
      latest = realtimeDebugTurnCompletion(snapshot, traceId, profile);
      return latest.complete ? snapshot : null;
    }, profile.terminalTimeoutMs, "realtime debug terminal event families");
  } catch (error) {
    const wrapped = error instanceof Error ? error : new Error(String(error));
    const missing = latest?.missingEventTypes || { agentrun: [], hwlab: [] };
    wrapped.message = "realtime debug terminal event families missing agentrun=[" + missing.agentrun.join(",") + "] hwlab=[" + missing.hwlab.join(",") + "] after " + profile.terminalTimeoutMs + "ms";
    wrapped.details = { ...(wrapped.details || {}), traceId, completion: latest, valuesRedacted: true };
    throw wrapped;
  }
}

function realtimeDebugTurnCompletion(snapshot, traceId, profile) {
  const observedEventTypes = {};
  const recordCounts = {};
  for (const stream of profile.debugStreams || []) {
    const rows = (snapshot.records?.[stream] || []).filter((item) => item.traceId === traceId);
    observedEventTypes[stream] = [...new Set(rows.map((item) => item.eventType).filter(Boolean))];
    recordCounts[stream] = rows.length;
  }
  const missingEventTypes = {
    agentrun: (profile.requiredEventTypes?.agentrun || []).filter((type) => !observedEventTypes.agentrun?.includes(type)),
    hwlab: (profile.requiredEventTypes?.hwlab || []).filter((type) => !observedEventTypes.hwlab?.includes(type)),
  };
  const missingStreams = (profile.debugStreams || []).filter((stream) => !recordCounts[stream]);
  return {
    complete: missingStreams.length === 0 && missingEventTypes.agentrun.length === 0 && missingEventTypes.hwlab.length === 0,
    traceId,
    observedEventTypes,
    missingEventTypes,
    missingStreams,
    recordCounts,
    valuesRedacted: true,
  };
}

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 realtimeAssertConnectedContract(connected, sessionId, label, profile, options = {}) {
  if (connected?.realtimeSource !== profile.expectedKafka.topics.hwlab) throw new Error(label + " realtime source mismatch");
  const expected = profile.expectedProductSse;
  for (const field of ["deliverySemantics", "liveOnly", "replay", "replaySupported", "lossPossible"]) {
    if (connected?.[field] !== expected[field]) throw new Error(label + " product SSE " + field + " differs from the owning YAML");
  }
  if (connected?.filters?.sessionId !== sessionId) throw new Error(label + " session filter mismatch");
  if (connected?.filters?.traceId !== null) throw new Error(label + " unexpectedly used a trace-scoped product SSE filter");
  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");
  }
  if (expected.refreshHandoff === true) realtimeAssertRefreshHandoff(connected, label, profile, options.requireRetainedEvents === true);
  else if (connected?.refreshReplay !== undefined && connected?.refreshReplay !== null) throw new Error(label + " disclosed an unexpected refresh handoff");
}

function realtimeAssertRefreshHandoff(connected, label, profile, requireRetainedEvents) {
  const handoff = connected?.refreshReplay;
  if (!handoff || handoff.phase !== "live") throw new Error(label + " refresh handoff did not reach live phase");
  if (handoff.topic !== profile.expectedKafka.topics.hwlab) throw new Error(label + " refresh handoff topic mismatch");
  if (!Array.isArray(handoff.barrier) || handoff.barrier.length === 0) throw new Error(label + " refresh handoff barrier is missing");
  if (handoff.barrier.some((entry) => !Number.isInteger(entry?.partition) || typeof entry?.endOffset !== "string" || !entry.endOffset)) {
    throw new Error(label + " refresh handoff barrier identity is invalid");
  }
  if (!Array.isArray(handoff.topicPartitions) || handoff.topicPartitions.some((partition) => !Number.isInteger(partition))) {
    throw new Error(label + " refresh handoff topicPartitions are invalid");
  }
  const counts = handoff.counts;
  for (const field of ["matched", "replayed", "buffered", "bufferedDelivered", "liveDelivered", "deduplicated"]) {
    if (!Number.isInteger(counts?.[field]) || counts[field] < 0) throw new Error(label + " refresh handoff count " + field + " is invalid");
  }
  if (counts.replayed > counts.matched) throw new Error(label + " refresh handoff replayed count exceeds matched count");
  if (counts.bufferedDelivered > counts.buffered) throw new Error(label + " refresh handoff delivered more buffered events than observed");
  if (requireRetainedEvents && counts.replayed < 1) throw new Error(label + " refresh handoff replayed no retained events");
}

function realtimeAssertReplayedTrace(debug, product, traceId, label) {
  const kafkaRows = (debug.records?.hwlab || []).filter((item) => item.traceId === traceId);
  const productRows = (product.records || []).filter((item) => item.traceId === traceId);
  if (!productRows.some((item) => item.terminal === true)) throw new Error(label + " lacks the retained terminal event");
  realtimeAssertEnvelopePassthrough(kafkaRows, productRows, true, label + " retained trace");
}

async function realtimeWaitReplayedTraceJointQuiet(debug, key, subscriber, traceId, label, profile) {
  const timeoutMs = profile.barrierTimeoutMs;
  const deadline = Date.now() + timeoutMs;
  let previousSignature = null;
  let quietSince = null;
  let latest = null;
  while (Date.now() < deadline) {
    const debugSnapshot = await realtimeDebugSnapshot(debug, key);
    const productSnapshot = await realtimeProductSnapshot(subscriber);
    const kafkaRows = (debugSnapshot.records?.hwlab || []).filter((item) => item.traceId === traceId);
    const productRows = (productSnapshot.records || []).filter((item) => item.traceId === traceId);
    const comparison = realtimeEnvelopeMultisetComparison(kafkaRows, productRows);
    const signature = JSON.stringify({
      kafka: realtimeEnvelopeMultisetSignature(kafkaRows),
      product: realtimeEnvelopeMultisetSignature(productRows),
    });
    const now = Date.now();
    if (signature !== previousSignature) {
      previousSignature = signature;
      quietSince = now;
    }
    const quietForMs = quietSince === null ? 0 : now - quietSince;
    const terminalSeen = productRows.some((item) => item.terminal === true);
    const sourceTransport = {
      openCount: debugSnapshot.openCount?.hwlab || 0,
      connectedEventCount: debugSnapshot.connectedEventCount?.hwlab || 0,
      errorCount: (debugSnapshot.errors?.hwlab || []).length,
    };
    const sourceTransportStable = sourceTransport.openCount === 1 && sourceTransport.connectedEventCount === 1 && sourceTransport.errorCount === 0;
    latest = { comparison, terminalSeen, quietForMs, sourceTransport, sourceTransportStable };
    if (comparison.equal && terminalSeen && sourceTransportStable && quietForMs >= profile.terminalQuietMs) {
      return { debug: debugSnapshot, product: productSnapshot, comparison, quietForMs, valuesRedacted: true };
    }
    await sleep(250);
  }
  const error = new Error(label + " Kafka source and product replay envelope multisets did not converge before the joint quiet timeout");
  error.details = {
    traceId,
    timeoutMs,
    terminalQuietMs: profile.terminalQuietMs,
    terminalSeen: latest?.terminalSeen === true,
    sourceTransport: latest?.sourceTransport || null,
    sourceTransportStable: latest?.sourceTransportStable === true,
    quietForMs: latest?.quietForMs || 0,
    sourceCount: latest?.comparison?.sourceCount || 0,
    productCount: latest?.comparison?.productCount || 0,
    sourceOnly: latest?.comparison?.sourceOnly || [],
    productOnly: latest?.comparison?.productOnly || [],
    invalidSource: latest?.comparison?.invalidSource || [],
    invalidProduct: latest?.comparison?.invalidProduct || [],
    valuesRedacted: true,
  };
  throw error;
}

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 || {}),
    refreshReplay: connected?.refreshReplay ? sanitize(connected.refreshReplay) : null,
    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");
    const expectedSessionId = realtimeExpectedStreamSessionId(stream, input.sessionId);
    if (rows.some((item) => item.hwlabSessionId !== expectedSessionId)) {
      throw new Error("turn " + input.turn + " " + stream + " session lineage 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);
  realtimeValidateConditionalEventTypePairs(input);
  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);
  const agentrunUserRows = input.debug.records.agentrun.filter((item) => item.traceId === input.traceId && item.eventType === "user_message");
  const hwlabUserRows = hwlabRows.filter((item) => item.eventType === "user");
  if (agentrunUserRows.length !== 1 || hwlabUserRows.length !== 1) throw new Error("turn " + input.turn + " formal user event is not one-to-one across AgentRun and HWLAB");
  const userMessageIds = new Set([...agentrunUserRows, ...hwlabUserRows].map((item) => item.userMessageId).filter(Boolean));
  if (userMessageIds.size !== 1) throw new Error("turn " + input.turn + " formal user event lacks one stable userMessageId");
  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");
    const productUserRows = rows.filter((item) => item.eventType === "user");
    if (productUserRows.length !== 1 || productUserRows[0].userMessageId !== hwlabUserRows[0].userMessageId) {
      throw new Error("turn " + input.turn + " product SSE lacks the formal user event");
    }
  }
  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 realtimeValidateConditionalEventTypePairs(input) {
  const pairs = input.profile.conditionalEventTypePairs || {};
  const agentrunRows = (input.debug.records.agentrun || []).filter((item) => item.traceId === input.traceId);
  const hwlabRows = (input.debug.records.hwlab || []).filter((item) => item.traceId === input.traceId);
  for (const [id, pair] of Object.entries(pairs)) {
    const sourceRows = agentrunRows.filter((item) => item.eventType === pair.agentrun);
    const projectedRows = hwlabRows.filter((item) => item.eventType === pair.hwlab);
    if (sourceRows.length === 0 && projectedRows.length === 0) continue;
    if (sourceRows.length !== projectedRows.length) {
      throw new Error("turn " + input.turn + " conditional event pair " + id + " count differs AgentRun=" + sourceRows.length + " HWLAB=" + projectedRows.length);
    }
    for (const source of sourceRows) {
      if (!source.eventId || !projectedRows.some((projected) => projected.sourceEventId === source.eventId)) {
        throw new Error("turn " + input.turn + " conditional event pair " + id + " lacks HWLAB sourceEventId lineage");
      }
    }
    for (const projected of projectedRows) {
      if (!projected.sourceEventId || !sourceRows.some((source) => source.eventId === projected.sourceEventId)) {
        throw new Error("turn " + input.turn + " conditional event pair " + id + " has an orphan HWLAB event");
      }
    }
  }
}

function realtimeExpectedStreamSessionId(stream, hwlabSessionId) {
  if (stream !== "stdio") return hwlabSessionId;
  const base = String(hwlabSessionId || "")
    .replace(/^ses_/u, "")
    .replace(/[^A-Za-z0-9_]+/gu, "_")
    .replace(/^_+|_+$/gu, "");
  return base ? "ses_agentrun_" + base : null;
}

function realtimeEnvelopeIdentity(row) {
  return {
    envelopeFingerprint: row?.envelopeFingerprint || null,
    eventId: row?.eventId || null,
    sourceEventId: row?.sourceEventId || null,
    sourceSeq: Number.isFinite(row?.sourceSeq) ? row.sourceSeq : null,
    eventType: row?.eventType || null,
    traceId: row?.traceId || null,
    hwlabSessionId: row?.hwlabSessionId || null,
    runId: row?.runId || null,
    commandId: row?.commandId || null,
    userMessageId: row?.userMessageId || null,
    terminal: row?.terminal === true,
  };
}

function realtimeEnvelopeIdentityKey(row) {
  const identity = realtimeEnvelopeIdentity(row);
  return identity.envelopeFingerprint ? JSON.stringify(identity) : null;
}

function realtimeEnvelopeMultiset(rows) {
  const counts = new Map();
  const invalid = [];
  for (const row of rows) {
    const identity = realtimeEnvelopeIdentity(row);
    const key = realtimeEnvelopeIdentityKey(row);
    if (!key) {
      if (invalid.length < 8) invalid.push(identity);
      continue;
    }
    const current = counts.get(key);
    if (current) current.count += 1;
    else counts.set(key, { count: 1, identity });
  }
  return { counts, invalid };
}

function realtimeEnvelopeMultisetSignature(rows) {
  const multiset = realtimeEnvelopeMultiset(rows);
  const counts = [...multiset.counts.entries()].map(([key, entry]) => [key, entry.count]).sort((left, right) => left[0].localeCompare(right[0]));
  return JSON.stringify({ counts, invalid: multiset.invalid });
}

function realtimeEnvelopeMultisetComparison(kafkaRows, productRows) {
  const source = realtimeEnvelopeMultiset(kafkaRows);
  const product = realtimeEnvelopeMultiset(productRows);
  const sourceOnly = [];
  const productOnly = [];
  let differs = source.invalid.length > 0 || product.invalid.length > 0;
  const keys = new Set([...source.counts.keys(), ...product.counts.keys()]);
  for (const key of keys) {
    const sourceEntry = source.counts.get(key);
    const productEntry = product.counts.get(key);
    const sourceCount = sourceEntry?.count || 0;
    const productCount = productEntry?.count || 0;
    if (sourceCount > productCount) {
      differs = true;
      if (sourceOnly.length < 8) sourceOnly.push({ identity: sourceEntry.identity, count: sourceCount - productCount });
    }
    if (productCount > sourceCount) {
      differs = true;
      if (productOnly.length < 8) productOnly.push({ identity: productEntry.identity, count: productCount - sourceCount });
    }
  }
  return {
    equal: !differs && kafkaRows.length === productRows.length,
    sourceCount: kafkaRows.length,
    productCount: productRows.length,
    sourceOnly,
    productOnly,
    invalidSource: source.invalid,
    invalidProduct: product.invalid,
    valuesRedacted: true,
  };
}

function realtimeAssertEnvelopePassthrough(kafkaRows, productRows, requireAllKafkaRows, label) {
  const comparison = realtimeEnvelopeMultisetComparison(kafkaRows, productRows);
  if (comparison.productOnly.length > 0 || comparison.invalidProduct.length > 0) {
    const error = new Error(label + " received a product SSE envelope absent from hwlab.event.v1");
    error.details = comparison;
    throw error;
  }
  if (requireAllKafkaRows && !comparison.equal) {
    const message = comparison.sourceCount !== comparison.productCount
      ? label + " envelope count differs from hwlab.event.v1"
      : label + " did not receive an unchanged hwlab.event.v1 envelope";
    const error = new Error(message);
    error.details = comparison;
    throw error;
  }
}

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);
  const agentrunUser = debug.records.agentrun.find((item) => item.traceId === turn.traceId && item.eventType === "user_message") || null;
  const hwlabUser = debug.records.hwlab.find((item) => item.traceId === turn.traceId && item.eventType === "user") || null;
  const productUser = productRows.find((item) => item.eventType === "user") || null;
  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,
    formalUserEvent: {
      userMessageId: hwlabUser?.userMessageId || agentrunUser?.userMessageId || null,
      agentrunEventId: agentrunUser?.eventId || null,
      hwlabEventId: hwlabUser?.eventId || null,
      hwlabSourceEventId: hwlabUser?.sourceEventId || null,
      kafkaEnvelopeFingerprint: hwlabUser?.envelopeFingerprint || null,
      productEnvelopeFingerprint: productUser?.envelopeFingerprint || null,
      valuesRedacted: true,
    },
    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, artifactPrefix = "realtime-fanout") {
  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: artifactPrefix + "-request-ledger", commandId: activeCommandId, ...artifacts.requestLedger });
  await appendJsonl(files.artifacts, { ts: new Date().toISOString(), kind: artifactPrefix + "-events", commandId: activeCommandId, ...artifacts.events });
  return artifacts;
}

async function realtimeWriteReport(reportDir, report, artifactKind = "realtime-fanout-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: artifactKind, 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 "";
}
`;
}