feat: add streaming tail consumer configuration support (#10703)

* feat: add streaming tail consumer configuration support

Implement [[streaming_tail_consumer]] configuration similar to tail_consumer,
sending to streamingTails instead of tails. Includes:

- Configuration schema and validation for streaming_tail_consumers
- Miniflare integration with streamingTails support
- Development and deployment workflow updates
- Print bindings integration for CLI output
- Comprehensive test coverage

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <[email protected]>

* fixups

* Create clever-eggs-sit.md

* fix types

* Work in Vite + Vitest

* fix tests

* Add E2E test

* Address comments

---------

Co-authored-by: Claude <[email protected]>
Co-authored-by: Samuel Macleod <[email protected]>
Co-authored-by: Somhairle MacLeòid <[email protected]>

Author: 3 people

.changeset/clever-eggs-sit.md

@@ -0,0 +1,6 @@
+---
+"miniflare": minor
+"wrangler": minor
+---
+
+Add support for streaming tail consumers in local dev. This is an experimental new feature that allows you to register a `tailStream()` handler (compared to the existing `tail()` handler), which will receeive streamed tail events from your Worker (compared to the `tail()` handler, which only receives batched events after your Worker has finished processing).

packages/miniflare/src/plugins/core/index.ts

@@ -171,6 +171,7 @@ const CoreOptionsSchemaInput = z.intersection(
 		hasAssetsAndIsVitest: z.boolean().optional(),
 
 		tails: z.array(ServiceDesignatorSchema).optional(),
+		streamingTails: z.array(ServiceDesignatorSchema).optional(),
 
 		// Strip the CF-Connecting-IP header from outbound fetches
 		stripCfConnectingIp: z.boolean().default(true),
@@ -847,19 +848,28 @@ export const CORE_PLUGIN: Plugin<
 						sharedOptions.unsafeModuleFallbackService !== undefined
 							? `localhost:${loopbackPort}`
 							: undefined,
-					tails:
-						options.tails === undefined
-							? undefined
-							: options.tails.map<ServiceDesignator>((service) => {
-									return getCustomServiceDesignator(
-										/* referrer */ options.name,
-										workerIndex,
-										CustomServiceKind.UNKNOWN,
-										name,
-										service,
-										options.hasAssetsAndIsVitest
-									);
-								}),
+					tails: options.tails?.map<ServiceDesignator>((service) => {
+						return getCustomServiceDesignator(
+							/* referrer */ options.name,
+							workerIndex,
+							CustomServiceKind.UNKNOWN,
+							name,
+							service,
+							options.hasAssetsAndIsVitest
+						);
+					}),
+					streamingTails: options.streamingTails?.map<ServiceDesignator>(
+						(service) => {
+							return getCustomServiceDesignator(
+								/* referrer */ options.name,
+								workerIndex,
+								CustomServiceKind.UNKNOWN,
+								name,
+								service,
+								options.hasAssetsAndIsVitest
+							);
+						}
+					),
 					containerEngine: getContainerEngine(options.containerEngine),
 				},
 			});
@@ -878,16 +888,24 @@ export const CORE_PLUGIN: Plugin<
 			}
 		}
 
