# Effect AI Streaming

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

- Skill: `mpsuesser/effect-ai-streaming` (Agent Skill)
- Install (CLI): `npx skillmds@latest add mpsuesser/effect-ai-streaming`
- Raw SKILL.md: https://api.skillmd.com/api/skills/mpsuesser/effect-ai-streaming/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Coding & Dev Tools
- Author: mpsuesser (https://skillmd.com/u/mpsuesser)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/mpsuesser/effect-ai-streaming

---


# 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';
import * as Response from 'effect/unstable/ai/Response';
import * as SubscriptionRef from 'effect/SubscriptionRef';

const streamWithHistory = Stream.suspend(() => {
	const accumulated: Array<Response.AnyPart> = [];
	return stream.pipe(
		Stream.mapArrayEffect(
			Effect.fnUntraced(function* (parts) {
				accumulated.push(...parts);

				// Fold accumulated parts so start/delta/end IDs are visible together.
				const combined = Prompt.fromResponseParts(accumulated);
				yield* SubscriptionRef.set(history, Prompt.concat(checkpoint, combined));
				return parts;
			})
		)
	);
});
```

Key insight: `Stream.mapArrayEffect` enables side-effectful accumulation while preserving stream semantics. Its input/output batches are non-empty arrays, not v3 Chunks. Allocate mutable accumulators inside `Stream.suspend` so separate stream runs do not share history.

## 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 only the permit so release covers checkpoint setup too.
		semaphore.take(1),

		// Use: Prepare history, then stream with per-run accumulation.
		() => Stream.unwrap(Effect.gen(function* () {
			const checkpoint = Prompt.concat(yield* SubscriptionRef.get(history), newPrompt);
			yield* SubscriptionRef.set(history, checkpoint);
			const accumulated: Array<Response.AnyPart> = [];
			return LanguageModel.streamText({ prompt: checkpoint }).pipe(
				Stream.mapArrayEffect(Effect.fnUntraced(function* (parts) {
					accumulated.push(...parts);
					yield* SubscriptionRef.set(history,
						Prompt.concat(checkpoint, Prompt.fromResponseParts(accumulated)));
					return parts;
				}))
			);
		})).pipe(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`)

Keep steps 2–5 in the use phase. If checkpoint setup fails after acquisition,
the finalizer must already own the permit.

## 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, RawInput) → Prompt
// Prompt.fromResponseParts: ReadonlyArray<Response.AnyPart> → Prompt

// Pattern: checkpoint + accumulated response fold, scoped to each stream run.
const streamWithHistory = Stream.suspend(() => {
  const accumulated: Array<Response.AnyPart> = [];
  return stream.pipe(Stream.mapArrayEffect(Effect.fnUntraced(function* (parts) {
    accumulated.push(...parts);

    // Fold accumulated parts, not only this batch, so start/delta/end IDs align.
    const combined = Prompt.fromResponseParts(accumulated);
    yield* SubscriptionRef.set(history, Prompt.concat(filteredCheckpoint, combined));
    return parts;
  })));
});
```

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.
- `Prompt.fromResponseParts` routes framework-executed final results into a tool message, but keeps provider-executed final results in the assistant message. It preserves `providerExecuted` and uses `encodedResult` in both cases.
- 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

<!-- typecheck -->
```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 Channel from 'effect/Channel';
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 only the permit; all following work is covered by release.
				semaphore.take(1),
				() => Stream.unwrap(Effect.gen(function* () {
					const checkpoint = Prompt.concat(yield* SubscriptionRef.get(history), prompt);
					yield* SubscriptionRef.set(history, checkpoint);
					const accumulated: Array<Response.AnyPart> = [];
					return LanguageModel.streamText({ prompt: checkpoint }).pipe(
						Stream.mapArrayEffect(Effect.fnUntraced(function* (parts) {
							accumulated.push(...parts);
							yield* SubscriptionRef.set(history,
								Prompt.concat(checkpoint, Prompt.fromResponseParts(accumulated)));
							return parts;
						}))
					);
				})).pipe(Stream.toChannel),
				() => semaphore.release(1)
			)
		);

	return { streamText };
});

// Consume with a LanguageModel layer provided by the application.
const consume = Effect.gen(function* () {
	const chat = yield* Chat;
	yield* chat.streamText('Hello').pipe(
	Stream.runForEach((part) =>
		Match.value(part).pipe(
			Match.when({ type: 'text-delta' }, ({ delta }) =>
				Effect.logInfo(delta)
			),
			Match.when({ type: 'finish' }, ({ usage }) =>
				Effect.logDebug(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.mapArrayEffect
Stream.mapArrayEffect(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))
```

## ExecutionPlan Streaming Nuances

`Stream.withExecutionPlan(plan, { onEvent })` emits the same ordered `AttemptStart` / `AttemptSuccess` / `AttemptFailure` lifecycle as the Effect combinator. A downstream consumer that stops pulling early reports `AttemptSuccess`, because the consumer ended the attempt rather than the source failing.

Set `preventFallbackOnPartialStream: true` when a provider failure after emitted chunks must fail the stream rather than append fallback-provider output to the partial response. Lifecycle observer defects are ignored and cannot change stream attempt outcomes.

## 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.mapArrayEffect (not a side-effecting 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)