fix: validate Kafka refresh delivery in web probe

scripts/src/hwlab-node-lanes.ts

@@ -192,6 +192,15 @@ export interface HwlabRuntimeWebProbeRealtimeFanoutProfileSpec {
   readonly navigationMaxAttempts: number;
   readonly navigationRetryDelayMs: number;
   readonly reconnectQuietMs: number;
+  readonly expectedProductSse: Readonly<{
+    deliverySemantics: string;
+    liveOnly: boolean;
+    replay: boolean;
+    replaySupported: boolean;
+    lossPossible: boolean;
+    refreshHandoff: boolean;
+    replayPriorTraceOnReconnect: boolean;
+  }>;
   readonly forbiddenRequestPaths: readonly string[];
   readonly requiredEventTypes: Readonly<{
     agentrun: readonly string[];
@@ -1435,7 +1444,23 @@ function webProbeRealtimeFanoutProfileConfig(value: unknown, path: string): Hwla
     throw new Error(`${path}.debugStreams must contain stdio, agentrun, and hwlab exactly once`);
   }
   const fromBeginning = booleanField(raw, "fromBeginning", path);
-  if (fromBeginning !== false) throw new Error(`${path}.fromBeginning must be false for live-only validation`);
+  if (fromBeginning !== false) throw new Error(`${path}.fromBeginning must be false for turn-scoped debug streams`);
+  const expectedProductSseRaw = asRecord(raw.expectedProductSse, `${path}.expectedProductSse`);
+  const expectedProductSse = {
+    deliverySemantics: stringField(expectedProductSseRaw, "deliverySemantics", `${path}.expectedProductSse`),
+    liveOnly: booleanField(expectedProductSseRaw, "liveOnly", `${path}.expectedProductSse`),
+    replay: booleanField(expectedProductSseRaw, "replay", `${path}.expectedProductSse`),
+    replaySupported: booleanField(expectedProductSseRaw, "replaySupported", `${path}.expectedProductSse`),
+    lossPossible: booleanField(expectedProductSseRaw, "lossPossible", `${path}.expectedProductSse`),
+    refreshHandoff: booleanField(expectedProductSseRaw, "refreshHandoff", `${path}.expectedProductSse`),
+    replayPriorTraceOnReconnect: booleanField(expectedProductSseRaw, "replayPriorTraceOnReconnect", `${path}.expectedProductSse`),
+  };
+  if (expectedProductSse.refreshHandoff && (!expectedProductSse.replay || !expectedProductSse.replaySupported || expectedProductSse.liveOnly)) {
+    throw new Error(`${path}.expectedProductSse refresh handoff requires replay=true, replaySupported=true, and liveOnly=false`);
+  }
+  if (expectedProductSse.replayPriorTraceOnReconnect && !expectedProductSse.replay) {
+    throw new Error(`${path}.expectedProductSse replayPriorTraceOnReconnect requires replay=true`);
+  }
   const requiredEventTypes = asRecord(raw.requiredEventTypes, `${path}.requiredEventTypes`);
   const forbiddenRequestPaths = nonEmptyStringArrayField(raw, "forbiddenRequestPaths", path);
   if (new Set(forbiddenRequestPaths).size !== forbiddenRequestPaths.length) throw new Error(`${path}.forbiddenRequestPaths must not contain duplicates`);
@@ -1466,6 +1491,7 @@ function webProbeRealtimeFanoutProfileConfig(value: unknown, path: string): Hwla
     navigationMaxAttempts: boundedIntegerField(raw, "navigationMaxAttempts", path, 1, 10),
     navigationRetryDelayMs: boundedIntegerField(raw, "navigationRetryDelayMs", path, 100, 10_000),
     reconnectQuietMs: boundedIntegerField(raw, "reconnectQuietMs", path, 250, 30_000),
+    expectedProductSse,
     forbiddenRequestPaths,
     requiredEventTypes: {
       agentrun: agentrunEventTypes,

scripts/src/hwlab-node-web-observe-runner-realtime-source.test.ts

@@ -25,12 +25,21 @@ test("realtime fanout keeps the YAML terminal timeout when durationMs is absent"
     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"] },
     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 },
+      capabilities: { directPublish: true, liveKafkaSse: true, kafkaRefreshReplay: true },
     },
   };
 
@@ -83,3 +92,150 @@ test("realtime fanout validates stdio against the scoped AgentRun session lineag
   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("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,
+  );
+});

scripts/src/hwlab-node-web-observe-runner-realtime-source.ts

