[pull] main from vercel:main

.changeset/analytics-utc-dates.md

@@ -0,0 +1,5 @@
+---
+'@workflow/world': patch
+---
+
+Parse timezone-naive analytics timestamps as UTC so CLI and local web output shows correct times in any timezone.

.changeset/pre.json

@@ -353,6 +353,7 @@
     "stream-failure-diagnostics",
     "stream-otel-spans",
     "stream-read-v3-reconnect",
+    "stream-write-flush-span",
     "strong-cups-sip",
     "sveltekit-config-loader",
     "sveltekit-h2-noderequire",

.changeset/stream-write-flush-span.md

@@ -0,0 +1,6 @@
+---
+'@workflow/core': patch
+'@workflow/world-vercel': patch
+---
+
+Emit a client-observed `workflow.stream.flush` span per stream-write batch, with a `buffer_dwell_ms` attribute separating client-side batching cost from network/server time. Log under `DEBUG=workflow:*` when `@opentelemetry/api` fails to load in world-vercel.

docs/content/docs/v5/observability/tracing.mdx

@@ -42,6 +42,7 @@ No workflow-specific configuration is required. As soon as a tracer provider and
 | `step.execute <name>` | internal (inline) / consumer + root (queue-delivered) | a step function executes |
 | `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.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`) |
 
@@ -59,8 +60,9 @@ Stream spans are emitted by the SDK's world backend on the client that writes or
 | `workflow.trace.propagated` | Whether the invocation received trace context from the queue message. |
 | `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`, or `read`. |
