fix: 校正 WebProbe 重连回放源集合时序
This commit is contained in:
Codex
@@ -407,3 +407,174 @@ test("realtime fanout proves one formal user event through AgentRun, HWLAB Kafka
|
||||
/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);
|
||||
});
|
||||
|
||||
@@ -112,7 +112,7 @@ async function validateRealtimeFanout(command) {
|
||||
const firstTerminal = await realtimeWaitTraceEvent(subscriberA, first.traceId, true, effective.terminalTimeoutMs);
|
||||
await realtimeWaitUiTerminal(first.traceId, effective.terminalTimeoutMs);
|
||||
const firstStable = await realtimeWaitTurnStable(debug, first.debugKey, first.traceId, [subscriberA], effective);
|
||||
const firstDebug = firstStable.debug;
|
||||
let firstDebug = firstStable.debug;
|
||||
realtimeEnrichTurnExecutionIds(first, firstDebug);
|
||||
const firstProductA = firstStable.products[0];
|
||||
realtimeAssertStableProductTransport(firstProductA, "subscriber-a-first-terminal");
|
||||
@@ -127,13 +127,25 @@ async function validateRealtimeFanout(command) {
|
||||
preTerminalSnapshots: [bBeforeDisconnect],
|
||||
profile: effective,
|
||||
});
|
||||
await realtimeCloseDebugStreams(debug, first.debugKey);
|
||||
|
||||
subscriberB2 = await realtimeNewSubscriber(storageState, sessionId, "subscriber-b-reconnected", effective);
|
||||
const bReconnected = await realtimeWaitProductConnected(subscriberB2, effective.barrierTimeoutMs);
|
||||
realtimeAssertConnectedContract(bReconnected, sessionId, "subscriber-b-reconnected", effective, { requireRetainedEvents: effective.expectedProductSse.replayPriorTraceOnReconnect });
|
||||
await sleep(effective.reconnectQuietMs);
|
||||
const reconnectBeforeSecond = await realtimeProductSnapshot(subscriberB2);
|
||||
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");
|
||||
@@ -161,7 +173,17 @@ async function validateRealtimeFanout(command) {
|
||||
realtimeAssertStableProductTransport(secondProductA, "subscriber-a-second-terminal");
|
||||
realtimeAssertStableProductTransport(secondProductB, "subscriber-b-reconnected-second-terminal");
|
||||
if (effective.expectedProductSse.replayPriorTraceOnReconnect) {
|
||||
realtimeAssertReplayedTrace(firstDebug, secondProductB, first.traceId, "subscriber-b-reconnected-after-second-turn");
|
||||
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");
|
||||
}
|
||||
@@ -179,12 +201,25 @@ async function validateRealtimeFanout(command) {
|
||||
await realtimeCloseDebugStreams(debug, second.debugKey);
|
||||
|
||||
const productA = await realtimeProductSnapshot(subscriberA);
|
||||
const productB2 = await realtimeProductSnapshot(subscriberB2);
|
||||
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");
|
||||
@@ -1214,6 +1249,61 @@ function realtimeAssertReplayedTrace(debug, product, traceId, label) {
|
||||
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,
|
||||
@@ -1331,26 +1421,97 @@ function realtimeExpectedStreamSessionId(stream, hwlabSessionId) {
|
||||
return base ? "ses_agentrun_" + base : null;
|
||||
}
|
||||
|
||||
function realtimeAssertEnvelopePassthrough(kafkaRows, productRows, requireAllKafkaRows, label) {
|
||||
const sameEnvelope = (left, right) => left.envelopeFingerprint && left.envelopeFingerprint === right.envelopeFingerprint
|
||||
&& left.eventId === right.eventId
|
||||
&& left.sourceEventId === right.sourceEventId
|
||||
&& left.sourceSeq === right.sourceSeq
|
||||
&& left.eventType === right.eventType
|
||||
&& left.traceId === right.traceId
|
||||
&& left.hwlabSessionId === right.hwlabSessionId
|
||||
&& left.runId === right.runId
|
||||
&& left.commandId === right.commandId
|
||||
&& left.userMessageId === right.userMessageId
|
||||
&& left.terminal === right.terminal;
|
||||
for (const productRow of productRows) {
|
||||
if (!kafkaRows.some((kafkaRow) => sameEnvelope(kafkaRow, productRow))) throw new Error(label + " received a product SSE envelope absent from hwlab.event.v1");
|
||||
}
|
||||
if (requireAllKafkaRows) {
|
||||
if (productRows.length !== kafkaRows.length) throw new Error(label + " envelope count differs from hwlab.event.v1");
|
||||
for (const kafkaRow of kafkaRows) {
|
||||
if (!productRows.some((productRow) => sameEnvelope(kafkaRow, productRow))) throw new Error(label + " did not receive an unchanged hwlab.event.v1 envelope");
|
||||
function 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;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user