Merge pull request #1794 from pikasTech/fix/1793-webprobe-reconnect-barrier-source

修复 WebProbe 重连回放的陈旧 source-set 误判

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<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);
+});

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;
   }
 }