+| `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.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. |
 
 ## Trace shape: one trace per invocation

packages/ai/package.json

@@ -69,6 +69,7 @@
   "devDependencies": {
     "@workflow/tsconfig": "workspace:*",
     "ai": "catalog:",
+    "typescript": "catalog:",
     "workflow": "workspace:*"
   },
   "peerDependencies": {

packages/astro/CHANGELOG.md

@@ -1,5 +1,14 @@
 # @workflow/astro
 
+## 5.0.0-beta.32
+
+### Patch Changes
+
+- Updated dependencies []:
+  - @workflow/builders@5.0.0-beta.32
+  - @workflow/rollup@5.0.0-beta.32
+  - @workflow/vite@5.0.0-beta.32
+
 ## 5.0.0-beta.31
 
 ### Patch Changes

packages/astro/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@workflow/astro",
-  "version": "5.0.0-beta.31",
+  "version": "5.0.0-beta.32",
   "description": "Astro integration for Workflow SDK",
   "type": "module",
   "main": "dist/index.js",
@@ -36,6 +36,7 @@
   "devDependencies": {
     "@types/node": "catalog:",
     "@workflow/tsconfig": "workspace:*",
-    "astro": "7.0.6"
+    "astro": "7.0.6",
+    "typescript": "catalog:"
   }
 }

packages/builders/CHANGELOG.md

@@ -1,5 +1,13 @@
 # @workflow/builders
 
+## 5.0.0-beta.32
+
+### Patch Changes
+
+- Updated dependencies [[`4a43e39`](https://github.com/vercel/workflow/commit/4a43e39fec61519a2756f4f5e7bae5ccdac6f662)]:
+  - @workflow/core@5.0.0-beta.32
+  - @workflow/world-vercel@5.0.0-beta.28
+
 ## 5.0.0-beta.31
 
 ### Patch Changes

packages/builders/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@workflow/builders",
-  "version": "5.0.0-beta.31",
+  "version": "5.0.0-beta.32",
   "description": "Shared builder infrastructure for Workflow SDK",
   "type": "module",
   "main": "./dist/index.js",
@@ -36,7 +36,8 @@
   },
   "devDependencies": {
     "@types/node": "catalog:",
-    "@workflow/tsconfig": "workspace:*"
+    "@workflow/tsconfig": "workspace:*",
+    "typescript": "catalog:"
   },
   "dependencies": {
     "@swc/core": "catalog:",

packages/cli/CHANGELOG.md

@@ -1,5 +1,15 @@
 # @workflow/cli
 
+## 5.0.0-beta.32
+
+### Patch Changes
+
+- Updated dependencies [[`4a43e39`](https://github.com/vercel/workflow/commit/4a43e39fec61519a2756f4f5e7bae5ccdac6f662)]:
+  - @workflow/core@5.0.0-beta.32
+  - @workflow/world-vercel@5.0.0-beta.28
+  - @workflow/builders@5.0.0-beta.32
+  - @workflow/web@5.0.0-beta.32
+
 ## 5.0.0-beta.31
 
 ### Patch Changes

packages/cli/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@workflow/cli",
-  "version": "5.0.0-beta.31",
+  "version": "5.0.0-beta.32",
   "description": "Command-line interface for Workflow SDK",
   "type": "module",
   "bin": {
@@ -37,7 +37,8 @@
   },
   "devDependencies": {
     "@types/node": "catalog:",
-    "@workflow/tsconfig": "workspace:*"
+    "@workflow/tsconfig": "workspace:*",
+    "typescript": "catalog:"
   },
   "dependencies": {
     "@oclif/core": "4.11.4",

packages/core/CHANGELOG.md

@@ -1,5 +1,14 @@
 # @workflow/core
 
+## 5.0.0-beta.32
+
+### Patch Changes
+
+- [#2891](https://github.com/vercel/workflow/pull/2891) [`4a43e39`](https://github.com/vercel/workflow/commit/4a43e39fec61519a2756f4f5e7bae5ccdac6f662) Thanks [@karthikscale3](https://github.com/karthikscale3)! - Emit a client-observed `workflow.stream.flush` span per stream-write batch, with a `buffer_dwell_ms` attribute separating client-side batching cost from network/server time. Log under `DEBUG=workflow:*` when `@opentelemetry/api` fails to load in world-vercel.
+
+- Updated dependencies [[`4a43e39`](https://github.com/vercel/workflow/commit/4a43e39fec61519a2756f4f5e7bae5ccdac6f662)]:
+  - @workflow/world-vercel@5.0.0-beta.28
+
 ## 5.0.0-beta.31
 
 ### Minor Changes

packages/core/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@workflow/core",
-  "version": "5.0.0-beta.31",
+  "version": "5.0.0-beta.32",
   "description": "Core runtime and engine for Workflow SDK",
   "type": "module",
   "main": "dist/index.js",
@@ -126,7 +126,8 @@
     "@types/semver": "7.7.1",
     "@workflow/tsconfig": "workspace:*",
     "cross-env": "10.1.0",
-    "genversion": "3.2.0"
+    "genversion": "3.2.0",
+    "typescript": "catalog:"
   },
   "peerDependencies": {
     "@opentelemetry/api": "1"

packages/core/src/serialization.ts

@@ -871,6 +871,39 @@ const getStreamFlushIntervalMs = (): number =>
     integer: true,
   });
 
+/**
+ * Emit the client-observed span for one flushed batch of stream writes: first
+ * `write()` of the batch (`startEpochMs`) → the server write settling. The
+ * span's duration is therefore the app-perceived latency of the batch
+ * (buffer dwell + backpressure + RPC); `buffer_dwell_ms` isolates the
+ * pre-dispatch share so client-side batching cost can be told apart from
+ * network/server time. Named `workflow.stream.flush` — the per-request RPC
+ * beneath it is world-vercel's `workflow.stream.write` span (chunk_rtt), and
+ * the two must stay distinguishable. Fire-and-forget; no-op without OTEL.
+ */
+function recordStreamWriteFlush(
+  startEpochMs: number,
+  dispatchEpochMs: number,
+  runId: string,
+  name: string,
+  chunkCount: number,
+  byteCount: number
+): void {
+  void (async () => {
+    await recordElapsedSpan('workflow.stream.flush', startEpochMs, {
+      kind: await getSpanKind('CLIENT'),
+      attributes: {
+        'workflow.run.id': runId,
+        'workflow.stream.name': name,
+        'workflow.stream.operation': 'flush',
+        'workflow.stream.flush.buffer_dwell_ms': dispatchEpochMs - startEpochMs,
+        'workflow.stream.flush.chunks': chunkCount,
+        'workflow.stream.flush.bytes': byteCount,
+      },
+    });
+  })();
+}
+
 export class WorkflowServerWritableStream extends WritableStream<Uint8Array> {
   /**
    * @param runReadyBarrier Turbo mode only: a promise that resolves once the
@@ -915,6 +948,12 @@ export class WorkflowServerWritableStream extends WritableStream<Uint8Array> {
     let flushTimer: ReturnType<typeof setTimeout> | null = null;
     let flushPromise: Promise<void> | null = null;
     let resolvedFlushIntervalMs: number | undefined;
+    // Client-observed write-batch timing: stamped at `write()` entry for the
+    // first chunk of a batch — before the backpressure wait on any in-flight
+    // flush — so the emitted span covers the full app-perceived latency
+    // (queueing + flush-timer dwell + RPC). Cleared when a flush takes the
+    // batch; restored on write failure so a retried batch keeps its true t0.
+    let batchStartAt: number | undefined;
 
     const flush = async (): Promise<void> => {
       if (flushTimer) {
@@ -931,24 +970,51 @@ export class WorkflowServerWritableStream extends WritableStream<Uint8Array> {
       // Copy chunks to flush, but don't clear buffer until write succeeds
       // This prevents data loss if the write operation fails
       const chunksToFlush = buffer.slice();
+      const batchStart = batchStartAt;
+      batchStartAt = undefined;
+      const dispatchAt = Date.now();
 
       const world = await worldPromise;
       // Cache the flush interval from the world on first use
       if (resolvedFlushIntervalMs === undefined) {
         resolvedFlushIntervalMs =
           world.streamFlushIntervalMs ?? getStreamFlushIntervalMs();
       }
-      // Use writeMulti if available for batch writes
-      if (
-        typeof world.streams.writeMulti === 'function' &&
-        chunksToFlush.length > 1
-      ) {
-        await world.streams.writeMulti(runId, name, chunksToFlush);
-      } else {
-        // Fall back to sequential writes
-        for (const chunk of chunksToFlush) {
-          await world.streams.write(runId, name, chunk);
+      try {
+        // Use writeMulti if available for batch writes
+        if (
+          typeof world.streams.writeMulti === 'function' &&
+          chunksToFlush.length > 1
+        ) {
+          await world.streams.writeMulti(runId, name, chunksToFlush);
+        } else {
+          // Fall back to sequential writes
+          for (const chunk of chunksToFlush) {
+            await world.streams.write(runId, name, chunk);
+          }
         }
+      } catch (error) {
+        // The batch stays buffered for retry — restore its original t0 (the
+        // oldest, if a newer write stamped one meanwhile) so the eventually
+        // successful flush reports the full dwell.
+        if (batchStart !== undefined) {
+          batchStartAt =
+            batchStartAt === undefined
+              ? batchStart
+              : Math.min(batchStartAt, batchStart);
+        }
+        throw error;
+      }
+
+      if (batchStart !== undefined) {
+        recordStreamWriteFlush(
+          batchStart,
+          dispatchAt,
+          runId,
+          name,
+          chunksToFlush.length,
+          chunksToFlush.reduce((sum, c) => sum + c.byteLength, 0)
+        );
       }
 
       // Only clear buffer after successful write to prevent data loss
@@ -981,6 +1047,11 @@ export class WorkflowServerWritableStream extends WritableStream<Uint8Array> {
 
     super({
       async write(chunk) {
+        // Batch t0 for the write-flush span: at entry, before the
+        // backpressure wait below, so queueing behind an in-flight flush
+        // counts toward the app-perceived dwell.
+        if (batchStartAt === undefined) batchStartAt = Date.now();
+
         // Wait for any in-progress flush to complete before adding to buffer
         if (flushPromise) {
           await flushPromise;