Thegreatsura
diff --git a/‎.changeset/dedupe-otel-api.md‎
Lines changed: 6 additions & 0 deletions b/‎.changeset/dedupe-otel-api.md‎
Lines changed: 6 additions & 0 deletions
diff --git a/‎docs/content/docs/v5/observability/tracing.mdx‎
Lines changed: 4 additions & 1 deletion b/‎docs/content/docs/v5/observability/tracing.mdx‎
Lines changed: 4 additions & 1 deletion
diff --git a/‎packages/core/src/readable-stream-telemetry.test.ts‎
Lines changed: 179 additions & 0 deletions b/‎packages/core/src/readable-stream-telemetry.test.ts‎
Lines changed: 179 additions & 0 deletions
@@ -0,0 +1,6 @@
+---
+'@workflow/core': patch
+'@workflow/world-vercel': patch
+---
+
+Move client-observed stream telemetry to core: `chunk_rtt` on the flush span, `connect_ms` on the read span, and new `workflow.stream.close` and `workflow.stream.read.complete` spans. Dedupe `@opentelemetry/api` to one workspace instance.
@@ -43,6 +43,8 @@ No workflow-specific configuration is required. As soon as a tracer provider and
 | `http <method>` | client | the SDK calls the workflow backend (event reads/writes) |
 | `workflow.stream.write` | client | a stream chunk (or the stream close) is flushed to the backend |
 | `workflow.stream.flush` | client | a buffered batch of stream writes settles; back-dated to the batch's first `write()`, so its duration is the app-perceived batch latency (buffer dwell + RPC) |
+| `workflow.stream.close` | client | the stream-close RPC; its duration is the close round trip |
+| `workflow.stream.read.complete` | client | a stream read drains; back-dated to the read dispatch, so its duration is the total read (`workflow.stream.read.chunks` / `.bytes` carry throughput counts) |
 | `workflow.stream.read.connect` | client | a live stream read opens; the span covers dispatch → response headers (network connect) |
 | `workflow.stream.read` | client | a live stream read receives its first chunk; the span's duration is the end-to-end time-to-first-chunk (see `workflow.stream.read.ttfc_ms`) |
 
@@ -61,9 +63,10 @@ Stream spans are emitted by the SDK's world backend on the client that writes or
 | `workflow.queue.overhead_ms` | Time between the message being enqueued and the handler starting — queue dwell plus any cold start. |
 | `workflow.stream.name` | The stream name, on stream write/read spans. |
 | `workflow.stream.operation` | The stream operation: `write`, `write_multi`, `close`, `read`, or `flush`. |