@@ -75,14 +75,13 @@ async function validateRealtimeFanout(command) {
     debug = await realtimeNewDebugContext(storageState, effective);
     const connectedA = await realtimeWaitProductConnected(subscriberA, effective.barrierTimeoutMs);
     const connectedB = await realtimeWaitProductConnected(subscriberB, effective.barrierTimeoutMs);
-    realtimeAssertLiveConnected(connectedA, sessionId, "subscriber-a", effective);
-    realtimeAssertLiveConnected(connectedB, sessionId, "subscriber-b", effective);
+    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,
-      liveOnly: true,
-      replay: false,
+      expectedProductSse: effective.expectedProductSse,
       valuesRedacted: true,
     }));
 
@@ -132,12 +131,14 @@ async function validateRealtimeFanout(command) {
 
     subscriberB2 = await realtimeNewSubscriber(storageState, sessionId, "subscriber-b-reconnected", effective);
     const bReconnected = await realtimeWaitProductConnected(subscriberB2, effective.barrierTimeoutMs);
-    realtimeAssertLiveConnected(bReconnected, sessionId, "subscriber-b-reconnected", effective);
+    realtimeAssertConnectedContract(bReconnected, sessionId, "subscriber-b-reconnected", effective, { requireRetainedEvents: effective.expectedProductSse.replayPriorTraceOnReconnect });
     await sleep(effective.reconnectQuietMs);
     const reconnectBeforeSecond = await realtimeProductSnapshot(subscriberB2);
     realtimeAssertStableProductTransport(reconnectBeforeSecond, "subscriber-b-reconnected-before-second-turn");
-    if (reconnectBeforeSecond.records.some((item) => item.traceId === first.traceId)) {
-      throw new Error("reconnected subscriber replayed first-turn events");
+    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({
@@ -159,8 +160,10 @@ async function validateRealtimeFanout(command) {
     const [secondProductA, secondProductB] = secondStable.products;
     realtimeAssertStableProductTransport(secondProductA, "subscriber-a-second-terminal");
     realtimeAssertStableProductTransport(secondProductB, "subscriber-b-reconnected-second-terminal");
-    if (secondProductB.records.some((item) => item.traceId === first.traceId)) {
-      throw new Error("reconnected subscriber received first-turn events after the second submit");
+    if (effective.expectedProductSse.replayPriorTraceOnReconnect) {
+      realtimeAssertReplayedTrace(firstDebug, secondProductB, 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,
@@ -179,6 +182,9 @@ async function validateRealtimeFanout(command) {
     const productB2 = await realtimeProductSnapshot(subscriberB2);
     realtimeAssertStableProductTransport(productA, "subscriber-a-final");
     realtimeAssertStableProductTransport(productB2, "subscriber-b-reconnected-final");
+    if (effective.expectedProductSse.replayPriorTraceOnReconnect) {
+      realtimeAssertReplayedTrace(firstDebug, productB2, first.traceId, "subscriber-b-reconnected-final");
+    }
     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");
@@ -187,14 +193,14 @@ async function validateRealtimeFanout(command) {
     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 live fanout validated",
+      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: 0,
+      replayedEventCount: bReconnected?.refreshReplay?.counts?.replayed || 0,
       valuesRedacted: true,
     });
     const subscriberAScreenshot = await realtimeCaptureSubscriberScreenshot(subscriberA.page, reportDir, "subscriber-a-final.png", {
@@ -206,9 +212,9 @@ async function validateRealtimeFanout(command) {
       valuesRedacted: true,
     });
     const subscriberB2Screenshot = await realtimeCaptureSubscriberScreenshot(subscriberB2.page, reportDir, "subscriber-b-reconnected-final.png", {
-      title: "Subscriber B reconnected live-only",
+      title: "Subscriber B reconnected through Kafka retention then live",
       sessionId,
-      traceIds: [second.traceId],
+      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,
@@ -229,11 +235,20 @@ async function validateRealtimeFanout(command) {
         lastEventIdCount: 0,
         subscriberAEventCount: productA.records.length,
         subscriberBReconnectEventCount: productB2.records.length,
-        subscriberBFirstTraceReplayCount: 0,
+        subscriberBFirstTraceReplayCount: productB2.records.filter((item) => item.traceId === first.traceId).length,
         valuesRedacted: true,
       },
-      liveContract: {
+      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),
@@ -268,7 +283,7 @@ async function validateRealtimeFanout(command) {
       commandIds: turns.map((item) => item.commandId),
       warmRunnerReused: report.warmRunnerReused,
       forbiddenRequestCount: 0,
-      replayedEventCount: 0,
+      replayedEventCount: bReconnected?.refreshReplay?.counts?.replayed || 0,
       reportPath: reportArtifact.path,
       reportSha256: reportArtifact.sha256,
       screenshots: report.screenshots,
@@ -319,11 +334,19 @@ function realtimeFanoutEffectiveProfile(profile, durationMs) {
     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)))
@@ -331,14 +354,15 @@ function realtimeFanoutEffectiveProfile(profile, durationMs) {
   return {
     ...profile,
     debugStreams,
-    barrierTimeoutMs: boundedInteger(profile.barrierTimeoutMs, 45000, 1000, 120000),
-    terminalTimeoutMs: boundedInteger(terminalTimeoutMs, 480000, 1000, 600000),
-    terminalQuietMs: boundedInteger(profile.terminalQuietMs, 2000, 250, 10000),
-    navigationMaxAttempts: boundedInteger(profile.navigationMaxAttempts, 4, 1, 10),
-    navigationRetryDelayMs: boundedInteger(profile.navigationRetryDelayMs, 1000, 100, 10000),
-    reconnectQuietMs: boundedInteger(profile.reconnectQuietMs, 3000, 250, 30000),
+    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 : {},
+    expectedProductSse: sanitize(expectedProductSse),
     expectedKafka: sanitize(expectedKafka),
   };
 }
@@ -383,6 +407,7 @@ async function realtimeNewSubscriber(storageState, sessionId, label, profile) {
         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),
@@ -480,7 +505,7 @@ async function realtimeRunTurn(input) {
     }
     for (const subscriber of input.subscribers) {
       const connected = await realtimeWaitProductConnected(subscriber, input.profile.barrierTimeoutMs);
-      realtimeAssertLiveConnected(connected, input.sessionId, subscriber.label, input.profile);
+      realtimeAssertConnectedContract(connected, input.sessionId, subscriber.label, input.profile);
     }
     await appendJsonl(files.control, eventRecord("realtime-fanout-ready-barrier", {
       turn,
@@ -587,6 +612,7 @@ async function realtimeOpenDebugStreams(debug, key, traceId, profile) {
           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",
@@ -730,15 +756,47 @@ async function realtimeCloseDebugStreams(debug, key) {
   }, key);
 }
 
-function realtimeAssertLiveConnected(connected, sessionId, label, profile) {
+function realtimeAssertConnectedContract(connected, sessionId, label, profile, options = {}) {
   if (connected?.realtimeSource !== profile.expectedKafka.topics.hwlab) throw new Error(label + " realtime source mismatch");
-  if (connected?.deliverySemantics !== "live-only" || connected?.liveOnly !== true) throw new Error(label + " is not live-only");
-  if (connected?.replay !== false || connected?.replaySupported !== false || connected?.lossPossible !== true) throw new Error(label + " replay/loss disclosure 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");
 }
 
 function realtimeConnectedSummary(connected) {
@@ -751,6 +809,7 @@ function realtimeConnectedSummary(connected) {
     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,
   };
 }
@@ -802,10 +861,19 @@ function realtimeValidateTurnEvidence(input) {
   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);
@@ -833,6 +901,7 @@ function realtimeAssertEnvelopePassthrough(kafkaRows, productRows, requireAllKaf
     && 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");
@@ -890,6 +959,9 @@ function realtimeTurnSummary(turn, debug, product, terminal) {
     };
   }
   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,
@@ -901,6 +973,15 @@ function realtimeTurnSummary(turn, debug, product, terminal) {
     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,
   };

scripts/src/hwlab-node/web-probe-observe-options.test.ts

@@ -32,6 +32,17 @@ test("validateRealtimeFanout parses the YAML profile and two independent prompts
   assert.equal(options.commandSecondText, "second turn");
   assert.deepEqual(spec.webProbe?.realtimeFanoutProfiles?.["pure-kafka-live"]?.debugStreams, ["stdio", "agentrun", "hwlab"]);
   assert.equal(spec.webProbe?.realtimeFanoutProfiles?.["pure-kafka-live"]?.fromBeginning, false);
+  assert.deepEqual(spec.webProbe?.realtimeFanoutProfiles?.["pure-kafka-live"]?.expectedProductSse, {
+    deliverySemantics: "kafka-retention-then-live",
+    liveOnly: false,
+    replay: true,
+    replaySupported: true,
+    lossPossible: false,
+    refreshHandoff: true,
+    replayPriorTraceOnReconnect: true,
+  });
+  assert.ok(spec.webProbe?.realtimeFanoutProfiles?.["pure-kafka-live"]?.requiredEventTypes.agentrun.includes("user_message"));
+  assert.ok(spec.webProbe?.realtimeFanoutProfiles?.["pure-kafka-live"]?.requiredEventTypes.hwlab.includes("user"));
 });
 
 test("validateRealtimeFanout refuses body as a substitute for the second prompt", () => {