# Error Handling

A stream's failure type `E` is part of its signature, just like an `Effect`'s.
When a stream fails, every downstream operator short-circuits and the failure
surfaces wherever you run it. Effect gives you the same recovery vocabulary you
use elsewhere — switch to a fallback stream, match on a specific tagged error,
retry on a schedule, or run cleanup on failure — all expressed *inside* the
stream so the result type reflects what can still go wrong.

```ts
import { Effect, Stream } from "effect"

// A stream that emits a couple of values and then fails.
const flaky = Stream.make(1, 2, 3).pipe(
  Stream.concat(Stream.fail("boom" as const)),
  Stream.concat(Stream.make(4, 5))
)

const fallback = Stream.make(10, 20, 30)

// `Stream.catch` is the catch-all: on any failure it switches to the stream you
// return. Values emitted *before* the failure are kept; the rest come from the
// fallback. The recovered stream's error channel is now `never`.
export const recovered = flaky.pipe(
  Stream.catch(() => fallback),
  Stream.runCollect
)
// => [1, 2, 3, 10, 20, 30]
```
**Note:** In v4 the catch-all on streams is `Stream.catch` (not `catchAll`), mirroring
  `Effect.catch`. For defects and interruptions, use `Stream.catchCause`, which
  receives the full `Cause` rather than just the typed error.

## Recovering from specific errors

When a stream can fail with several tagged errors, recover from each one
individually with `Stream.catchTag` (or `catchTags` for several at once). This
keeps the other error types in the signature so the compiler still forces you to
handle them.

```ts
import { Effect, Schema, Stream } from "effect"

class NotFound extends Schema.TaggedErrorClass<NotFound>()("NotFound", {
  id: Schema.String
}) {}

class RateLimited extends Schema.TaggedErrorClass<RateLimited>()("RateLimited", {
  retryAfterMillis: Schema.Number
}) {}

declare const records: Stream.Stream<string, NotFound | RateLimited>

// Handle `NotFound` by substituting an empty stream; `RateLimited` still
// propagates, so the result type is `Stream<string, RateLimited>`.
export const handled = records.pipe(
  Stream.catchTag("NotFound", () => Stream.empty)
)

// Handle several tags at once with a record of handlers.
export const handledBoth = records.pipe(
  Stream.catchTags({
    NotFound: () => Stream.empty,
    RateLimited: (e) => Stream.fromEffect(
      Effect.as(Effect.sleep(e.retryAfterMillis), "retried-placeholder")
    )
  })
)
```

