Codex
1693 lines
89 KiB
TypeScript
1693 lines
89 KiB
TypeScript
// 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 "";
|
|
}
|
|
`;
|
|
}
|