[web-shared] Mark ignored duplicate events in the observability UI

.changeset/silly-pears-jam.md

@@ -0,0 +1,5 @@
+---
+'@workflow/web-shared': patch
+---
+
+Mark ignored duplicate events in the observability UI and, when the whole event log is loaded, exclude them from derived step status, durations, and trace spans

packages/core/src/duplicate-event-fixtures.test.ts

@@ -0,0 +1,120 @@
+import { type Event, entityEventClass } from '@workflow/world';
+import { DUPLICATE_EVENT_FIXTURES } from '@workflow/world/test-support/duplicate-event-fixtures.js';
+import { describe, expect, it, vi } from 'vitest';
+import { EventConsumerResult, EventsConsumer } from './events-consumer.js';
+
+/**
+ * The runtime's half of {@link DUPLICATE_EVENT_FIXTURES}. The observability
+ * UI's half runs the same fixtures through its own classifier, so a fixture
+ * whose expectation moves fails on both sides.
+ *
+ * What this drives is the walk in `EventsConsumer`: given consumers that claim
+ * their entity's events for as long as the entity is open, which events does
+ * it step over. The claim that the consumers behave that way is what
+ * `duplicate-events.test.ts` checks, against the real step and sleep
+ * primitives.
+ */
+
+/** No deliveries are modeled here, so the delivery gate is always open. */
+const OPEN_GATE = {
+  getPromiseQueue: () => Promise.resolve(),
+  isDeliveryIdle: () => true,
+};
+
+/** Classes whose event deregisters its entity's consumer. */
+const TERMINAL_CLASSES = new Set([
+  'step_terminal',
+  'wait_completed',
+  'hook_disposed',
+]);
+
+/**
+ * One callback standing in for every consumer a replay registers.
+ *
+ * `step.ts` keeps a step's consumer alive from `step_created` until the step's
+ * outcome, claiming each attempt's `step_started` and `step_retrying` on the
+ * way; `sleep.ts` does the same for a wait. Both delete the queue item on the
+ * entity's terminal event, after which nothing claims that correlation id.
+ * `workflow.ts` declines a second `run_started` outright.
+ */
+function replayConsumers(): (event: Event | null) => EventConsumerResult {
+  const claimed = new Set<string>();
+  const closed = new Set<string>();
+
+  return (event) => {
+    if (event === null) return EventConsumerResult.NotConsumed;
+
+    const eventClass = entityEventClass(event.eventType);
+    // Belongs to no class: a delivery, or an event that precedes every replay.
+    // Their consumers subscribe lazily and take every copy.
+    if (eventClass === undefined) return EventConsumerResult.Consumed;
+
+    const entity = event.correlationId ?? '';
+    if (closed.has(entity)) return EventConsumerResult.NotConsumed;
+
+    const classKey = `${eventClass}:${entity}`;
+    if (eventClass === 'run_started' && claimed.has(classKey)) {
+      return EventConsumerResult.NotConsumed;
+    }
+
+    claimed.add(classKey);
+    if (TERMINAL_CLASSES.has(eventClass)) closed.add(entity);
+    return EventConsumerResult.Consumed;
+  };
+}
+
+function buildLog(fixture: (typeof DUPLICATE_EVENT_FIXTURES)[number]): Event[] {
+  return fixture.events.map(
+    (spec, index) =>
+      ({
+        eventId: `evnt_${String(index).padStart(26, '0')}`,
+        runId: 'wrun_test',
+        eventType: spec.eventType,
+        correlationId: spec.entity,
+        eventData: {},
+        createdAt: new Date(),
+      }) as unknown as Event
+  );
+}
+
+/**
+ * Resolve once the walk has decided every event: it reached the end of the
+ * log, or it stopped on one nothing can claim, which is the divergence the
+ * skip exists to tell apart from a repeat.
+ */
+async function settle(
+  consumer: EventsConsumer,
+  length: number,
+  onUnconsumedEvent: { mock: { calls: unknown[] } }
+) {
+  const deadline = Date.now() + 5000;
+  while (
+    consumer.eventIndex < length &&
+    onUnconsumedEvent.mock.calls.length === 0 &&
+    Date.now() < deadline
+  ) {
+    await new Promise((resolve) => setTimeout(resolve, 10));
+  }
+}
+
+describe('shared duplicate-event fixtures', () => {
+  for (const fixture of DUPLICATE_EVENT_FIXTURES) {
+    it(`steps over the right events: ${fixture.name}`, async () => {
+      const events = buildLog(fixture);
+      const onDuplicateEvent = vi.fn();
+      const onUnconsumedEvent = vi.fn();
+      const consumer = new EventsConsumer(events, {
+        ...OPEN_GATE,
+        onUnconsumedEvent,
+        onDuplicateEvent,
+      });
+
+      consumer.subscribe(replayConsumers());
+      await settle(consumer, events.length, onUnconsumedEvent);
+
+      expect(onDuplicateEvent.mock.calls.map(([event]) => event)).toEqual(
+        fixture.ignoredIndices.map((index) => events[index])
+      );
+    });
+  }
+});

