effect-ai-streaming

name	effect-ai-streaming
description	Master Effect AI streaming response patterns including start/delta/end protocol, accumulation strategies, resource-safe consumption, and history management with SubscriptionRef.

Effect AI Streaming

When to Use This Skill

• Real-time streaming responses from language models
• Building chat interfaces with incremental updates
• Managing conversation history with streaming
• Protecting concurrent stream operations
• Accumulating stream parts with side effects
• Converting stream responses to prompt history

Import Patterns

CRITICAL: Always use namespace imports:

import * as Stream from 'effect/Stream';
import * as Effect from 'effect/Effect';
import * as Channel from 'effect/Channel';
import * as SubscriptionRef from 'effect/SubscriptionRef';
import * as Match from 'effect/Match';
import * as Response from 'effect/unstable/ai/Response';

StreamPart Protocol

stream := start → delta* → end

StreamPart lifecycle for each content type follows a three-phase protocol:

text      :: text-start → text-delta* → text-end
reasoning :: reasoning-start → reasoning-delta* → reasoning-end
toolParam :: tool-params-start → tool-params-delta* → tool-params-end
finish    :: { type: "finish", reason: FinishReason, usage: Usage }

Each streaming sequence has a unique id field that links start/delta/end parts.

Part Type Matching

Stream parts use a type field (not _tag), so use Match.when with type checks:

import * as Match from 'effect/Match';
import * as Effect from 'effect/Effect';

const processPart = (part: StreamPart) =>
	Match.value(part).pipe(
		Match.when({ type: 'text-delta' }, ({ delta }) =>
			Effect.sync(() => console.log(delta))
		),
		Match.when({ type: 'reasoning-delta' }, ({ delta }) =>
			Effect.sync(() => logReasoning(delta))
		),
		Match.when({ type: 'finish' }, ({ usage, reason }) =>
			Effect.sync(() => recordUsage(usage, reason))
		),
		Match.orElse(() => Effect.void)
	);

Direct type checks also work well for simple branching:

if (part.type === 'text-delta') {
	console.log(part.delta);
}

Accumulation Pattern

Accumulate stream parts incrementally using mutable state for efficiency:

import * as Stream from 'effect/Stream';
import * as Effect from 'effect/Effect';
import * as Prompt from 'effect/unstable/ai/Prompt';

const accumulated: Array<StreamPart> = [];
let combined = Prompt.empty;

stream.pipe(
	Stream.mapChunksEffect(
		Effect.fnUntraced(function* (chunk) {
			const parts = Array.from(chunk);

			// Append to mutable accumulator
			accumulated.push(...parts);

			// Fold the accumulated response so start/delta/end IDs are visible together
			combined = Prompt.fromResponseParts(accumulated);

			// Update history incrementally
			yield* SubscriptionRef.set(
				history,
				Prompt.concat(checkpoint, combined)
			);

			return chunk;
		})
	)
);

Key insight: Stream.mapChunksEffect enables side-effectful accumulation while preserving stream semantics.

Resource-Safe Streaming

Prevent concurrent stream operations using semaphore protection:

import * as Channel from 'effect/Channel';
import * as Semaphore from 'effect/Semaphore';
import * as Stream from 'effect/Stream';

const streamWithProtection = Stream.fromChannel(
	Channel.acquireUseRelease(
		// Acquire: Take semaphore, get checkpoint
		semaphore.take(1).pipe(
			Effect.zipRight(SubscriptionRef.get(history)),
			Effect.map((hist) => Prompt.concat(hist, newPrompt)),
			Effect.tap((checkpoint) => SubscriptionRef.set(history, checkpoint))
		),

		// Use: Stream with accumulation
		(checkpoint) =>
			LanguageModel.streamText({ prompt: checkpoint }).pipe(
				Stream.mapChunksEffect(accumulateAndUpdate),
				Stream.toChannel
			),

		// Release: Always release semaphore
		() => semaphore.release(1)
	)
);

Resource acquisition order:

