---
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:

```typescript
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:

```haskell
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:

```typescript
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:

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

## Accumulation Pattern

Accumulate stream parts incrementally using mutable state for efficiency:

```typescript
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:

```typescript
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<A>, E, R>

```typescript
// 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:

```typescript
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

```typescript
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

```typescript
// ❌ 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

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

### Source Parts

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

### Metadata Parts

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

### Error Parts

```typescript
{ 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)