-		if (options.tails !== undefined) {
-			for (const service of options.tails) {
-				const maybeService = maybeGetCustomServiceService(
-					workerIndex,
-					CustomServiceKind.UNKNOWN,
-					name,
-					service
-				);
-				if (maybeService !== undefined) services.push(maybeService);
-			}
+		for (const service of options.tails ?? []) {
+			const maybeService = maybeGetCustomServiceService(
+				workerIndex,
+				CustomServiceKind.UNKNOWN,
+				name,
+				service
+			);
+			if (maybeService !== undefined) services.push(maybeService);
+		}
+
+		for (const service of options.streamingTails ?? []) {
+			const maybeService = maybeGetCustomServiceService(
+				workerIndex,
+				CustomServiceKind.UNKNOWN,
+				name,
+				service
+			);
+			if (maybeService !== undefined) services.push(maybeService);
 		}
 
 		if (options.outboundService !== undefined) {

packages/miniflare/src/runtime/config/workerd.ts

@@ -73,6 +73,7 @@ export type Worker = (
 	durableObjectStorage?: Worker_DurableObjectStorage;
 	moduleFallback?: string;
 	tails?: ServiceDesignator[];
+	streamingTails?: ServiceDesignator[];
 	containerEngine?: Worker_ContainerEngine;
 };
 

packages/vite-plugin-cloudflare/src/workers/runner-worker/index.ts

@@ -46,7 +46,7 @@ interface WorkflowEntrypointConstructor<T = Cloudflare.Env> {
 }
 
 /** Keys that should be ignored during RPC property access */
-const IGNORED_KEYS = ["self", "tailStream"] as const;
+const IGNORED_KEYS = ["self"] as const;
 
 /** Available methods for `WorkerEntrypoint` class */
 const WORKER_ENTRYPOINT_KEYS = [
@@ -56,6 +56,7 @@ const WORKER_ENTRYPOINT_KEYS = [
 	"test",
 	"trace",
 	"scheduled",
+	"tailStream",
 ] as const;
 
 /** Available methods for `DurableObject` class */

packages/vitest-pool-workers/src/worker/entrypoints.ts

@@ -33,7 +33,7 @@ function importModule(
 	});
 }
 
-const IGNORED_KEYS = ["self", "tailStream"];
+const IGNORED_KEYS = ["self"];
 
 /**
  * Create a class extending `superClass` with a `Proxy` as a `prototype`.
@@ -183,6 +183,7 @@ const WORKER_ENTRYPOINT_KEYS = [
 	"scheduled",
 	"queue",
 	"test",
+	"tailStream",
 ] as const;
 const DURABLE_OBJECT_KEYS = [
 	"fetch",

packages/workers-utils/src/config/config.ts

@@ -385,6 +385,7 @@ export const defaultWranglerConfig: Config = {
 	unsafe: {},
 	mtls_certificates: [],
 	tail_consumers: undefined,
+	streaming_tail_consumers: undefined,
 	pipelines: [],
 	vpc_services: [],
 };

packages/workers-utils/src/config/environment.ts

@@ -1126,6 +1126,17 @@ export interface EnvironmentNonInheritable {
 	 */
 	tail_consumers?: TailConsumer[];
 
+	/**
+	 * Specifies a list of Streaming Tail Workers that are bound to this Worker environment
+	 *
+	 * NOTE: This field is not automatically inherited from the top level environment,
+	 * and so must be specified in every named environment.
+	 *
+	 * @default []
+	 * @nonInheritable
+	 */
+	streaming_tail_consumers?: StreamingTailConsumer[];
+
 	/**
 	 * Specifies namespace bindings that are bound to this Worker environment.
 	 *
@@ -1295,6 +1306,11 @@ export type TailConsumer = {
 	environment?: string;
 };
 
+export type StreamingTailConsumer = {
+	/** The name of the service streaming tail events will be forwarded to. */
+	service: string;
+};
+
 export interface DispatchNamespaceOutbound {
 	/** Name of the service handling the outbound requests */
 	service: string;

packages/workers-utils/src/config/validation.ts

@@ -43,6 +43,7 @@ import type {
 	Observability,
 	RawEnvironment,
 	Rule,
+	StreamingTailConsumer,
 	TailConsumer,
 } from "./environment";
 import type { TypeofType, ValidatorFn } from "./validation-helpers";
@@ -1056,6 +1057,59 @@ const validateTailConsumers: ValidatorFn = (diagnostics, field, value) => {
 	return isValid;
 };
 
+/**
+ * Streaming tail consumers should match:
+ * {
+ *   service: string,
+ *   entrypoint?: string
+ * }
+ */
+function validateStreamingTailConsumer(
+	diagnostics: Diagnostics,
+	field: string,
+	value: StreamingTailConsumer
+) {
+	if (typeof value !== "object" || value === null || Array.isArray(value)) {
+		diagnostics.errors.push(
+			`"${field}" should be an object but got ${JSON.stringify(value)}.`
+		);
+		return false;
+	}
+
+	return validateRequiredProperty(
+		diagnostics,
+		field,
+		"service",
+		value.service,
+		"string"
+	);
+}
+
+const validateStreamingTailConsumers: ValidatorFn = (
+	diagnostics,
+	field,
+	value
+) => {
+	if (!value) {
+		return true;
+	}
+	if (!Array.isArray(value)) {
+		diagnostics.errors.push(
+			`Expected "${field}" to be an array but got ${JSON.stringify(value)}.`
+		);
+		return false;
+	}
+
+	let isValid = true;
+	for (let i = 0; i < value.length; i++) {
+		isValid =
+			validateStreamingTailConsumer(diagnostics, `${field}[${i}]`, value[i]) &&
+			isValid;
+	}
+
+	return isValid;
+};
+
 /**
  * Validate top-level environment configuration and return the normalized values.
  */
@@ -1453,6 +1507,16 @@ function normalizeAndValidateEnvironment(
 			validateTailConsumers,
 			undefined
 		),
+		streaming_tail_consumers: notInheritable(
+			diagnostics,
+			topLevelEnv,
+			rawConfig,
+			rawEnv,
+			envName,
+			"streaming_tail_consumers",
+			validateStreamingTailConsumers,
+			undefined
+		),
 		unsafe: notInheritable(
 			diagnostics,
 			topLevelEnv,

packages/workers-utils/src/types.ts

@@ -165,6 +165,7 @@ type WorkerMetadataPut = {
 	logpush?: boolean;
 	placement?: CfPlacement;
 	tail_consumers?: CfTailConsumer[];
+	streaming_tail_consumers?: CfTailConsumer[];
 	limits?: CfUserLimits;
 
 	assets?: {

packages/workers-utils/src/worker.ts

@@ -453,6 +453,7 @@ export interface CfWorkerInit {
 	logpush: boolean | undefined;
 	placement: CfPlacement | undefined;
 	tail_consumers: CfTailConsumer[] | undefined;
+	streaming_tail_consumers?: CfTailConsumer[] | undefined;
 	limits: CfUserLimits | undefined;
 	annotations?: Record<string, string | undefined>;
 	keep_assets?: boolean | undefined;