1. Take semaphore (exclusive access)
2. Get current history snapshot
3. Merge with new prompt
4. Update history with checkpoint
5. Stream response (with incremental updates)
6. Release semaphore (guaranteed via acquireUseRelease)

Consumption Patterns

runForEach :: (A → Effect<R, E>) → Stream<A, E, R> → Effect<Unit, E, R> runDrain :: Stream<A, E, R> → Effect<Unit, E, R> runLast :: Stream<A, E, R> → Effect<Option, E, R>

// Process each part with side effects
stream.pipe(
	Stream.runForEach((part) =>
		Match.value(part).pipe(
			Match.when({ type: 'text-delta' }, ({ delta }) => updateUI(delta)),
			Match.when({ type: 'finish' }, ({ usage }) => recordMetrics(usage)),
			Match.orElse(() => Effect.void)
		)
	)
);

// Consume without collecting (memory efficient)
stream.pipe(Stream.tap(logPart), Stream.runDrain);

// Get final accumulated value
stream.pipe(
	Stream.runFold(initialState, (acc, part) => merge(acc, part)),
	Effect.map(Option.some)
);

History Update Pattern

Incremental merge strategy for conversation history:

Prompt.concat :: Prompt → Prompt → Prompt
Prompt.fromResponseParts :: Array<StreamPart> → Prompt

// Pattern: checkpoint + accumulated response fold
const accumulated: Array<StreamPart> = []
let combined = Prompt.empty

Stream.mapChunksEffect(function* (chunk) {
  const parts = Array.from(chunk)
  accumulated.push(...parts)

  // Fold accumulated parts, not only this chunk, so start/delta/end IDs align
  combined = Prompt.fromResponseParts(accumulated)

  // Update history: base checkpoint + accumulated response
  yield* SubscriptionRef.set(
    history,
    Prompt.concat(filteredCheckpoint, combined)
  )

  return chunk
})

Why checkpoint-based merging:

• Prevents re-merging entire history on each chunk
• Separates base state (checkpoint) from streaming accumulation (combined)
• Enables atomic history updates via SubscriptionRef
• Ensures Prompt.fromResponseParts sees matching start/delta/end parts for each id

Tool Streaming, Finish, and Approvals

• With automatic framework tool resolution enabled, finish is deferred until tool handler streams complete so emitted tool results appear before finish.
• tool-result parts can be preliminary or final. Use preliminary results for progress updates only; Prompt.fromResponseParts skips preliminary results and persists final results.
• Tools requiring approval emit tool-approval-request. Append a matching Prompt.toolApprovalResponsePart in a tool message and call the model again; approved/denied responses are pre-resolved into final tool results before the next provider call.
• In OpenAI-specific SSE code, unknown future events decode through OpenAiSchema.ResponseStreamEvent and are ignored by OpenAiLanguageModel; malformed known events still fail decoding.

Complete Example

import * as Prompt from 'effect/unstable/ai/Prompt';
import * as Response from 'effect/unstable/ai/Response';
import * as LanguageModel from 'effect/unstable/ai/LanguageModel';
import * as Stream from 'effect/Stream';
import * as Effect from 'effect/Effect';
import * as SubscriptionRef from 'effect/SubscriptionRef';
import * as Semaphore from 'effect/Semaphore';
import * as Match from 'effect/Match';

const Chat = Effect.gen(function* () {
	const history = yield* SubscriptionRef.make(Prompt.empty);
	const semaphore = yield* Semaphore.make(1);

	const streamText = (prompt: string) =>
		Stream.fromChannel(
			Channel.acquireUseRelease(
				// Acquire
				semaphore.take(1).pipe(
					Effect.zipRight(SubscriptionRef.get(history)),
					Effect.map((hist) =>
						Prompt.concat(hist, Prompt.make(prompt))
					),
					Effect.tap((checkpoint) => {
						combined = Prompt.empty;
						return SubscriptionRef.set(history, checkpoint);
					})
				),

				// Use
				(checkpoint) => {
					let combined = Prompt.empty;
					const accumulated: Array<Response.StreamPart> = [];

					return LanguageModel.streamText({
						prompt: checkpoint
					}).pipe(
						Stream.mapChunksEffect(
							Effect.fnUntraced(function* (chunk) {
								const parts = Array.from(chunk);
								accumulated.push(...parts);

								combined = Prompt.fromResponseParts(accumulated);

								yield* SubscriptionRef.set(
									history,
									Prompt.concat(checkpoint, combined)
								);

								return chunk;
							})
						),
						Stream.toChannel
					);
				},

				// Release
				() => semaphore.release(1)
			)
		);

	return { streamText };
});

