From ba0a5a07794f7a582e5653f7065aa0ce96925aba Mon Sep 17 00:00:00 2001 From: Codex Date: Sat, 11 Jul 2026 21:51:00 +0200 Subject: [PATCH] =?UTF-8?q?fix:=20=E6=A0=A1=E6=AD=A3=20WebProbe=20?= =?UTF-8?q?=E9=87=8D=E8=BF=9E=E5=9B=9E=E6=94=BE=E6=BA=90=E9=9B=86=E5=90=88?= =?UTF-8?q?=E6=97=B6=E5=BA=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- ...web-observe-runner-realtime-source.test.ts | 171 ++++++++++++++ ...node-web-observe-runner-realtime-source.ts | 211 +++++++++++++++--- 2 files changed, 357 insertions(+), 25 deletions(-) diff --git a/scripts/src/hwlab-node-web-observe-runner-realtime-source.test.ts b/scripts/src/hwlab-node-web-observe-runner-realtime-source.test.ts index cd22896b..11dc173c 100644 --- a/scripts/src/hwlab-node-web-observe-runner-realtime-source.test.ts +++ b/scripts/src/hwlab-node-web-observe-runner-realtime-source.test.ts @@ -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, + date: { now: () => number }, + ) => ( + debug: Record, + key: string, + subscriber: Record, + traceId: string, + label: string, + profile: Record, + ) => Promise>; + 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, + date: { now: () => number }, + ) => ( + debug: Record, + key: string, + subscriber: Record, + traceId: string, + label: string, + profile: Record, + ) => Promise>; + 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 }) => { + 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[], productRows: Record[]) => Record; + assertEqual: (kafkaRows: Record[], productRows: Record[], 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); +}); diff --git a/scripts/src/hwlab-node-web-observe-runner-realtime-source.ts b/scripts/src/hwlab-node-web-observe-runner-realtime-source.ts index cfd7b8c0..fe13655f 100644 --- a/scripts/src/hwlab-node-web-observe-runner-realtime-source.ts +++ b/scripts/src/hwlab-node-web-observe-runner-realtime-source.ts @@ -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; } }