browser/scripts/local-debug/verify-current-log-session-sse.mjs

#!/usr/bin/env node

import fs from "node:fs";

function fail(message, details = undefined) {
  console.error(message);
  if (details !== undefined) console.error(JSON.stringify(details, null, 2));
  process.exit(1);
}

function parseEvents(file) {
  const body = fs.readFileSync(file, "utf8").replace(/\r\n/g, "\n");
  const events = [];
  for (const block of body.split("\n\n")) {
    const lines = block.split("\n");
    let event = "message";
    let id = "";
    const data = [];
    for (const line of lines) {
      if (line.startsWith("event:")) event = line.slice(6).trim();
      if (line.startsWith("id:")) id = line.slice(3).trim();
      if (line.startsWith("data:")) data.push(line.slice(5).trimStart());
    }
    if (data.length === 0) continue;
    try {
      events.push({ event, id, data: JSON.parse(data.join("\n")) });
    } catch {
      // The final block may be incomplete while curl is still appending.
    }
  }
  return events;
}

function markerEvent(events, marker) {
  return events.find((item) => item.event === "log" && item.data?.entry?.line === marker);
}

function requireMarker(events, marker) {
  const item = markerEvent(events, marker);
  if (!item) fail(`missing SSE log marker: ${marker}`);
  return item;
}

const [command, file, ...args] = process.argv.slice(2);
if (!command || !file) {
  fail("usage: verify-current-log-session-sse.mjs <session-for-marker|verify> <sse-file> [...args]");
}

const events = parseEvents(file);
if (command === "session-for-marker") {
  const [marker] = args;
  const item = requireMarker(events, marker);
  if (!item.data.logSessionId) fail(`marker has no logSessionId: ${marker}`, item);
  process.stdout.write(item.data.logSessionId);
  process.exit(0);
}

if (command !== "verify") fail(`unknown command: ${command}`);

const [serverID, prefix, expectedSessionA, forbiddenIDsText = ""] = args;
if (!serverID || !prefix || !expectedSessionA) fail("verify requires server id, marker prefix, and generation A session id");

const markers = {
  aStdout: `${prefix}-A-START-STDOUT`,
  aStderr: `${prefix}-A-START-STDERR`,
  aResumeStdout: `${prefix}-A-RUN-RESUME-STDOUT`,
  aResumeStderr: `${prefix}-A-RUN-RESUME-STDERR`,
  bStdout: `${prefix}-B-START-STDOUT`,
  bStderr: `${prefix}-B-START-STDERR`
};
const markerItems = Object.fromEntries(Object.entries(markers).map(([key, marker]) => [key, requireMarker(events, marker)]));
const eventIndex = (needle) => events.indexOf(needle);
const sessionEvents = events.filter((item) => item.event === "session" && item.data?.logSessionId);
const sessionIDs = sessionEvents.map((item) => item.data.logSessionId);

if (sessionIDs.length !== 2 || sessionIDs[0] !== expectedSessionA) {
  fail("expected exactly one generation A boundary followed by one generation B boundary", { sessionIDs, expectedSessionA });
}
const sessionB = sessionIDs[1];
if (!sessionB || sessionB === expectedSessionA) fail("generation B did not receive a new session id", { sessionIDs });

for (const key of ["aStdout", "aStderr", "aResumeStdout", "aResumeStderr"]) {
  if (markerItems[key].data.logSessionId !== expectedSessionA) {
    fail(`generation A marker changed session at ${key}`, markerItems[key]);
  }
}
for (const key of ["bStdout", "bStderr"]) {
  if (markerItems[key].data.logSessionId !== sessionB) fail(`generation B marker has the wrong session at ${key}`, markerItems[key]);
}

const boundaryBIndex = eventIndex(sessionEvents[1]);
const firstBLogIndex = Math.min(eventIndex(markerItems.bStdout), eventIndex(markerItems.bStderr));
if (boundaryBIndex < 0 || boundaryBIndex >= firstBLogIndex) {
  fail("generation B session boundary was not delivered before generation B output", { boundaryBIndex, firstBLogIndex });
}
if (eventIndex(markerItems.aResumeStdout) >= boundaryBIndex || eventIndex(markerItems.aResumeStderr) >= boundaryBIndex) {
  fail("Run-resume output arrived after the process-generation boundary");
}

const allowedStreamKeys = new Set(["smoke.console.stdout", "smoke.console.stderr"]);
const forbiddenIDs = new Set(forbiddenIDsText.split(",").filter(Boolean));
const streamIDsBySession = new Map();
for (let index = 0; index < events.length; index += 1) {
  const item = events[index];
  if (item.data?.serverInstanceId && item.data.serverInstanceId !== serverID) {
    fail("SSE feed included another server instance", item);
  }
  const streamID = item.event === "stream" ? item.data?.id : item.data?.streamId;
  if (streamID && forbiddenIDs.has(streamID)) fail("SSE feed included an explicitly historical stream", item);
  if (item.event !== "stream" && item.event !== "log") continue;
  const sessionID = item.data?.logSessionId;
  const source = item.data?.source;
  const streamKey = item.data?.streamKey;
  if (item.data?.entry?.line === `${prefix}-HISTORICAL-JOB-STDOUT` || item.data?.entry?.line === `${prefix}-HISTORICAL-JOB-STDERR`) {
    fail("live SSE included historical job output", item);
  }
  if (source !== "process" || !allowedStreamKeys.has(streamKey)) {
    fail("live SSE included a legacy, job, file-tail, or undeclared stream", item);
  }
  if (index < boundaryBIndex && sessionID !== expectedSessionA) {
    fail("generation A feed mixed output from another or sessionless stream", item);
  }
  if (index > boundaryBIndex && sessionID !== sessionB) {
    fail("post-switch SSE mixed output from another session", item);
  }
  if (index > boundaryBIndex && item.data?.entry?.line?.startsWith(`${prefix}-A-`)) {
    fail("post-switch SSE retained generation A output", item);
  }
  if (streamID && sessionID) {
    if (!streamIDsBySession.has(sessionID)) streamIDsBySession.set(sessionID, new Set());
    streamIDsBySession.get(sessionID).add(streamID);
  }
}

const streamsA = streamIDsBySession.get(expectedSessionA) ?? new Set();
const streamsB = streamIDsBySession.get(sessionB) ?? new Set();
if (streamsA.size < 2 || streamsB.size < 2) fail("stdout/stderr stream metadata was incomplete", { streamsA: [...streamsA], streamsB: [...streamsB] });
for (const streamID of streamsA) {
  if (streamsB.has(streamID)) fail("process generations reused a stream id", { streamID });
}

process.stdout.write(`${JSON.stringify({
  serverInstanceId: serverID,
  sessionA: expectedSessionA,
  sessionB,
  sessionBoundaries: sessionIDs,
  markerOrder: Object.fromEntries(Object.entries(markerItems).map(([key, item]) => [key, eventIndex(item)])),
  generationAStreamIds: [...streamsA].sort(),
  generationBStreamIds: [...streamsB].sort(),
  excludedHistoricalStreamIds: [...forbiddenIDs].sort(),
  eventCount: events.length
}, null, 2)}\n`);