// Consume stream
chat.streamText('Hello').pipe(
	Stream.runForEach((part) =>
		Match.value(part).pipe(
			Match.when({ type: 'text-delta' }, ({ delta }) =>
				Effect.sync(() => console.log(delta))
			),
			Match.when({ type: 'finish' }, ({ usage }) =>
				Effect.sync(() => console.log(usage))
			),
			Match.orElse(() => Effect.void)
		)
	)
);

Anti-Patterns

// ❌ Avoid Effect.either for pattern matching
Effect.either(effect).pipe(
  Effect.map((result) => result._tag === "Left" ? ... : ...)
)

// ✓ Use Effect.match
effect.pipe(
  Effect.match({
    onFailure: (error) => ...,
    onSuccess: (value) => ...
  })
)

// ❌ Using Match.tag on stream parts (stream parts use `type`, not `_tag`)
Match.value(part).pipe(Match.tag("text-delta", handler))

// ✓ Use Match.when with type checks (stream parts have `type` field, not `_tag`)
Match.value(part).pipe(Match.when({ type: "text-delta" }, handler))

// ✓ Direct type checks are also correct
if (part.type === "text-delta") { handler(part) }

// ❌ Accumulating in Stream.map (loses effects)
Stream.map((chunk) => {
  accumulated.push(...chunk) // side effect ignored
  return chunk
})

// ✓ Use Stream.mapChunksEffect
Stream.mapChunksEffect(Effect.fnUntraced(function* (chunk) {
  accumulated.push(...chunk)
  yield* updateHistory()
  return chunk
}))

Additional Stream Part Types

File Parts

{ type: "file", mediaType: "image/png", data: Uint8Array }

Source Parts

{ type: "document-source", id: string, title?: string }
{ type: "url-source", url: string, title?: string }

Metadata Parts

{ type: "response-metadata", id: string, modelId: string, timestamp: Date }

Error Parts

{ type: "error", error: AiError }
// Handle with:
Match.when({ type: "error" }, ({ error }) => Effect.fail(error))

Quality Checklist

• Use start/delta/end protocol for streaming content
• Match stream parts with Match.when({ type: ... }) or direct part.type checks (NOT Match.tag — parts use type, not _tag)
• Accumulate using Stream.mapChunksEffect (not Stream.map)
• Use SubscriptionRef for reactive history updates
• Protect concurrent streams with Semaphore
• Use Channel.acquireUseRelease for resource safety
• Handle error parts appropriately
• Checkpoint history before streaming

Related Skills

• effect-ai-language-model - streamText method that produces these streams
• effect-ai-prompt - Converting stream responses to history with fromResponseParts
• effect-ai-tool - Tool call streaming parts
• effect-ai-provider - Provider-specific streaming behavior

Reference

StreamPart types:

• text-start, text-delta, text-end - Text content streaming
• reasoning-start, reasoning-delta, reasoning-end - Chain-of-thought streaming
• tool-params-start, tool-params-delta, tool-params-end - Tool parameter streaming
• tool-call - Complete tool invocation (non-streaming)
• tool-result - Tool execution result
• finish - Stream completion with usage stats
• error - Error part

Key modules:

• effect/unstable/ai/Response - Response part schemas and constructors
• effect/unstable/ai/Prompt - Prompt construction and merging
• effect/Stream - Stream combinators (mapChunksEffect, runForEach, runDrain)
• effect/Channel - Low-level resource management (acquireUseRelease)
• effect/SubscriptionRef - Reactive shared state
• effect/Match - Pattern matching (use Match.when({ type: ... }) for stream parts)