// Process each part with side effects
stream.pipe(
  Stream.runForEach((part) =>
    Match.value(part).pipe(
      Match.tag("text-delta", ({ delta }) => updateUI(delta)),
      Match.tag("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.merge :: Prompt → Prompt → Prompt
Prompt.fromResponseParts :: Array<StreamPart> → Prompt

// Pattern: checkpoint + incremental merge
let combined = Prompt.empty

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

  // Merge new parts into combined prompt
  combined = Prompt.merge(combined, Prompt.fromResponseParts(parts))

  // Update history: base checkpoint + accumulated response
  yield* SubscriptionRef.set(
    history,
    Prompt.merge(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

Complete Example

import * as AI from "@effect/ai"
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(AI.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) => AI.Prompt.merge(hist, AI.Prompt.make(prompt))),
          Effect.tap((checkpoint) => {
            combined = AI.Prompt.empty
            return SubscriptionRef.set(history, checkpoint)
          })
        ),

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

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

              combined = AI.Prompt.merge(
                combined,
                AI.Prompt.fromResponseParts(parts)
              )

              yield* SubscriptionRef.set(
                history,
                AI.Prompt.merge(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.tag("text-delta", ({ delta }) => Effect.sync(() => console.log(delta))),
      Match.tag("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 Match.typeTags or Effect.match
effect.pipe(
  Effect.match({
    onFailure: (error) => ...,
    onSuccess: (value) => ...
  })
)

// ❌ Manual type checking (use Match.tag instead)
if (part.type === "text-delta") { ... }

// ✓ Use Match.tag or type guards
Match.value(part).pipe(Match.tag("text-delta", handler))
isTextDelta(part) ? handler(part.delta) : ...

// ❌ 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.tag("error", ({ error }) => Effect.fail(error))

Quality Checklist

• Use start/delta/end protocol for streaming content
• Match stream parts with Match.tag (not manual type checks)
• 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/ai/Response - Response part schemas and constructors
• @effect/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 on tagged types