packages/web-shared/src/components/event-list-view.tsx

@@ -10,6 +10,7 @@ import type {
 } from 'react';
 import { useCallback, useEffect, useMemo, useRef, useState } from 'react';
 import { Virtuoso, type VirtuosoHandle } from 'react-virtuoso';
+import { findDuplicateEventIds } from '../lib/duplicate-events';
 import {
   type ExactIdSearchResult,
   type ExactWorkflowSearchIdKind,
@@ -23,6 +24,7 @@ import { AttrSetEventBlock } from './sidebar/attributes-block';
 import { ContextCardProvider } from './ui/context-card';
 import { DataInspector, DecryptClickContext } from './ui/data-inspector';
 import { DecryptButton } from './ui/decrypt-button';
+import { DuplicateEventTooltip } from './ui/duplicate-event-tooltip';
 import {
   ErrorStackBlock,
   isStructuredError,
@@ -190,15 +192,24 @@ export interface DurationInfo {
  * Build a map from correlationId → duration info by diffing
  * created ↔ started (queued) and started ↔ completed/failed/cancelled (ran).
  * Also computes run-level durations under the key '__run__'.
+ *
+ * Events every replay reads past as repeats are excluded: a second
+ * `step_completed` written by a concurrent replay would otherwise stretch the
+ * step's measured runtime to whenever that replay happened to commit. The
+ * caller supplies them, because whether an event is a repeat is a property of
+ * the whole log and this function may be handed a page of it.
  */
-export function buildDurationMap(events: Event[]): Map<string, DurationInfo> {
+export function buildDurationMap(
+  events: Event[],
+  duplicateEventIds: ReadonlySet<string> = new Set()
+): Map<string, DurationInfo> {
   // Process events in chronological order so the result doesn't depend on
   // the caller's sort direction. Retried steps emit multiple `step_started`
   // events for the same correlationId; the queued duration must be measured
   // against the first one, not the last.
-  const chronological = [...events].sort(
-    (a, b) => getEffectiveEventTime(a) - getEffectiveEventTime(b)
-  );
+  const chronological = [...events]
+    .filter((event) => !duplicateEventIds.has(event.eventId))
+    .sort((a, b) => getEffectiveEventTime(a) - getEffectiveEventTime(b));
 
   const createdTimes = new Map<string, number>();
   const firstStartedTimes = new Map<string, number>();
@@ -832,6 +843,7 @@ export function EventRow({
   onEncryptedDataDetected,
   suppressGroupDimming = false,
   showSeparateEventOccurrenceTimestamps = false,
+  isDuplicate = false,
 }: {
   event: Event;
   index: number;
@@ -856,6 +868,8 @@ export function EventRow({
   suppressGroupDimming?: boolean;
   /** Show occurredAt separately instead of folding it into the Created timestamp. */
   showSeparateEventOccurrenceTimestamps?: boolean;
+  /** The event repeats a class already in the log, so the runtime ignored it. */
+  isDuplicate?: boolean;
 }) {
   const [isLoading, setIsLoading] = useState(false);
   const [loadedEventData, setLoadedEventData] = useState<unknown | null>(
@@ -1095,43 +1109,49 @@ export function EventRow({
 
           {/* Event Type */}
           <div className="font-medium min-w-0 px-4" style={{ flex: '2 1 0%' }}>
-            <span
-              className="inline-flex items-center gap-1.5"
-              style={{ color: 'var(--ds-gray-900)' }}
-            >
+            <DuplicateEventTooltip isDuplicate={isDuplicate}>
               <span
+                className="inline-flex items-center gap-1.5"
                 style={{
-                  position: 'relative',
-                  display: 'inline-flex',
-                  width: 6,
-                  height: 6,
-                  flexShrink: 0,
+                  color: isDuplicate
+                    ? 'var(--ds-gray-700)'
+                    : 'var(--ds-gray-900)',
                 }}
               >
-                {isPulsing && (
-                  <span
-                    style={{
-                      position: 'absolute',
-                      inset: 0,
-                      borderRadius: '50%',
-                      backgroundColor: statusDotColor,
-                      opacity: 0.75,
-                      animation: DOT_PULSE_ANIMATION,
-                    }}
-                  />
-                )}
                 <span
                   style={{
                     position: 'relative',
+                    display: 'inline-flex',
                     width: 6,
                     height: 6,
-                    borderRadius: '50%',
-                    backgroundColor: statusDotColor,
+                    flexShrink: 0,
                   }}
-                />
+                >
+                  {isPulsing && (
+                    <span
+                      style={{
+                        position: 'absolute',
+                        inset: 0,
+                        borderRadius: '50%',
+                        backgroundColor: statusDotColor,
+                        opacity: 0.75,
+                        animation: DOT_PULSE_ANIMATION,
+                      }}
+                    />
+                  )}
+                  <span
+                    style={{
+                      position: 'relative',
+                      width: 6,
+                      height: 6,
+                      borderRadius: '50%',
+                      backgroundColor: statusDotColor,
+                    }}
+                  />
+                </span>
+                {formatEventType(event.eventType)}
               </span>
-              {formatEventType(event.eventType)}
-            </span>
+            </DuplicateEventTooltip>
           </div>
 
           {/* Name */}
@@ -1310,6 +1330,23 @@ function EventListViewInner({
     );
   }, [events, effectiveSortOrder, isExactSearchActive, searchResults]);
 
+  // Events every replay reads past as repeats. Computed from the source list
+  // rather than `sortedEvents` because which occurrence counted is a property
+  // of the log, not of the direction the table happens to be sorted in.
+  //
+  // A page short of the whole log, or an exact-ID search that returns one
+  // event, cannot answer that question: the event a repeat lost to may be
+  // outside the window, and reading it the other way round would mark the
+  // event the run acted on and drop it from the durations. Both cases classify
+  // nothing.
+  const duplicateEventIds = useMemo(
+    () =>
+      findDuplicateEventIds(events ?? [], {
+        isCompleteHistory: !hasMoreEvents && !isExactSearchActive,
+      }),
+    [events, hasMoreEvents, isExactSearchActive]
+  );
+
   // Detect encrypted fields across all loaded events (inline eventData).
   const hasEncryptedInlineData = useMemo(() => {
     const sourceEvents = isExactSearchActive ? searchResults : events;
@@ -1342,8 +1379,8 @@ function EventListViewInner({
   );
 
   const durationMap = useMemo(
-    () => buildDurationMap(sortedEvents),
-    [sortedEvents]
+    () => buildDurationMap(sortedEvents, duplicateEventIds),
+    [sortedEvents, duplicateEventIds]
   );
 
   const [selectedGroupKey, setSelectedGroupKey] = useState<string | undefined>(
@@ -1804,6 +1841,7 @@ function EventListViewInner({
                   encryptionKey={encryptionKey}
                   onEncryptedDataDetected={handleEncryptedDataDetected}
                   suppressGroupDimming={isExactSearchActive}
+                  isDuplicate={duplicateEventIds.has(ev.eventId)}
                   showSeparateEventOccurrenceTimestamps={
                     showSeparateEventOccurrenceTimestamps
                   }

packages/web-shared/src/components/sidebar/entity-detail-panel.tsx

@@ -47,6 +47,11 @@ export interface SelectedSpanInfo {
   spanId?: string;
   /** Raw correlated events from the store (NOT from the trace worker pipeline) */
   rawEvents?: Event[];
+  /**
+   * Events every replay reads past as repeats, computed from the whole log.
+   * `rawEvents` is one entity's slice, which cannot answer that on its own.
+   */
+  duplicateEventIds?: ReadonlySet<string>;
 }
 
 /**
@@ -402,6 +407,7 @@ export function EntityDetailPanel({
               showSeparateEventOccurrenceTimestamps={
                 showSeparateEventOccurrenceTimestamps
               }
+              duplicateEventIds={selectedSpan?.duplicateEventIds}
             />
           )}
         </div>