-| `workflow.stream.write.chunk_rtt` | Time between emissions of a chunk to the wire, and receiving the `ack` message for that chunk. |
+| `workflow.stream.write.chunk_rtt` | Time between emissions of a chunk to the wire, and receiving the `ack` message for that chunk. Also stamped on `workflow.stream.flush` (the batch's write RPC duration, network included). |
 | `workflow.stream.flush.buffer_dwell_ms` | On `workflow.stream.flush`: time the batch's first chunk waited in the client-side write buffer (flush timer, run-ready barrier) before the request was dispatched. `workflow.stream.flush.chunks` / `.bytes` carry the batch shape. |
 | `workflow.stream.read.ttfc_ms` | Time between opening a read connection and observing and receiving the first chunk back. |
+| `workflow.stream.read.connect_ms` | On `workflow.stream.read`: the connect portion (read dispatch → stream handle/response headers), network included. |
 
 ## Trace shape: one trace per invocation
 
 
@@ -0,0 +1,179 @@
+import { trace as otelTrace, SpanKind } from '@opentelemetry/api';
+import {
+  BasicTracerProvider,
+  InMemorySpanExporter,
+  type ReadableSpan,
+  SimpleSpanProcessor,
+} from '@opentelemetry/sdk-trace-base';
+import { SPEC_VERSION_CURRENT } from '@workflow/world';
+import {
+  afterAll,
+  afterEach,
+  beforeAll,
+  beforeEach,
+  describe,
+  expect,
+  it,
+  vi,
+} from 'vitest';
+import { setWorld } from './runtime/world.js';
+import {
+  createReconnectingFramedStream,
+  WorkflowServerReadableStream,
+} from './serialization.js';
+
+/** 4-byte BE length prefix + payload — the framed-v1 wire layout. */
+function frame(payload: Uint8Array): Uint8Array {
+  const out = new Uint8Array(4 + payload.byteLength);
+  new DataView(out.buffer).setUint32(0, payload.byteLength, false);
+  out.set(payload, 4);
+  return out;
+}
+
+const exporter = new InMemorySpanExporter();
+const provider = new BasicTracerProvider();
+
+beforeAll(() => {
+  provider.addSpanProcessor(new SimpleSpanProcessor(exporter));
+  otelTrace.setGlobalTracerProvider(provider);
+});
+
+afterAll(async () => {
+  await provider.shutdown();
+  otelTrace.disable();
+});
+
+/** The read span is emitted fire-and-forget on the first chunk; poll for it. */
+async function waitForSpans(
+  name: string,
+  count: number
+): Promise<ReadableSpan[]> {
+  for (let i = 0; i < 50; i++) {
+    const spans = exporter.getFinishedSpans().filter((s) => s.name === name);
+    if (spans.length >= count) return spans;
+    await new Promise((r) => setTimeout(r, 10));
+  }
+  return exporter.getFinishedSpans().filter((s) => s.name === name);
+}
+
+describe('WorkflowServerReadableStream read telemetry', () => {
+  beforeEach(() => {
+    exporter.reset();
+    setWorld({
+      specVersion: SPEC_VERSION_CURRENT,
+      streams: {
+        get: vi.fn().mockImplementation(
+          async () =>
+            new ReadableStream<Uint8Array>({
+              start(controller) {
+                controller.enqueue(new Uint8Array([1, 2, 3]));
+                controller.close();
+              },
+            })
+        ),
+      },
+    } as any);
+  });
+
+  afterEach(() => {
+    setWorld(undefined);
+    vi.clearAllMocks();
+  });
+
+  it('emits a workflow.stream.read span with ttfc and connect durations', async () => {
+    const stream = new WorkflowServerReadableStream('run-123', 'test-stream');
+    const reader = stream.getReader();
+    // Drain: empty header chunk, real chunk, done.
+    for (let i = 0; i < 5; i++) {
+      const { done } = await reader.read();
+      if (done) break;
+    }
+
+    const [span] = await waitForSpans('workflow.stream.read', 1);
+    expect(span).toBeDefined();
+    expect(span.kind).toBe(SpanKind.CLIENT);
+    expect(span.attributes['workflow.run.id']).toBe('run-123');
+    expect(span.attributes['workflow.stream.name']).toBe('test-stream');
+    expect(span.attributes['workflow.stream.operation']).toBe('read');
+
+    const ttfc = span.attributes['workflow.stream.read.ttfc_ms'];
+    expect(typeof ttfc).toBe('number');
+    expect(ttfc as number).toBeGreaterThanOrEqual(0);
+
+    // Client-observed connect duration (the world.streams.get await).
+    const connect = span.attributes['workflow.stream.read.connect_ms'];
+    expect(typeof connect).toBe('number');
+    expect(connect as number).toBeGreaterThanOrEqual(0);
+    expect(connect as number).toBeLessThanOrEqual((ttfc as number) + 1);
+  });
+
+  it('emits a workflow.stream.read.complete span with totals when the read drains', async () => {
+    const stream = new WorkflowServerReadableStream('run-123', 'test-stream');
+    const reader = stream.getReader();
+    for (let i = 0; i < 5; i++) {
+      const { done } = await reader.read();
+      if (done) break;
+    }
+
+    const [span] = await waitForSpans('workflow.stream.read.complete', 1);
+    expect(span).toBeDefined();
+    expect(span.kind).toBe(SpanKind.CLIENT);
+    expect(span.attributes['workflow.stream.operation']).toBe('read_complete');
+    expect(span.attributes['workflow.stream.read.chunks']).toBe(1);
+    expect(span.attributes['workflow.stream.read.bytes']).toBe(3);
+    expect(typeof span.attributes['workflow.stream.read.total_ms']).toBe(
+      'number'
+    );
+  });
+});
+
+describe('createReconnectingFramedStream read telemetry', () => {
+  beforeEach(() => {
+    exporter.reset();
+    setWorld({
+      specVersion: SPEC_VERSION_CURRENT,
+      streams: {
+        get: vi.fn().mockImplementation(
+          async () =>
+            new ReadableStream<Uint8Array>({
+              start(controller) {
+                controller.enqueue(frame(new Uint8Array([1, 2, 3])));
+                controller.enqueue(frame(new Uint8Array([4, 5])));
+                controller.close();
+              },
+            })
+        ),
+      },
+    } as any);
+  });
+
+  afterEach(() => {
+    setWorld(undefined);
+    vi.clearAllMocks();
+  });
+
+  it('emits read and read.complete spans with connect, ttfc, totals, and reconnects', async () => {
+    const stream = createReconnectingFramedStream('run-123', 'test-stream');
+    const reader = stream.getReader();
+    for (let i = 0; i < 10; i++) {
+      const { done } = await reader.read();
+      if (done) break;
+    }
+
+    const [readSpan] = await waitForSpans('workflow.stream.read', 1);
+    expect(readSpan).toBeDefined();
+    expect(typeof readSpan.attributes['workflow.stream.read.ttfc_ms']).toBe(
+      'number'
+    );
+    expect(typeof readSpan.attributes['workflow.stream.read.connect_ms']).toBe(
+      'number'
+    );
+
+    const [doneSpan] = await waitForSpans('workflow.stream.read.complete', 1);
+    expect(doneSpan).toBeDefined();
+    expect(doneSpan.attributes['workflow.stream.read.chunks']).toBe(2);
+    // 2 frames of (4-byte header + payload): (4+3) + (4+2)
+    expect(doneSpan.attributes['workflow.stream.read.bytes']).toBe(13);
+    expect(doneSpan.attributes['workflow.stream.read.reconnects']).toBe(0);
+  });
+});