Codex
581 lines
23 KiB
TypeScript
581 lines
23 KiB
TypeScript
import assert from "node:assert/strict";
|
|
import { test } from "bun:test";
|
|
|
|
import { nodeWebObserveRunnerRealtimeSource } from "./hwlab-node-web-observe-runner-realtime-source";
|
|
|
|
test("realtime fanout keeps the YAML terminal timeout when durationMs is absent", () => {
|
|
const source = nodeWebObserveRunnerRealtimeSource();
|
|
const build = new Function(
|
|
"boundedInteger",
|
|
"sanitize",
|
|
`${source}\nreturn realtimeFanoutEffectiveProfile;`,
|
|
) as (
|
|
boundedInteger: (value: unknown, fallback: number, min: number, max: number) => number,
|
|
sanitize: <T>(value: T) => T,
|
|
) => (profile: Record<string, unknown>, durationMs: unknown) => Record<string, unknown>;
|
|
const effectiveProfile = build(
|
|
(value, fallback, min, max) => Math.min(max, Math.max(min, Number.isFinite(Number(value)) ? Number(value) : fallback)),
|
|
(value) => value,
|
|
);
|
|
const profile = {
|
|
subscriberCount: 2,
|
|
debugStreams: ["stdio", "agentrun", "hwlab"],
|
|
fromBeginning: false,
|
|
barrierTimeoutMs: 45_000,
|
|
terminalTimeoutMs: 480_000,
|
|
terminalQuietMs: 2_000,
|
|
reconnectQuietMs: 3_000,
|
|
expectedProductSse: {
|
|
deliverySemantics: "kafka-retention-then-live",
|
|
liveOnly: false,
|
|
replay: true,
|
|
replaySupported: true,
|
|
lossPossible: false,
|
|
refreshHandoff: true,
|
|
replayPriorTraceOnReconnect: true,
|
|
},
|
|
forbiddenRequestPaths: [],
|
|
requiredEventTypes: { agentrun: ["terminal_status"], hwlab: ["terminal"] },
|
|
conditionalEventTypePairs: { "tool-call": { agentrun: "tool_call", hwlab: "tool" } },
|
|
expectedKafka: {
|
|
topics: { stdio: "codex-stdio.raw.v1", agentrun: "agentrun.event.v1", hwlab: "hwlab.event.v1" },
|
|
groups: { directPublish: "direct", transactionalProjector: "projector", liveSse: "live" },
|
|
capabilities: { directPublish: true, liveKafkaSse: true, kafkaRefreshReplay: true },
|
|
},
|
|
};
|
|
|
|
assert.equal(effectiveProfile(profile, null).terminalTimeoutMs, 480_000);
|
|
assert.equal(effectiveProfile(profile, undefined).terminalTimeoutMs, 480_000);
|
|
assert.equal(effectiveProfile(profile, 5_000).terminalTimeoutMs, 5_000);
|
|
assert.deepEqual(effectiveProfile(profile, null).conditionalEventTypePairs, profile.conditionalEventTypePairs);
|
|
});
|
|
|
|
test("realtime fanout terminal completion does not require absent optional tool families", () => {
|
|
const source = nodeWebObserveRunnerRealtimeSource();
|
|
const build = new Function(`${source}\nreturn realtimeDebugTurnCompletion;`) as () => (
|
|
snapshot: Record<string, any>, traceId: string, profile: Record<string, any>,
|
|
) => Record<string, any>;
|
|
const completion = build();
|
|
const traceId = "trc_direct_answer";
|
|
const profile = {
|
|
debugStreams: ["stdio", "agentrun", "hwlab"],
|
|
requiredEventTypes: {
|
|
agentrun: ["user_message", "assistant_message", "terminal_status"],
|
|
hwlab: ["user", "assistant", "terminal"],
|
|
},
|
|
conditionalEventTypePairs: {
|
|
"tool-call": { agentrun: "tool_call", hwlab: "tool" },
|
|
"command-output": { agentrun: "command_output", hwlab: "status" },
|
|
},
|
|
};
|
|
const snapshot = {
|
|
records: {
|
|
stdio: [{ traceId, eventType: "codex.stdio.stdout" }],
|
|
agentrun: ["user_message", "backend_status", "assistant_message", "terminal_status"].map((eventType) => ({ traceId, eventType })),
|
|
hwlab: ["user", "backend", "assistant", "terminal"].map((eventType) => ({ traceId, eventType })),
|
|
},
|
|
};
|
|
|
|
const directAnswer = completion(snapshot, traceId, profile);
|
|
assert.equal(directAnswer.complete, true);
|
|
assert.deepEqual(directAnswer.missingEventTypes, { agentrun: [], hwlab: [] });
|
|
|
|
const missingTerminal = completion({
|
|
records: {
|
|
...snapshot.records,
|
|
agentrun: snapshot.records.agentrun.filter((row: Record<string, string>) => row.eventType !== "terminal_status"),
|
|
hwlab: snapshot.records.hwlab.filter((row: Record<string, string>) => row.eventType !== "terminal"),
|
|
},
|
|
}, traceId, profile);
|
|
assert.equal(missingTerminal.complete, false);
|
|
assert.deepEqual(missingTerminal.missingEventTypes, { agentrun: ["terminal_status"], hwlab: ["terminal"] });
|
|
assert.ok(missingTerminal.observedEventTypes.agentrun.includes("assistant_message"));
|
|
});
|
|
|
|
test("realtime fanout validates optional tool events only when they occur", () => {
|
|
const source = nodeWebObserveRunnerRealtimeSource();
|
|
const build = new Function(`${source}\nreturn realtimeValidateConditionalEventTypePairs;`) as () => (
|
|
input: Record<string, any>,
|
|
) => void;
|
|
const validate = build();
|
|
const traceId = "trc_optional_tool";
|
|
const base = {
|
|
turn: 1,
|
|
traceId,
|
|
profile: {
|
|
conditionalEventTypePairs: {
|
|
"tool-call": { agentrun: "tool_call", hwlab: "tool" },
|
|
"command-output": { agentrun: "command_output", hwlab: "status" },
|
|
},
|
|
},
|
|
};
|
|
|
|
assert.doesNotThrow(() => validate({ ...base, debug: { records: { agentrun: [], hwlab: [] } } }));
|
|
assert.doesNotThrow(() => validate({
|
|
...base,
|
|
debug: {
|
|
records: {
|
|
agentrun: [{ traceId, eventType: "tool_call", eventId: "evt_tool" }],
|
|
hwlab: [{ traceId, eventType: "tool", sourceEventId: "evt_tool" }],
|
|
},
|
|
},
|
|
}));
|
|
assert.throws(
|
|
() => validate({
|
|
...base,
|
|
debug: {
|
|
records: {
|
|
agentrun: [{ traceId, eventType: "command_output", eventId: "evt_output" }],
|
|
hwlab: [],
|
|
},
|
|
},
|
|
}),
|
|
/conditional event pair command-output count differs/u,
|
|
);
|
|
});
|
|
|
|
test("realtime fanout retries a YAML-bounded transient health navigation", async () => {
|
|
const source = nodeWebObserveRunnerRealtimeSource();
|
|
const build = new Function(
|
|
"sleep",
|
|
"isRetryableNavigationError",
|
|
"baseUrl",
|
|
`${source}\nreturn realtimeGotoHealth;`,
|
|
) as (
|
|
sleep: (milliseconds: number) => Promise<void>,
|
|
isRetryableNavigationError: (message: string) => boolean,
|
|
baseUrl: string,
|
|
) => (page: { goto: () => Promise<{ status: () => number }> }, timeoutMs: number, profile: Record<string, number>, label: string) => Promise<Record<string, unknown>>;
|
|
const gotoHealth = build(
|
|
async () => {},
|
|
(message) => /ERR_NETWORK_CHANGED/u.test(message),
|
|
"https://hwlab.example.test",
|
|
);
|
|
let calls = 0;
|
|
const result = await gotoHealth({
|
|
async goto() {
|
|
calls += 1;
|
|
if (calls === 1) throw new Error("page.goto: net::ERR_NETWORK_CHANGED");
|
|
return { status: () => 200 };
|
|
},
|
|
}, 45_000, { navigationMaxAttempts: 4, navigationRetryDelayMs: 1_000 }, "subscriber-a");
|
|
|
|
assert.equal(result.ok, true);
|
|
assert.equal(calls, 2);
|
|
assert.equal((result.attempts as unknown[]).length, 2);
|
|
});
|
|
|
|
test("realtime fanout validates stdio against the scoped AgentRun session lineage", () => {
|
|
const source = nodeWebObserveRunnerRealtimeSource();
|
|
const build = new Function(`${source}\nreturn realtimeExpectedStreamSessionId;`) as () => (
|
|
stream: string,
|
|
hwlabSessionId: string,
|
|
) => string;
|
|
const expectedStreamSessionId = build();
|
|
const hwlabSessionId = "ses_5ec4e141-6abc-466d-9afe-049f7c0ac105";
|
|
|
|
assert.equal(expectedStreamSessionId("stdio", hwlabSessionId), "ses_agentrun_5ec4e141_6abc_466d_9afe_049f7c0ac105");
|
|
assert.equal(expectedStreamSessionId("agentrun", hwlabSessionId), hwlabSessionId);
|
|
assert.equal(expectedStreamSessionId("hwlab", hwlabSessionId), hwlabSessionId);
|
|
});
|
|
|
|
test("realtime fanout validates YAML-owned refresh and legacy live-only connected contracts", () => {
|
|
const source = nodeWebObserveRunnerRealtimeSource();
|
|
const build = new Function(`${source}\nreturn realtimeAssertConnectedContract;`) as () => (
|
|
connected: Record<string, unknown>,
|
|
sessionId: string,
|
|
label: string,
|
|
profile: Record<string, unknown>,
|
|
options?: Record<string, unknown>,
|
|
) => void;
|
|
const assertConnected = build();
|
|
const sessionId = "ses_refresh_contract";
|
|
const expectedKafka = {
|
|
topics: { hwlab: "hwlab.event.v1" },
|
|
capabilities: { directPublish: true, liveKafkaSse: true, kafkaRefreshReplay: true },
|
|
};
|
|
const refreshProfile = {
|
|
expectedKafka,
|
|
expectedProductSse: {
|
|
deliverySemantics: "kafka-retention-then-live",
|
|
liveOnly: false,
|
|
replay: true,
|
|
replaySupported: true,
|
|
lossPossible: false,
|
|
refreshHandoff: true,
|
|
replayPriorTraceOnReconnect: true,
|
|
},
|
|
};
|
|
const refreshConnected = {
|
|
realtimeSource: "hwlab.event.v1",
|
|
deliverySemantics: "kafka-retention-then-live",
|
|
liveOnly: false,
|
|
replay: true,
|
|
replaySupported: true,
|
|
lossPossible: false,
|
|
filters: { sessionId, traceId: null },
|
|
capabilities: expectedKafka.capabilities,
|
|
refreshReplay: {
|
|
phase: "live",
|
|
topic: "hwlab.event.v1",
|
|
topicPartitions: [0],
|
|
barrier: [{ partition: 0, endOffset: "18" }],
|
|
counts: { matched: 8, replayed: 8, buffered: 1, bufferedDelivered: 1, liveDelivered: 0, deduplicated: 0 },
|
|
},
|
|
};
|
|
|
|
assert.doesNotThrow(() => assertConnected(refreshConnected, sessionId, "refresh", refreshProfile, { requireRetainedEvents: true }));
|
|
assert.throws(
|
|
() => assertConnected({ ...refreshConnected, deliverySemantics: "live-only" }, sessionId, "refresh", refreshProfile),
|
|
/differs from the owning YAML/u,
|
|
);
|
|
assert.throws(
|
|
() => assertConnected({ ...refreshConnected, refreshReplay: { ...refreshConnected.refreshReplay, counts: { ...refreshConnected.refreshReplay.counts, replayed: 9 } } }, sessionId, "refresh", refreshProfile),
|
|
/replayed count exceeds matched count/u,
|
|
);
|
|
|
|
const liveProfile = {
|
|
expectedKafka: { topics: { hwlab: "hwlab.event.v1" }, capabilities: { directPublish: true, liveKafkaSse: true, kafkaRefreshReplay: false } },
|
|
expectedProductSse: {
|
|
deliverySemantics: "live-only",
|
|
liveOnly: true,
|
|
replay: false,
|
|
replaySupported: false,
|
|
lossPossible: true,
|
|
refreshHandoff: false,
|
|
replayPriorTraceOnReconnect: false,
|
|
},
|
|
};
|
|
assert.doesNotThrow(() => assertConnected({
|
|
realtimeSource: "hwlab.event.v1",
|
|
deliverySemantics: "live-only",
|
|
liveOnly: true,
|
|
replay: false,
|
|
replaySupported: false,
|
|
lossPossible: true,
|
|
filters: { sessionId, traceId: null },
|
|
capabilities: liveProfile.expectedKafka.capabilities,
|
|
}, sessionId, "live", liveProfile));
|
|
});
|
|
|
|
test("existing session refresh rejects duplicate or reordered Workbench identities", () => {
|
|
const source = nodeWebObserveRunnerRealtimeSource();
|
|
const build = new Function(`${source}\nreturn realtimeAssertSameWorkbenchDom;`) as () => (
|
|
before: Record<string, unknown>, after: Record<string, unknown>,
|
|
) => void;
|
|
const assertSame = build();
|
|
const before = { messageCount: 2, identities: ["user:m1:", "agent:m2:t1"], missingIdentityRows: [], duplicateIdentities: [] };
|
|
assert.doesNotThrow(() => assertSame(before, { ...before }));
|
|
assert.throws(() => assertSame(before, { ...before, identities: [...before.identities].reverse() }), /identity or order changed/u);
|
|
assert.throws(() => assertSame(before, { ...before, duplicateIdentities: ["agent:m2:t1"] }), /duplicate message identities/u);
|
|
assert.throws(() => assertSame(before, { ...before, missingIdentityRows: [1] }), /without a stable message identity/u);
|
|
});
|
|
|
|
test("existing session refresh failures preserve typed phase and code", () => {
|
|
const source = nodeWebObserveRunnerRealtimeSource();
|
|
const build = new Function(`function errorSummary(error) { return { message: error.message }; }\n${source}\nreturn { realtimeExistingRefreshFailure, realtimeExistingRefreshStreamError };`) as () => {
|
|
realtimeExistingRefreshFailure: (phase: string, error: Error & { details?: Record<string, unknown> }) => Record<string, unknown>;
|
|
realtimeExistingRefreshStreamError: (value: Record<string, unknown>) => Error & { details?: Record<string, unknown> };
|
|
};
|
|
const helpers = build();
|
|
const error = helpers.realtimeExistingRefreshStreamError({
|
|
phase: "flushing",
|
|
error: { code: "workbench_kafka_refresh_barrier_gap", message: "Kafka retention barrier gap", retryable: true },
|
|
fallback: false,
|
|
});
|
|
const failure = helpers.realtimeExistingRefreshFailure("retention-barrier", error);
|
|
assert.equal(failure.phase, "retention-barrier");
|
|
assert.equal(failure.code, "retention_barrier_failed");
|
|
assert.equal(failure.streamPhase, "flushing");
|
|
assert.equal(failure.streamCode, "workbench_kafka_refresh_barrier_gap");
|
|
assert.equal((failure.streamFailure as Record<string, unknown>).fallback, false);
|
|
assert.deepEqual(failure.valuesRedacted, true);
|
|
});
|
|
|
|
test("existing session refresh preserves rejected connected evidence and rejects missing browser replay delivery", () => {
|
|
const source = nodeWebObserveRunnerRealtimeSource();
|
|
const build = new Function(
|
|
"sanitize",
|
|
`${source}\nreturn { realtimeExistingRefreshConnectedEvidence, realtimeAssertExistingRefreshDelivery };`,
|
|
) as (sanitize: <T>(value: T) => T) => {
|
|
realtimeExistingRefreshConnectedEvidence: (connected: Record<string, any>, sessionId: string, profile: Record<string, any>) => Record<string, unknown>;
|
|
realtimeAssertExistingRefreshDelivery: (browser: Record<string, unknown>, refreshReplay: Record<string, any>) => void;
|
|
};
|
|
const helpers = build((value) => value);
|
|
const sessionId = "ses_existing";
|
|
const refreshReplay = {
|
|
phase: "live",
|
|
topic: "hwlab.event.v1",
|
|
topicPartitions: [0],
|
|
barrier: [{ partition: 0, endOffset: "18" }],
|
|
counts: { matched: 8, replayed: 8, buffered: 1, bufferedDelivered: 1, liveDelivered: 0, deduplicated: 0 },
|
|
};
|
|
const profile = {
|
|
expectedKafka: { topics: { hwlab: "hwlab.event.v1" }, capabilities: { directPublish: true, liveKafkaSse: true, kafkaRefreshReplay: true } },
|
|
expectedProductSse: { deliverySemantics: "kafka-retention-then-live", liveOnly: false, replay: true, replaySupported: true, lossPossible: false, refreshHandoff: true },
|
|
};
|
|
const connected = {
|
|
realtimeSource: "hwlab.event.v1",
|
|
deliverySemantics: "invalid",
|
|
liveOnly: false,
|
|
replay: true,
|
|
replaySupported: true,
|
|
lossPossible: false,
|
|
filters: { sessionId, traceId: null },
|
|
capabilities: profile.expectedKafka.capabilities,
|
|
refreshReplay,
|
|
};
|
|
assert.throws(
|
|
() => helpers.realtimeExistingRefreshConnectedEvidence(connected, sessionId, profile),
|
|
(error: Error & { details?: Record<string, any> }) => {
|
|
assert.deepEqual(error.details?.refreshReplay?.barrier, refreshReplay.barrier);
|
|
assert.equal(error.details?.refreshReplay?.counts?.replayed, 8);
|
|
return true;
|
|
},
|
|
);
|
|
assert.doesNotThrow(() => helpers.realtimeAssertExistingRefreshDelivery({ businessEventCount: 9 }, refreshReplay));
|
|
assert.throws(
|
|
() => helpers.realtimeAssertExistingRefreshDelivery({ businessEventCount: 8 }, refreshReplay),
|
|
/fewer than replayed plus buffered delivery 9/u,
|
|
);
|
|
});
|
|
|
|
test("realtime fanout proves one formal user event through AgentRun, HWLAB Kafka, and product SSE", () => {
|
|
const source = nodeWebObserveRunnerRealtimeSource();
|
|
const build = new Function(`${source}\nreturn realtimeValidateTurnEvidence;`) as () => (input: Record<string, unknown>) => void;
|
|
const validate = build();
|
|
const traceId = "trc_user_passthrough";
|
|
const sessionId = "ses_user_passthrough";
|
|
const runId = "run_user_passthrough";
|
|
const commandId = "cmd_user_passthrough";
|
|
const userMessageId = "msg_user_passthrough";
|
|
const base = { traceId, runId, commandId, terminal: false };
|
|
const userHwlab = {
|
|
...base,
|
|
partition: 0,
|
|
offset: "0",
|
|
hwlabSessionId: sessionId,
|
|
eventId: "hwlab:evt_user",
|
|
sourceEventId: "evt_user",
|
|
sourceSeq: 1,
|
|
eventType: "user",
|
|
userMessageId,
|
|
envelopeFingerprint: "fingerprint:user",
|
|
};
|
|
const terminalHwlab = {
|
|
...base,
|
|
partition: 0,
|
|
offset: "1",
|
|
hwlabSessionId: sessionId,
|
|
eventId: "hwlab:evt_terminal",
|
|
sourceEventId: "evt_terminal",
|
|
sourceSeq: 2,
|
|
eventType: "terminal",
|
|
userMessageId: null,
|
|
terminal: true,
|
|
envelopeFingerprint: "fingerprint:terminal",
|
|
};
|
|
const debug = {
|
|
connected: {
|
|
stdio: { consumerReady: true, topic: "codex-stdio.raw.v1" },
|
|
agentrun: { consumerReady: true, topic: "agentrun.event.v1" },
|
|
hwlab: { consumerReady: true, topic: "hwlab.event.v1" },
|
|
},
|
|
openCount: { stdio: 1, agentrun: 1, hwlab: 1 },
|
|
connectedEventCount: { stdio: 1, agentrun: 1, hwlab: 1 },
|
|
errors: { stdio: [], agentrun: [], hwlab: [] },
|
|
records: {
|
|
stdio: [{ ...base, partition: 0, offset: "0", frameSeq: 1, hwlabSessionId: "ses_agentrun_user_passthrough" }],
|
|
agentrun: [
|
|
{ ...base, partition: 0, offset: "0", hwlabSessionId: sessionId, eventId: "evt_user", eventType: "user_message", userMessageId },
|
|
{ ...base, partition: 0, offset: "1", hwlabSessionId: sessionId, eventId: "evt_terminal", eventType: "terminal_status", terminal: true },
|
|
],
|
|
hwlab: [userHwlab, terminalHwlab],
|
|
},
|
|
};
|
|
const product = { openCount: 1, connectedEventCount: 1, errors: [], records: [{ ...userHwlab }, { ...terminalHwlab }] };
|
|
const profile = {
|
|
debugStreams: ["stdio", "agentrun", "hwlab"],
|
|
requiredEventTypes: { agentrun: ["user_message", "terminal_status"], hwlab: ["user", "terminal"] },
|
|
expectedKafka: { topics: { stdio: "codex-stdio.raw.v1", agentrun: "agentrun.event.v1", hwlab: "hwlab.event.v1" } },
|
|
};
|
|
|
|
assert.doesNotThrow(() => validate({ turn: 1, traceId, sessionId, runId, commandId, debug, terminalSnapshots: [product], preTerminalSnapshots: [], profile }));
|
|
const missingUserProduct = { ...product, records: [terminalHwlab] };
|
|
assert.throws(
|
|
() => validate({ turn: 1, traceId, sessionId, runId, commandId, debug, terminalSnapshots: [missingUserProduct], preTerminalSnapshots: [], profile }),
|
|
/product subscriber lacks pre-terminal or terminal event|product SSE lacks the formal user event/u,
|
|
);
|
|
});
|
|
|
|
function replayEnvelope(eventId: string, sourceSeq: number, terminal = false) {
|
|
return {
|
|
envelopeFingerprint: `fingerprint:${eventId}`,
|
|
eventId,
|
|
sourceEventId: `source:${eventId}`,
|
|
sourceSeq,
|
|
eventType: terminal ? "terminal" : "assistant",
|
|
traceId: "trc_joint_quiet",
|
|
hwlabSessionId: "ses_joint_quiet",
|
|
runId: "run_joint_quiet",
|
|
commandId: "cmd_joint_quiet",
|
|
userMessageId: null,
|
|
terminal,
|
|
};
|
|
}
|
|
|
|
test("realtime replay joint quiet waits for a late formal Kafka envelope before accepting 32/32", async () => {
|
|
const source = nodeWebObserveRunnerRealtimeSource();
|
|
let now = 0;
|
|
let sleepCount = 0;
|
|
const first = replayEnvelope("evt_first", 1);
|
|
const late = replayEnvelope("evt_late", 2, true);
|
|
let kafkaRows = [first];
|
|
const productRows = [first, late];
|
|
const build = new Function(
|
|
"sleep",
|
|
"Date",
|
|
`${source}\nreturn realtimeWaitReplayedTraceJointQuiet;`,
|
|
) as (
|
|
sleep: () => Promise<void>,
|
|
date: { now: () => number },
|
|
) => (
|
|
debug: Record<string, any>,
|
|
key: string,
|
|
subscriber: Record<string, any>,
|
|
traceId: string,
|
|
label: string,
|
|
profile: Record<string, number>,
|
|
) => Promise<Record<string, any>>;
|
|
const waitForJointQuiet = build(
|
|
async () => {
|
|
sleepCount += 1;
|
|
now += 250;
|
|
if (sleepCount === 1) kafkaRows = [first, late];
|
|
},
|
|
{ now: () => now },
|
|
);
|
|
const debug = {
|
|
page: {
|
|
async evaluate() {
|
|
return {
|
|
connected: {}, connectedEventCount: { hwlab: 1 }, openCount: { hwlab: 1 }, errors: { hwlab: [] },
|
|
records: { stdio: [], agentrun: [], hwlab: structuredClone(kafkaRows) },
|
|
};
|
|
},
|
|
},
|
|
};
|
|
const subscriber = {
|
|
page: {
|
|
async evaluate() {
|
|
return { connected: {}, connectedEventCount: 1, openCount: 1, errors: [], records: structuredClone(productRows) };
|
|
},
|
|
},
|
|
};
|
|
|
|
const result = await waitForJointQuiet(
|
|
debug,
|
|
"turn-1-trc_joint_quiet",
|
|
subscriber,
|
|
"trc_joint_quiet",
|
|
"reconnected",
|
|
{ barrierTimeoutMs: 2_000, terminalQuietMs: 500 },
|
|
);
|
|
|
|
assert.equal(result.comparison.equal, true);
|
|
assert.equal(result.comparison.sourceCount, 2);
|
|
assert.equal(result.comparison.productCount, 2);
|
|
assert.equal(result.quietForMs, 500);
|
|
assert.ok(sleepCount >= 3);
|
|
});
|
|
|
|
test("realtime replay joint quiet rejects a permanent 31/32 split with bounded identity diagnostics", async () => {
|
|
const source = nodeWebObserveRunnerRealtimeSource();
|
|
let now = 0;
|
|
const first = replayEnvelope("evt_first", 1);
|
|
const late = replayEnvelope("evt_late", 2, true);
|
|
const build = new Function(
|
|
"sleep",
|
|
"Date",
|
|
`${source}\nreturn realtimeWaitReplayedTraceJointQuiet;`,
|
|
) as (
|
|
sleep: () => Promise<void>,
|
|
date: { now: () => number },
|
|
) => (
|
|
debug: Record<string, any>,
|
|
key: string,
|
|
subscriber: Record<string, any>,
|
|
traceId: string,
|
|
label: string,
|
|
profile: Record<string, number>,
|
|
) => Promise<Record<string, any>>;
|
|
const waitForJointQuiet = build(async () => { now += 250; }, { now: () => now });
|
|
const debug = {
|
|
page: {
|
|
async evaluate() {
|
|
return { connected: {}, connectedEventCount: { hwlab: 1 }, openCount: { hwlab: 1 }, errors: { hwlab: [] }, records: { stdio: [], agentrun: [], hwlab: [first] } };
|
|
},
|
|
},
|
|
};
|
|
const subscriber = {
|
|
page: {
|
|
async evaluate() {
|
|
return { connected: {}, connectedEventCount: 1, openCount: 1, errors: [], records: [first, late] };
|
|
},
|
|
},
|
|
};
|
|
|
|
await assert.rejects(
|
|
() => waitForJointQuiet(
|
|
debug,
|
|
"turn-1-trc_joint_quiet",
|
|
subscriber,
|
|
"trc_joint_quiet",
|
|
"reconnected",
|
|
{ barrierTimeoutMs: 750, terminalQuietMs: 250 },
|
|
),
|
|
(error: Error & { details?: Record<string, any> }) => {
|
|
assert.match(error.message, /did not converge before the joint quiet timeout/u);
|
|
assert.equal(error.details?.sourceCount, 1);
|
|
assert.equal(error.details?.productCount, 2);
|
|
assert.equal(error.details?.productOnly?.[0]?.identity?.eventId, "evt_late");
|
|
assert.equal(error.details?.productOnly?.[0]?.count, 1);
|
|
return true;
|
|
},
|
|
);
|
|
});
|
|
|
|
test("realtime envelope equality is an exact multiset even when counts match", () => {
|
|
const source = nodeWebObserveRunnerRealtimeSource();
|
|
const build = new Function(`${source}\nreturn { compare: realtimeEnvelopeMultisetComparison, assertEqual: realtimeAssertEnvelopePassthrough };`) as () => {
|
|
compare: (kafkaRows: Record<string, any>[], productRows: Record<string, any>[]) => Record<string, any>;
|
|
assertEqual: (kafkaRows: Record<string, any>[], productRows: Record<string, any>[], requireAll: boolean, label: string) => void;
|
|
};
|
|
const helpers = build();
|
|
const first = replayEnvelope("evt_first", 1);
|
|
const sourceTerminal = replayEnvelope("evt_terminal", 2, true);
|
|
const changedTerminal = { ...sourceTerminal, envelopeFingerprint: "fingerprint:changed", sourceEventId: "source:changed" };
|
|
const comparison = helpers.compare([first, sourceTerminal], [first, changedTerminal]);
|
|
|
|
assert.equal(comparison.equal, false);
|
|
assert.equal(comparison.sourceCount, 2);
|
|
assert.equal(comparison.productCount, 2);
|
|
assert.equal(comparison.sourceOnly[0].identity.eventId, "evt_terminal");
|
|
assert.equal(comparison.productOnly[0].identity.sourceEventId, "source:changed");
|
|
assert.throws(
|
|
() => helpers.assertEqual([first, sourceTerminal], [first, changedTerminal], true, "reconnected"),
|
|
/product SSE envelope absent from hwlab\.event\.v1/u,
|
|
);
|
|
});
|
|
|
|
test("realtime fanout closes the first debug stream only after the final joint quiet gate", () => {
|
|
const source = nodeWebObserveRunnerRealtimeSource();
|
|
const afterSecond = source.indexOf("const firstReplayAfterSecond = await realtimeWaitReplayedTraceJointQuiet");
|
|
const finalGate = source.indexOf("const firstReplayFinal = await realtimeWaitReplayedTraceJointQuiet");
|
|
const close = source.indexOf("await realtimeCloseDebugStreams(debug, first.debugKey);");
|
|
|
|
assert.ok(afterSecond > 0);
|
|
assert.ok(finalGate > afterSecond);
|
|
assert.ok(close > finalGate);
|
|
});
|