For richer matching there is also `Stream.catchIf` (recover when a predicate or
refinement matches) and `Stream.catchCause` (inspect the full `Cause`, including
defects). See [Error Management](https://effect.plants.sh/error-management/) for how tagged errors,
`Cause`, and these combinators relate across all of Effect.

## Retrying a failing stream

Transient failures — a flaky socket, a rate-limited API — are best handled by
*retrying* rather than giving up. `Stream.retry` takes a
[Schedule](https://effect.plants.sh/scheduling/schedule/) and restarts the stream from the beginning
each time it fails, according to the policy.

```ts
import { Effect, Schedule, Stream } from "effect"

const readSensor = Effect.fn("readSensor")(function*() {
  // Imagine an effect that occasionally fails to read a hardware sensor.
  return yield* Effect.fail("sensor unavailable" as const)
})

// Retry with exponential backoff, giving up after 5 attempts. `Schedule.both`
// combines the two policies with AND semantics: it keeps recurring only while
// *both* the exponential backoff and the `recurs(5)` cap still want to recur.
export const sensorStream = Stream.fromEffect(readSensor()).pipe(
  Stream.retry(Schedule.exponential("100 millis").pipe(
    Schedule.both(Schedule.recurs(5))
  ))
)
```
**Caution:** `Stream.retry` re-runs the stream from its source on each attempt, so it
  re-executes all of the stream's acquire operations. For an unbounded source
  this restarts iteration; for a one-shot effect it simply re-runs the effect.
  Bound the policy (e.g. with `Schedule.recurs`) so a permanently-broken source
  does not retry forever.

## Cleanup and timeouts

`Stream.onError` registers a finalizer that runs if (and only if) the stream
fails, receiving the `Cause`. It does not change the failure — use it for
cleanup, logging, or metrics, not recovery.

```ts
import { Cause, Effect, Stream } from "effect"

const program = Stream.make(1, 2, 3).pipe(
  Stream.concat(Stream.fail("oops" as const)),
  // Runs on failure; the original error still propagates afterwards.
  Stream.onError((cause) =>
    Effect.logError(`stream failed: ${Cause.pretty(cause)}`)
  )
)
```

Two operators guard against streams that *stall* rather than fail. `Stream.timeout`
ends the stream if no value is produced within a duration; `Stream.timeoutOrElse`
switches to a fallback stream instead.

```ts
import { Effect, Stream } from "effect"

const slow: Stream.Stream<number> = Stream.fromEffect(
  Effect.as(Effect.never, 0)
)

// End the stream (emit nothing more) if no element arrives within 2 seconds.
export const bounded = slow.pipe(Stream.timeout("2 seconds"))

// Or switch to a fallback stream after the timeout.
export const orFallback = slow.pipe(
  Stream.timeoutOrElse({
    duration: "2 seconds",
    orElse: () => Stream.make(1, 2, 3)
  })
)
```

Together these let a long-running pipeline degrade gracefully: retry the
transient failures, time out the stalls, fall back where recovery makes sense,
and clean up on the paths that cannot recover.

## Recovery reference

Every recovery and lifecycle combinator on `Stream`, grouped by purpose. Each
returns a new stream; they compose with `.pipe(...)`.

### Catching

#### `Stream.catch`

Switches to the stream returned by `f` on any *typed* failure. The recovered
error channel is whatever the fallback stream contributes.

```ts
import { Effect, Stream } from "effect"

const program = Stream.make(1, 2).pipe(
  Stream.concat(Stream.fail("Oops!")),
  Stream.catch(() => Stream.make(999)),
  Stream.runCollect
)
// => [1, 2, 999]
```

#### `Stream.catchCause`

Recovers from the full `Cause`, so it sees defects and interruptions as well as
typed failures. Use this when a fallback should fire on *any* abnormal exit.

```ts
import { Effect, Stream } from "effect"

const program = Stream.make(1, 2).pipe(
  Stream.concat(Stream.fail("Oops!")),
  Stream.catchCause(() => Stream.make(999)),
  Stream.runCollect
)
// => [1, 2, 999]
```

#### `Stream.catchIf`

Recovers only when a predicate (or type refinement) matches the error.
Non-matching failures propagate, so the error type is preserved unless the
refinement narrows it. An optional third handler covers the non-matching case.

```ts
import { Effect, Stream } from "effect"

const program = Stream.make(1, 2).pipe(
  Stream.concat(Stream.fail(42)),
  Stream.catchIf(
    (error): error is 42 => error === 42,
    () => Stream.make(999)
  ),
  Stream.runCollect
)
// => [1, 2, 999]
```

#### `Stream.catchFilter`

Recovers using a reusable `Filter`, which can also narrow or transform the error
before choosing the recovery stream. Successful filter results go to `f`; failed
results re-fail (or go to the optional `orElse`).

```ts
import { Effect, Filter, Stream } from "effect"

// Filter keeps only negative numbers, passing them to the handler.
const negatives = Filter.fromPredicate((n: number) => n < 0)

const program = Stream.make(1, 2).pipe(
  Stream.concat(Stream.fail(-5)),
  Stream.catchFilter(negatives, (n) => Stream.make(`recovered ${n}`)),
  Stream.runCollect
)
// => [1, 2, "recovered -5"]
```

#### `Stream.catchTag`

Recovers from failures whose `_tag` matches the given value, switching to the
stream returned by the handler. Other tags stay in the error channel. Pass an
array of tags to match several at once.

```ts
import { Data, Effect, Stream } from "effect"

class HttpError extends Data.TaggedError("HttpError")<{ message: string }> {}

const program = Stream.fail(new HttpError({ message: "timeout" })).pipe(
  Stream.catchTag("HttpError", (e) => Stream.make(`Recovered: ${e.message}`)),
  Stream.runCollect
)
// => ["Recovered: timeout"]
```

#### `Stream.catchTags`

Recovers from multiple tagged failures with a record of `_tag`-keyed handlers.
Tags you omit stay in the error channel.

```ts
import { Effect, Stream } from "effect"

class NotFound {
  readonly _tag = "NotFound"
  constructor(readonly resource: string) {}
}
class Unauthorized {
  readonly _tag = "Unauthorized"
  constructor(readonly user: string) {}
}

const program = Stream.fail(new NotFound("profile")).pipe(
  Stream.catchTags({
    NotFound: () => Stream.succeed("fallback"),
    Unauthorized: () => Stream.succeed("login")
  }),
  Stream.runCollect
)
// => ["fallback"]
```

#### `Stream.catchReason`

Catches a specific nested *reason* inside a tagged error without removing the
parent error from the channel. The handler receives the unwrapped reason. Pass
the parent error tag, then the reason tag, then the handler.

```ts
import { Data, Effect, Stream } from "effect"

class RateLimitError extends Data.TaggedError("RateLimitError")<{
  retryAfter: number
}> {}
class AiError extends Data.TaggedError("AiError")<{
  reason: RateLimitError
}> {}

const program = Stream.fail(
  new AiError({ reason: new RateLimitError({ retryAfter: 60 }) })
).pipe(
  Stream.catchReason("AiError", "RateLimitError", (reason) =>
    Stream.succeed(`retry: ${reason.retryAfter}`)
  ),
  Stream.runCollect
)
// => ["retry: 60"]
```

#### `Stream.catchReasons`

Catches several reasons within one tagged error using an object of handlers
keyed by reason tag.

```ts
import { Data, Effect, Stream } from "effect"

class RateLimitError extends Data.TaggedError("RateLimitError")<{
  retryAfter: number
}> {}
class QuotaExceededError extends Data.TaggedError("QuotaExceededError")<{
  limit: number
}> {}
class AiError extends Data.TaggedError("AiError")<{
  reason: RateLimitError | QuotaExceededError
}> {}

const program = Stream.fail(
  new AiError({ reason: new RateLimitError({ retryAfter: 60 }) })
).pipe(
  Stream.catchReasons("AiError", {
    RateLimitError: (reason) => Stream.succeed(`retry: ${reason.retryAfter}`),
    QuotaExceededError: (reason) => Stream.succeed(`quota: ${reason.limit}`)
  }),
  Stream.runCollect
)
// => ["retry: 60"]
```

#### `Stream.catchCauseIf`

Recovers from a failure when a predicate over the full `Cause` matches.
Non-matching causes are re-emitted as failures, so the error type is preserved.

```ts
import { Cause, Effect, Stream } from "effect"

const program = Stream.fail("NetworkError").pipe(
  Stream.catchCauseIf(
    (cause) => Cause.hasFails(cause),
    (cause) => Stream.make(`Recovered: ${Cause.squash(cause)}`)
  ),
  Stream.runCollect
)
// => ["Recovered: NetworkError"]
```

#### `Stream.catchCauseFilter`

Recovers from causes selected by a `Filter` applied to the full `Cause`. The
handler receives both the filtered value and the original cause; a failed filter
result re-fails with the residual cause.

```ts
import { Cause, Effect, Filter, Stream } from "effect"

// Keep only causes that contain a typed failure.
const onlyFails = Filter.fromPredicate(Cause.hasFails)

const program = Stream.fail("NetworkError").pipe(
  Stream.catchCauseFilter(onlyFails, (cause) =>
    Stream.make(`Recovered: ${Cause.squash(cause)}`)
  ),
  Stream.runCollect
)
// => ["Recovered: NetworkError"]
```

### Mapping and observing errors

#### `Stream.mapError`

Transforms the typed error with `f` without recovering. Useful for wrapping
low-level errors in a domain error before they reach a boundary.

```ts
import { Effect, Stream } from "effect"

const program = Stream.fail("bad").pipe(
  Stream.mapError((error) => `mapped: ${error}`),
  Stream.catch((error) => Stream.make(`recovered from ${error}`)),
  Stream.runCollect
)
// => ["recovered from mapped: bad"]
```

#### `Stream.tapError`

Runs an effect when the stream fails, peeking at the typed error without
changing the values or error — unless the tap effect itself fails.

```ts
import { Console, Effect, Stream } from "effect"

const program = Stream.make(1, 2).pipe(
  Stream.concat(Stream.fail("boom")),
  Stream.tapError((error) => Console.log(`tapError: ${error}`)),
  Stream.catch(() => Stream.make(999)),
  Stream.runCollect
)
// logs: tapError: boom
// => [1, 2, 999]
```

#### `Stream.tapCause`

Like `tapError`, but the tap effect receives the full `Cause`, so it also fires
on defects and interruptions.

```ts
import { Cause, Effect, Stream } from "effect"

const program = Stream.make(1, 2).pipe(
  Stream.concat(Stream.fail("boom")),
  Stream.tapCause((cause) => Effect.logError(Cause.pretty(cause))),
  Stream.catch(() => Stream.succeed(0)),
  Stream.runCollect
)
// => [1, 2, 0]
```

#### `Stream.result`

Surfaces each failure as a [`Result`](https://effect.plants.sh/data-types/result/) value inside the
stream, turning the error channel into `never`. Successful elements become
`Result.succeed`; the failure becomes a trailing `Result.fail`.

```ts
import { Effect, Result, Stream } from "effect"

const program = Stream.make(1, 2).pipe(
  Stream.concat(Stream.fail("boom")),
  Stream.result,
  Stream.map(Result.match({
    onFailure: (error) => `failure: ${error}`,
    onSuccess: (value) => `success: ${value}`
  })),
  Stream.runCollect
)
// => ["success: 1", "success: 2", "failure: boom"]
```

### Fallbacks

#### `Stream.orElseIfEmpty`

Switches to a fallback stream only when the source completes *without emitting
any value*. Failures still propagate.

```ts
import { Effect, Stream } from "effect"

const program = Stream.empty.pipe(
  Stream.orElseIfEmpty(() => Stream.make(1, 2)),
  Stream.runCollect
)
// => [1, 2]
```

#### `Stream.orElseSucceed`

Recovers from a failure by emitting a single fallback value computed from the
error. The error channel becomes `never`.

```ts
import { Effect, Stream } from "effect"

const program = Stream.fail("NetworkError").pipe(
  Stream.orElseSucceed((error) => `Recovered: ${error}`),
  Stream.runCollect
)
// => ["Recovered: NetworkError"]
```

#### `Stream.orDie`

Turns typed failures into defects, making the stream infallible (`E` becomes
`never`). Use when a failure here is a bug, not a recoverable condition.

```ts
import { Effect, Stream } from "effect"

const infallible = Stream.make(1, 2, 3).pipe(Stream.orDie)
const program = Stream.runCollect(infallible)
// => [1, 2, 3]   (a failure would now surface as a defect)
```

#### `Stream.ignore`

Ignores typed failures and ends the stream gracefully, keeping whatever was
emitted before. The `log` option (default on) controls whether the failure is
logged first.

```ts
import { Effect, Stream } from "effect"

const program = Stream.make(1, 2, 3).pipe(
  Stream.concat(Stream.fail("boom")),
  Stream.ignore({ log: false }),
  Stream.runCollect
)
// => [1, 2, 3]
```

#### `Stream.ignoreCause`

Like `ignore`, but suppresses the entire `Cause` — including defects — rather
than only typed failures, then ends the stream.

```ts
import { Effect, Stream } from "effect"

const program = Stream.make(1, 2).pipe(
  Stream.concat(Stream.die("boom")),
  Stream.ignoreCause({ log: false }),
  Stream.runCollect
)
// => [1, 2]
```

### Retry and resilience

#### `Stream.retry`

Re-runs the whole stream on failure according to a [Schedule](https://effect.plants.sh/scheduling/schedule/).
The schedule resets once the first element passes through again.

```ts
import { Effect, Schedule, Stream } from "effect"

const program = Stream.make(1).pipe(
  Stream.concat(Stream.fail("boom")),
  Stream.retry(Schedule.recurs(1)),
  Stream.take(2),
  Stream.runCollect
)
// => [1, 1]
```

#### `Stream.withExecutionPlan`

Applies an `ExecutionPlan`: on failure it retries with
each step's provided resources until one succeeds or the plan is exhausted. Set
`preventFallbackOnPartialStream: true` to fail rather than mix partial output
with a later step's fallback.

```ts
import { Context, Effect, ExecutionPlan, Layer, Stream } from "effect"

class Service extends Context.Service<Service>()("Service", {
  make: Effect.succeed({
    stream: Stream.fail("A") as Stream.Stream<number, string>
  })
}) {
  static Bad = Layer.succeed(Service, Service.of({ stream: Stream.fail("A") }))
  static Good = Layer.succeed(Service, Service.of({ stream: Stream.make(1, 2, 3) }))
}

const plan = ExecutionPlan.make(
  { provide: Service.Bad },
  { provide: Service.Good }
)

const stream = Stream.unwrap(Effect.map(Service, (_) => _.stream))

const program = stream.pipe(Stream.withExecutionPlan(plan), Stream.runCollect)
// => [1, 2, 3]
```

### Timeouts

#### `Stream.timeout`

Ends the stream (emitting nothing more) if no element is produced within the
duration. The timeout is checked on each pull.

```ts
import { Effect, Stream } from "effect"

const slow = Stream.fromEffect(Effect.as(Effect.never, 0))

const program = slow.pipe(Stream.timeout("2 seconds"))
// => []  (after 2s with no element)
```

#### `Stream.timeoutOrElse`

Switches to a fallback stream if no value arrives within the duration. The
fallback is *not* itself timed after the switch.

```ts
import { Effect, Stream } from "effect"

const slow = Stream.fromEffect(Effect.as(Effect.never, 0))

const program = slow.pipe(
  Stream.timeoutOrElse({
    duration: "2 seconds",
    orElse: () => Stream.make(1, 2, 3)
  }),
  Stream.runCollect
)
// => [1, 2, 3]
```

### Lifecycle and finalizers

#### `Stream.onStart`

Runs an effect once, before the stream starts pulling. Its error and
requirements are added to the stream.

```ts
import { Console, Effect, Stream } from "effect"

const program = Stream.fromArray([1, 2, 3]).pipe(
  Stream.onStart(Console.log("Stream started")),
  Stream.runCollect
)
// logs: Stream started
// => [1, 2, 3]
```

#### `Stream.onFirst`

Runs an effect with the *first* element emitted by the stream (and only the
first).

```ts
import { Console, Effect, Stream } from "effect"

const program = Stream.fromArray([1, 2, 3]).pipe(
  Stream.onFirst((value) => Console.log(`first=${value}`)),
  Stream.runDrain
)
// logs: first=1
```

#### `Stream.onEnd`

Runs an effect when the stream ends *successfully* (not on failure).

```ts
import { Console, Effect, Stream } from "effect"

const program = Stream.make(1, 2, 3).pipe(
  Stream.onEnd(Console.log("Stream ended")),
  Stream.runCollect
)
// logs: Stream ended
// => [1, 2, 3]
```

#### `Stream.onError`

Runs a cleanup effect only when the stream fails, receiving the `Cause`. It does
not change the failure.

```ts
import { Cause, Effect, Stream } from "effect"

const program = Stream.make(1, 2, 3).pipe(
  Stream.concat(Stream.fail("boom")),
  Stream.onError((cause) => Effect.logError(`Stream failed: ${Cause.squash(cause)}`))
)
// logs on failure: Stream failed: boom (then the failure propagates)
```
**Caution:** Unlike `Effect.onError`, the `Stream.onError` cleanup effect carries no
  guarantee that it will not itself be interrupted. For guaranteed-to-run
  cleanup, prefer `Stream.ensuring`.

#### `Stream.onExit`

Runs a finalizer when the stream exits for *any* reason, receiving the
[`Exit`](https://effect.plants.sh/data-types/exit-and-cause/) so you can branch on success vs failure.

```ts
import { Console, Effect, Exit, Stream } from "effect"

const program = Stream.make(1, 2, 3).pipe(
  Stream.onExit((exit) =>
    Exit.isSuccess(exit)
      ? Console.log("completed successfully")
      : Console.log("failed")
  ),
  Stream.runCollect
)
// logs: completed successfully
// => [1, 2, 3]
```

#### `Stream.ensuring`

Runs a finalizer after the stream's own finalizers, regardless of how it exits.
This is the resilient cleanup hook.

```ts
import { Console, Effect, Stream } from "effect"

const program = Stream.fromArray([1, 2]).pipe(
  Stream.ensuring(Effect.orDie(Console.log("cleanup"))),
  Stream.runCollect
)
// logs: cleanup
// => [1, 2]
```

### Providing context

These mirror the `Effect.provide*` family and remove the corresponding service
requirements from a stream. See [Services](https://effect.plants.sh/services-and-layers/services/)
for the full story.

#### `Stream.provide`

Provides a [`Layer`](https://effect.plants.sh/services-and-layers/managing-layers/) (or a built `Context`) to the
stream, discharging its requirements. Layers are shared between provide calls
unless you pass `{ local: true }`.

```ts
import { Console, Context, Effect, Layer, Stream } from "effect"

class Env extends Context.Service<Env, { readonly name: string }>()("Env") {}

const layer = Layer.succeed(Env)({ name: "Ada" })

const program = Stream.fromEffect(
  Effect.map(Effect.service(Env), (env) => `Hello, ${env.name}`)
).pipe(
  Stream.provide(layer),
  Stream.runCollect,
  Effect.tap((values) => Console.log(values))
)
// => ["Hello, Ada"]
```

#### `Stream.provideContext`

Provides a whole prebuilt `Context` at once, eliminating several requirements.

```ts
import { Context, Effect, Stream } from "effect"

class Config extends Context.Service<Config, { readonly prefix: string }>()("Config") {}
class Greeter extends Context.Service<Greeter, { greet: (name: string) => string }>()("Greeter") {}

const context = Context.make(Config, { prefix: "Hello" }).pipe(
  Context.add(Greeter, { greet: (name: string) => `${name}!` })
)

const stream = Stream.fromEffect(
  Effect.gen(function*() {
    const config = yield* Effect.service(Config)
    const greeter = yield* Effect.service(Greeter)
    return greeter.greet(config.prefix)
  })
)

const program = Stream.runCollect(Stream.provideContext(stream, context))
// => ["Hello!"]
```

#### `Stream.provideService`

Provides a single service value directly by its key.

```ts
import { Context, Effect, Stream } from "effect"

class Greeter extends Context.Service<Greeter, {
  greet: (name: string) => string
}>()("Greeter") {}

const stream = Stream.fromEffect(
  Effect.map(Effect.service(Greeter), (g) => g.greet("Ada"))
)

const program = stream.pipe(
  Stream.provideService(Greeter, { greet: (name) => `Hello, ${name}` }),
  Stream.runCollect
)
// => ["Hello, Ada"]
```

#### `Stream.provideServiceEffect`

Provides a service whose value is produced by an effect, adding that effect's
error and requirements to the stream.

```ts
import { Console, Context, Effect, Stream } from "effect"

class ApiConfig extends Context.Service<ApiConfig, { readonly baseUrl: string }>()("ApiConfig") {}

const stream = Stream.fromEffect(
  Effect.map(Effect.service(ApiConfig), (c) => c.baseUrl)
)

const program = stream.pipe(
  Stream.provideServiceEffect(
    ApiConfig,
    Effect.succeed({ baseUrl: "https://example.com" }).pipe(
      Effect.tap(() => Console.log("Loading config..."))
    )
  ),
  Stream.runCollect
)
// logs: Loading config...
// => ["https://example.com"]
```

#### `Stream.updateContext`

Transforms the stream's required context by mapping the current `Context` to a
new one — handy for adding or replacing services without fully providing them.

```ts
import { Context, Effect, Stream } from "effect"

class Logger extends Context.Service<Logger, { prefix: string }>()("Logger") {}
class Config extends Context.Service<Config, { name: string }>()("Config") {}

const stream = Stream.fromEffect(
  Effect.gen(function*() {
    const logger = yield* Effect.service(Logger)
    const config = yield* Effect.service(Config)
    return `${logger.prefix}${config.name}`
  })
)

const updated = stream.pipe(
  Stream.updateContext((context: Context.Context<Logger>) =>
    Context.add(context, Config, { name: "World" })
  )
)
// requires only Logger now; Config is supplied by updateContext
```

#### `Stream.updateService`

Updates a single service in the environment by applying a function to its
current value (the requirement remains).

```ts
import { Context, Effect, Stream } from "effect"

class Counter extends Context.Service<Counter, { count: number }>()("Counter") {}

const stream = Stream.fromEffect(Effect.service(Counter)).pipe(
  Stream.updateService(Counter, (counter) => ({ count: counter.count + 1 }))
)

const program = Stream.runCollect(stream).pipe(
  Effect.provideService(Counter, { count: 0 })
)
// => [{ count: 1 }]
```

#### `Stream.withSpan`

Wraps the stream in a tracing span, eliminating the `ParentSpan` requirement.

```ts
import { Effect, Stream } from "effect"

const program = Stream.fromArray([1, 2, 3]).pipe(
  Stream.withSpan("numbers"),
  Stream.runCollect
)
// => [1, 2, 3]   (emitted under a "numbers" span)
```
**Note:** Two related operators stop a stream from the *outside* rather than recovering
  from an error: `Stream.interruptWhen` interrupts an in-progress pull as soon
  as an effect completes, while `Stream.haltWhen` stops cleanly after the
  current element. Both are covered on the
  [Concurrency](https://effect.plants.sh/streaming/concurrency-and-merging/) page.