/* oxlint-disable no-restricted-syntax */
import { assert, describe, it } from "@effect/vitest"
import { assertExitFailure, assertSuccess, assertTrue, deepStrictEqual, strictEqual } from "@effect/vitest/utils"
import {
  Array,
  Cause,
  Clock,
  Context,
  Data,
  Deferred,
  Duration,
  Effect,
  Exit,
  Fiber,
  Latch,
  Logger,
  type LogLevel,
  Option,
  type Pull,
  Queue,
  Ref,
  References,
  Result,
  Schedule,
  Sink,
  Stream,
  String as Str
} from "effect"
import { isReadonlyArrayNonEmpty, type NonEmptyArray } from "effect/Array"
import { constTrue, constVoid, pipe } from "effect/Function"
import { TestClock } from "effect/testing"
import * as fc from "effect/testing/FastCheck"
import { assertCauseFail, assertFailure } from "./utils/assert.ts"
import { chunkCoordination } from "./utils/chunkCoordination.ts"

describe("Stream", () => {
  describe("callback", () => {
    it.effect("with take", () =>
      Effect.gen(function*() {
        const array = [1, 2, 3, 4, 5]
        const result = yield* Stream.callback<number>((mb) =>
          Effect.sync(() => {
            array.forEach((n) => {
              Queue.offerUnsafe(mb, n)
            })
          })
        ).pipe(
          Stream.take(array.length),
          Stream.runCollect
        )
        assert.deepStrictEqual(result, array)
      }))

    it.effect("with cleanup", () =>
      Effect.gen(function*() {
        let cleanup = false
        const latch = yield* Latch.make()
        const fiber = yield* Stream.callback<void>(Effect.fnUntraced(function*(mb) {
          yield* Effect.addFinalizer(() =>
            Effect.sync(() => {
              cleanup = true
            })
          )
          yield* Queue.offer(mb, void 0)
        })).pipe(
          Stream.tap(() => latch.open),
          Stream.runDrain,
          Effect.forkChild
        )
        yield* latch.await
        yield* Fiber.interrupt(fiber)
        assert.isTrue(cleanup)
      }))

    it.effect("signals the end of the stream", () =>
      Effect.gen(function*() {
        const result = yield* Stream.callback<number>((mb) => {
          Queue.endUnsafe(mb)
          return Effect.void
        }).pipe(Stream.runCollect)
        assert.isTrue(result.length === 0)
      }))

    it.effect("handles errors", () =>
      Effect.gen(function*() {
        const error = new Error("boom")
        const result = yield* Stream.callback<number, Error>((mb) => {
          Queue.failCauseUnsafe(mb, Cause.fail(error))
          return Effect.void
        }).pipe(
          Stream.runCollect,
          Effect.exit
        )
        assert.deepStrictEqual(result, Exit.fail(error))
      }))

    it.effect("handles defects", () =>
      Effect.gen(function*() {
        const error = new Error("boom")
        const result = yield* Stream.callback<number, Error>(() => {
          throw error
        }).pipe(
          Stream.runCollect,
          Effect.exit
        )
        assert.deepStrictEqual(result, Exit.die(error))
      }))

    it.effect("backpressure", () =>
      Effect.gen(function*() {
        let count = 0
        let offered = 0
        let done = false
        const pull = yield* Stream.callback<number>((mb) =>
          Effect.forEach(
            [1, 2, 3, 4, 5, 6, 7],
            Effect.fnUntraced(function*(n) {
              count++
              yield* Queue.offer(mb, n)
              offered++
            }),
            { concurrency: "unbounded" }
          ).pipe(
            Effect.tap(() => Effect.sync(() => done = true))
          ), { bufferSize: 2 }).pipe(Stream.toPull)
        yield* Effect.yieldNow
        assert.strictEqual(count, 7)
        assert.strictEqual(offered, 2)
        assert.isFalse(done)
        yield* pull
        assert.strictEqual(offered, 4)
        assert.isFalse(done)
      }))
  })

  describe("destructors", () => {
    it.effect("runForEachWhile continues across chunk boundaries", () =>
      Effect.gen(function*() {
        const seen: Array<number> = []
        yield* Stream.fromArrays([1, 2], [3, 4]).pipe(
          Stream.runForEachWhile((n) =>
            Effect.sync(() => {
              seen.push(n)
              return true
            })
          )
        )
        assert.deepStrictEqual(seen, [1, 2, 3, 4])
      }))

    it.effect("runForEachWhile stops when predicate returns false", () =>
      Effect.gen(function*() {
        const seen: Array<number> = []
        yield* Stream.fromArrays([1, 2, 3], [4, 5]).pipe(
          Stream.runForEachWhile((n) =>
            Effect.sync(() => {
              seen.push(n)
              return n < 3
            })
          )
        )
        assert.deepStrictEqual(seen, [1, 2, 3])
      }))
  })

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

    it.effect("range - min less than max", () =>
      Effect.gen(function*() {
        const result = yield* Stream.range(1, 3).pipe(Stream.runCollect)
        assert.deepStrictEqual(result, [1, 2, 3])
      }))

    it.effect("service", () =>
      Effect.gen(function*() {
        const result = yield* Stream.service(Greeter).pipe(
          Stream.map((greeter) => greeter.greet("World")),
          Stream.provideService(Greeter, {
            greet: (name) => `Hello, ${name}!`
          }),
          Stream.runCollect
        )
        assert.deepStrictEqual(result, ["Hello, World!"])
      }))

    it.effect("serviceOption - some", () =>
      Effect.gen(function*() {
        const result = yield* Stream.serviceOption(Greeter).pipe(
          Stream.map((option) => Option.map(option, (greeter) => greeter.greet("World"))),
          Stream.provideService(Greeter, {
            greet: (name) => `Hello, ${name}!`
          }),
          Stream.runCollect
        )
        assert.deepStrictEqual(result, [Option.some("Hello, World!")])
      }))

    it.effect("serviceOption - none", () =>
      Effect.gen(function*() {
        const result = yield* Stream.serviceOption(Greeter).pipe(Stream.runCollect)
        assert.deepStrictEqual(result, [Option.none()])
      }))

    it.effect("range - min greater than max", () =>
      Effect.gen(function*() {
        const result = yield* Stream.range(4, 3).pipe(Stream.runCollect)
        assert.deepStrictEqual(result, [])
      }))

    it.effect("range - min equal to max", () =>
      Effect.gen(function*() {
        const result = yield* Stream.range(3, 3).pipe(Stream.runCollect)
        assert.deepStrictEqual(result, [3])
      }))

    it.effect("fromIterable - emits array as one chunk by default", () =>
      Effect.gen(function*() {
        const result = yield* Stream.fromIterable([1, 2, 3, 4]).pipe(
          Stream.chunks,
          Stream.runCollect
        )
        assert.deepStrictEqual(result, [[1, 2, 3, 4]])
      }))

    it.effect("fromIterable - supports chunkSize option", () =>
      Effect.gen(function*() {
        const result = yield* Stream.fromIterable([1, 2, 3, 4, 5], { chunkSize: 2 }).pipe(
          Stream.chunks,
          Stream.runCollect
        )
        assert.deepStrictEqual(result, [[1, 2], [3, 4], [5]])
      }))

    it.effect("scoped - provides scope to fromEffect pull effects", () =>
      Effect.gen(function*() {
        const releases = yield* Ref.make(0)
        const result = yield* Stream.fromEffect(
          Effect.acquireRelease(
            Effect.succeed("resource"),
            () => Ref.update(releases, (n) => n + 1)
          )
        ).pipe(
          Stream.scoped,
          Stream.runCollect
        )

        assert.deepStrictEqual(result, ["resource"])
        assert.strictEqual(yield* Ref.get(releases), 1)
      }))

    it.effect("scoped - provides scope to sequential mapEffect pull effects", () =>
      Effect.gen(function*() {
        const releases = yield* Ref.make(0)
        const result = yield* Stream.fromIterable([1, 2]).pipe(
          Stream.mapEffect((n) =>
            Effect.acquireRelease(
              Effect.succeed(n * 2),
              () => Ref.update(releases, (count) => count + 1)
            )
          ),
          Stream.scoped,
          Stream.runCollect
        )

        assert.deepStrictEqual(result, [2, 4])
        assert.strictEqual(yield* Ref.get(releases), 2)
      }))

    it.effect("fromReadableStream - errored streams fail with the mapped error, not a finalizer defect", () =>
      Effect.gen(function*() {
        const exit = yield* Stream.fromReadableStream({
          evaluate: () =>
            new ReadableStream<number>({
              start(controller) {
                controller.error(new Error("boom"))
              }
            }),
          onError: (error) => new Error(`mapped: ${(error as Error).message}`)
        }).pipe(Stream.runDrain, Effect.exit)

        assertTrue(Exit.isFailure(exit))
        if (Exit.isFailure(exit)) {
          assertTrue(exit.cause.reasons.every(Cause.isFailReason))
          deepStrictEqual(Cause.squash(exit.cause), new Error("mapped: boom"))
        }
      }))
  })

  describe("encoding", () => {
    it.effect("decodeText handles multi-byte characters split across chunks", () =>
      Effect.gen(function*() {
        const bytes = new TextEncoder().encode("🌍")
        const result = yield* Stream.make(bytes.slice(0, 2), bytes.slice(2)).pipe(
          Stream.decodeText(),
          Stream.mkString
        )
        assert.strictEqual(result, "🌍")
      }))

    it.effect("decodeText handles mixed ASCII and multi-byte characters across chunks", () =>
      Effect.gen(function*() {
        const bytes = new TextEncoder().encode("Hello 🌍!")
        const split = 8
        const result = yield* Stream.make(bytes.slice(0, split), bytes.slice(split)).pipe(
          Stream.decodeText(),
          Stream.mkString
        )
        assert.strictEqual(result, "Hello 🌍!")
      }))

    describe("splitLines", () => {
      const splitLines = (chunks: ReadonlyArray<string>) =>
        Stream.fromIterable(chunks).pipe(
          Stream.splitLines,
          Stream.runCollect
        )

      it.effect("handles an empty stream", () =>
        Effect.gen(function*() {
          const result = yield* splitLines([])
          assert.deepStrictEqual(result, [])
        }))

      it.effect("handles empty chunks and repeated line feeds", () =>
        Effect.gen(function*() {
          const result = yield* splitLines(["", "a\n\n", "", "b\n", ""])
          assert.deepStrictEqual(result, ["a", "", "b"])
        }))

      it.effect("handles line feeds across chunks", () =>
        Effect.gen(function*() {
          const result = yield* splitLines(["a\nb", "\nc\n", "d"])
          assert.deepStrictEqual(result, ["a", "b", "c", "d"])
        }))

      it.effect("handles carriage return / line feed pairs across chunks", () =>
        Effect.gen(function*() {
          const result = yield* splitLines(["a\r", "\nb\r", "\nc"])
          assert.deepStrictEqual(result, ["a", "b", "c"])
        }))

      it.effect("emits the final line when the stream does not end with a newline", () =>
        Effect.gen(function*() {
          const result = yield* splitLines(["a\nb\nc\n", "d\ne"])
          assert.deepStrictEqual(result, ["a", "b", "c", "d", "e"])
        }))

      it.effect("does not pull upstream again after flushing the final line", () =>
        Effect.gen(function*() {
          let pulls = 0
          const stream = Stream.fromPull(Effect.sync(() => {
            const pull = Effect.suspend(
              (): Pull.Pull<Array.NonEmptyReadonlyArray<string>, string> => {
                pulls++
                switch (pulls) {
                  case 1:
                    return Effect.succeed(Array.of("a"))
                  case 2:
                    return Cause.done()
                  default:
                    return Effect.fail("pulled after done")
                }
              }
            )
            return pull
          }))

          const result = yield* stream.pipe(
            Stream.splitLines,
            Stream.runCollect
          )

          assert.deepStrictEqual(result, ["a"])
          assert.strictEqual(pulls, 2)
        }))

      it.effect("handles standalone carriage returns within a chunk", () =>
        Effect.gen(function*() {
          const result = yield* splitLines(["a\rb\rc"])
          assert.deepStrictEqual(result, ["a", "b", "c"])
        }))

      it.effect("handles standalone carriage returns across chunk boundaries", () =>
        Effect.gen(function*() {
          const result = yield* splitLines(["a\r", "b\r", "c"])
          assert.deepStrictEqual(result, ["a", "b", "c"])
        }))

      it.effect.prop(
        "matches String.linesIterator regardless of chunk boundaries",
        {
          chunks: fc.array(
            fc.array(fc.constantFrom("a", "b", "\n", "\r"), { maxLength: 8 }).map((chars) => chars.join("")),
            { maxLength: 8 }
          )
        },
        Effect.fnUntraced(function*({ chunks }) {
          const result = yield* splitLines(chunks)
          const expected = globalThis.Array.from(Str.linesIterator(chunks.join("")))
          assert.deepStrictEqual(result, expected)
        })
      )
    })
  })

  describe("filtering", () => {
    it.effect("filterMap", () =>
      Effect.gen(function*() {
        const result = yield* Stream.make(1, 2, 3, 4).pipe(
          Stream.filterMap((n) => n % 2 === 0 ? Result.succeed(n + 10) : Result.failVoid),
          Stream.runCollect
        )
        assert.deepStrictEqual(result, [12, 14])
      }))

    it.effect("filterMapEffect", () =>
      Effect.gen(function*() {
        const result = yield* Stream.make(1, 2, 3, 4).pipe(
          Stream.filterMapEffect((n) => Effect.succeed(n % 2 === 0 ? Result.succeed(n + 1) : Result.failVoid)),
          Stream.runCollect
        )
        assert.deepStrictEqual(result, [3, 5])
      }))
  })

  describe("taking", () => {
    it.effect("take - pulls the first `n` values from a stream", () =>
      Effect.gen(function*() {
        const result = yield* Stream.range(1, 5).pipe(
          Stream.take(3),
          Stream.runCollect
        )
        assert.deepStrictEqual(result, [1, 2, 3])
      }))

    it.effect("take - short-circuits stream evaluation", () =>
      Effect.gen(function*() {
        const result = yield* Stream.succeed(1).pipe(
          Stream.concat(Stream.never),
          Stream.take(1),
          Stream.runCollect
        )
        assert.deepStrictEqual(result, [1])
      }))

    it.effect("take - taking 0 short-circuits stream evaluation", () =>
      Effect.gen(function*() {
        const result = yield* Stream.never.pipe(
          Stream.take(0),
          Stream.runCollect
        )
        assert.deepStrictEqual(result, [])
      }))

    it.effect("takeUntil - takes elements until a predicate is satisfied", () =>
      Effect.gen(function*() {
        const result = yield* Stream.range(1, 5).pipe(
          Stream.takeUntil((n) => n % 3 === 0),
          Stream.runCollect
        )
        assert.deepStrictEqual(result, [1, 2, 3])
      }))

    it.effect("takeWhile - takes elements while a predicate is satisfied", () =>
      Effect.gen(function*() {
        const result = yield* Stream.range(1, 5).pipe(
          Stream.takeWhile((n) => n % 3 !== 0),
          Stream.runCollect
        )
        assert.deepStrictEqual(result, [1, 2])
      }))

    it.effect("takeWhileFilter - takes elements while a filter succeeds", () =>
      Effect.gen(function*() {
        const result = yield* Stream.range(1, 5).pipe(
          Stream.takeWhileFilter((n) => n < 3 ? Result.succeed(n + 10) : Result.failVoid),
          Stream.runCollect
        )
        assert.deepStrictEqual(result, [11, 12])
      }))

    it.effect("takeUntilEffect - takes elements until an effectful predicate is satisfied", () =>
      Effect.gen(function*() {
        const result = yield* Stream.range(1, 5).pipe(
          Stream.takeUntilEffect((n) => Effect.succeed(n % 3 === 0)),
          Stream.runCollect
        )
        assert.deepStrictEqual(result, [1, 2, 3])
      }))

    it.effect("takeWhileEffect - takes elements while an effectful predicate is satisfied", () =>
      Effect.gen(function*() {
        const result = yield* Stream.range(1, 5).pipe(
          Stream.takeWhileEffect((n) => Effect.succeed(n % 3 !== 0)),
          Stream.runCollect
        )
        assert.deepStrictEqual(result, [1, 2])
      }))

    it.effect("takeRight", () =>
      Effect.gen(function*() {
        const take = 3
        const stream = Stream.range(1, 5)
        const { result1, result2 } = yield* (Effect.all({
          result1: pipe(stream, Stream.takeRight(take), Stream.runCollect),
          result2: pipe(Stream.runCollect(stream), Effect.map(Array.takeRight(take)))
        }))
        deepStrictEqual(result1, result2)
      }))
  })

  describe("pagination", () => {
    it.effect("paginate", () =>
      Effect.gen(function*() {
        const s: readonly [ReadonlyArray<number>, Array<number>] = [[0], [1, 2, 3, 4, 5]]
        const pageSize = 2
        const result = yield* Stream.paginate(s, ([chunk, nums]) =>
          nums.length === 0 ?
            Effect.succeed([chunk, Option.none<readonly [ReadonlyArray<number>, Array<number>]>()] as const) :
            Effect.succeed(
              [
                chunk,
                Option.some(
                  [
                    nums.slice(0, pageSize),
                    nums.slice(pageSize)
                  ] as const
                )
              ] as const
            )).pipe(Stream.runCollect)
        assert.deepStrictEqual(result, [0, 1, 2, 3, 4, 5])
      }))
  })

  describe("error handling", () => {
    it.effect("catch", () =>
      Effect.gen(function*() {
        let error: string | undefined = undefined
        const results = yield* Stream.make(1, 2, 3).pipe(
          Stream.concat(Stream.fail("boom")),
          Stream.catch((error_) => {
            error = error_
            return Stream.make(4, 5, 6)
          }),
          Stream.runCollect
        )
        assert.deepStrictEqual(results, [1, 2, 3, 4, 5, 6])
        assert.strictEqual(error, "boom")
      }))

    it.effect("tapCause preserves failures", () =>
      Effect.gen(function*() {
        const exit = yield* Stream.fail("boom").pipe(
          Stream.tapCause(() => Effect.void),
          Stream.runCollect,
          Effect.exit
        )
        assertExitFailure(exit, Cause.fail("boom"))
      }))

    it.effect("tapCause preserves defects", () =>
      Effect.gen(function*() {
        const defect = new Error("boom")
        const exit = yield* Stream.die(defect).pipe(
          Stream.tapCause(() => Effect.void),
          Stream.runCollect,
          Effect.exit
        )
        assertExitFailure(exit, Cause.die(defect))
      }))

    it.effect("catchIf with refinement", () =>
      Effect.gen(function*() {
        interface ErrorA {
          readonly _tag: "ErrorA"
        }
        interface ErrorB {
          readonly _tag: "ErrorB"
        }
        const stream: Stream.Stream<never, ErrorA | ErrorB> = Stream.fail({ _tag: "ErrorB" as const })
        const result = yield* pipe(
          stream,
          Stream.catchIf((error): error is ErrorA => error._tag === "ErrorA", () => Stream.succeed("ok")),
          Stream.runCollect,
          Effect.exit
        )
        assert.deepStrictEqual(result, Exit.fail({ _tag: "ErrorB" as const }))
      }))

    it.effect("catchIf with refinement orElse", () =>
      Effect.gen(function*() {
        interface ErrorA {
          readonly _tag: "ErrorA"
        }
        interface ErrorB {
          readonly _tag: "ErrorB"
        }
        const result = yield* (Stream.fail({ _tag: "ErrorB" as const }) as Stream.Stream<never, ErrorA | ErrorB>).pipe(
          Stream.catchIf(
            (error): error is ErrorA => error._tag === "ErrorA",
            () => Stream.succeed("caught"),
            () => Stream.succeed("fallback")
          ),
          Stream.runCollect
        )
        assert.deepStrictEqual(result, ["fallback"])
      }))

    it.effect("catchIf with predicate", () =>
      Effect.gen(function*() {
        const result = yield* pipe(
          Stream.fail("boom" as string | number),
          Stream.catchIf(
            (e) => typeof e === "string",
            (e) => Stream.succeed(`caught: ${e}`)
          ),
          Stream.runCollect
        )
        assert.deepStrictEqual(result, ["caught: boom"])
      }))

    it.effect("catchIf predicate no match", () =>
      Effect.gen(function*() {
        const result = yield* pipe(
          Stream.fail(42 as string | number),
          Stream.catchIf(
            (e) => typeof e === "string",
            () => Stream.succeed("caught")
          ),
          Stream.runCollect,
          Effect.exit
        )
        assert.deepStrictEqual(result, Exit.fail(42))
      }))

    it.effect("catchIf with filter input", () =>
      Effect.gen(function*() {
        const matched = yield* Stream.catchFilter(
          Stream.fail("boom" as string | number),
          (e: string | number) => typeof e === "string" ? Result.succeed(e.length) : Result.fail(`fallback: ${e}`),
          (length) => Stream.succeed(`caught: ${length}`),
          (failure) => Stream.succeed(failure)
        ).pipe(Stream.runCollect)
        const unmatched = yield* Stream.catchFilter(
          Stream.fail(42 as string | number),
          (e: string | number) => typeof e === "string" ? Result.succeed(e.length) : Result.fail(`fallback: ${e}`),
          (length) => Stream.succeed(`caught: ${length}`),
          (failure) => Stream.succeed(failure)
        ).pipe(Stream.runCollect)
        assert.deepStrictEqual(matched, ["caught: 4"])
        assert.deepStrictEqual(unmatched, ["fallback: 42"])
      }))

    it.effect("catchCauseFilter", () =>
      Effect.gen(function*() {
        const recovered = yield* Stream.fail("boom").pipe(
          Stream.catchCauseFilter(
            (cause) => Cause.hasFails(cause) ? Result.succeed(Cause.squash(cause)) : Result.fail(cause),
            (failure) => Stream.succeed(`recovered: ${failure}`)
          ),
          Stream.runCollect
        )
        assert.deepStrictEqual(recovered, ["recovered: boom"])
      }))

    it.effect("catchTag orElse", () =>
      Effect.gen(function*() {
        class HttpError extends Data.TaggedError("HttpError")<{
          readonly message: string
        }> {}
        class ValidationError extends Data.TaggedError("ValidationError")<{
          readonly field: string
        }> {}
        const result = yield* Stream.catchTag(
          Stream.fail(new ValidationError({ field: "email" })) as Stream.Stream<never, HttpError | ValidationError>,
          "HttpError",
          () => Stream.succeed("http"),
          () => Stream.succeed("fallback")
        ).pipe(Stream.runCollect)
        assert.deepStrictEqual(result, ["fallback"])
      }))

    describe("ignore", () => {
      type IgnoreOptions = { readonly log?: boolean | LogLevel.Severity }

      const makeTestLogger = () => {
        const capturedLogs: Array<{
          readonly logLevel: LogLevel.LogLevel
          readonly cause: Cause.Cause<unknown>
        }> = []
        const testLogger = Logger.make<unknown, void>((options) => {
          capturedLogs.push({ logLevel: options.logLevel, cause: options.cause })
        })
        return { capturedLogs, testLogger }
      }

      const runIgnore = (options?: IgnoreOptions, currentLogLevel: LogLevel.Severity = "Info") =>
        Effect.gen(function*() {
          const { capturedLogs, testLogger } = makeTestLogger()
          const program = options === undefined
            ? Stream.fail("boom").pipe(Stream.ignore, Stream.runDrain)
            : Stream.fail("boom").pipe(Stream.ignore(options), Stream.runDrain)
          yield* program.pipe(
            Effect.provide(Logger.layer([testLogger])),
            Effect.provideService(References.MinimumLogLevel, "Trace"),
            Effect.provideService(References.CurrentLogLevel, currentLogLevel)
          )
          return capturedLogs
        })

      it.effect("does not log when log is omitted", () =>
        Effect.gen(function*() {
          const logs = yield* runIgnore()
          assert.strictEqual(logs.length, 0)
        }))

      it.effect("does not log when log is false", () =>
        Effect.gen(function*() {
          const logs = yield* runIgnore({ log: false })
          assert.strictEqual(logs.length, 0)
        }))

      it.effect("logs with the current level when log is true", () =>
        Effect.gen(function*() {
          const logs = yield* runIgnore({ log: true }, "Warn")
          assert.strictEqual(logs.length, 1)
          assert.strictEqual(logs[0].logLevel, "Warn")
          assertCauseFail(logs[0].cause, "boom")
        }))

      it.effect("logs with the provided level when log is a LogLevel", () =>
        Effect.gen(function*() {
          const logs = yield* runIgnore({ log: "Error" }, "Warn")
          assert.strictEqual(logs.length, 1)
          assert.strictEqual(logs[0].logLevel, "Error")
          assertCauseFail(logs[0].cause, "boom")
        }))
    })

    describe("catchReason", () => {
      class RateLimitError extends Data.TaggedError("RateLimitError")<{
        readonly retryAfter: number
      }> {}

      class QuotaExceededError extends Data.TaggedError("QuotaExceededError")<{
        readonly limit: number
      }> {}

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

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

      it.effect("catches matching reason - handler succeeds", () =>
        Effect.gen(function*() {
          const result = yield* Stream.fail(
            new AiError({ reason: new RateLimitError({ retryAfter: 60 }) })
          ).pipe(
            Stream.catchReason("AiError", "RateLimitError", (r) => Stream.succeed(`retry: ${r.retryAfter}`)),
            Stream.runCollect
          )
          assert.deepStrictEqual(result, ["retry: 60"])
        }))

      it.effect("catches matching reason - handler fails", () =>
        Effect.gen(function*() {
          const reason = new RateLimitError({ retryAfter: 60 })
          const error = new OtherError({ message: "handled" })
          const exit = yield* Stream.fail(new AiError({ reason })).pipe(
            Stream.catchReason("AiError", "RateLimitError", () => Stream.fail(error)),
            Stream.runCollect,
            Effect.exit
          )
          assertExitFailure(exit, Cause.fail(error))
        }))

      it.effect("ignores non-matching reason", () =>
        Effect.gen(function*() {
          const reason = new QuotaExceededError({ limit: 100 })
          const error = new AiError({ reason })
          const exit = yield* Stream.fail(error).pipe(
            Stream.catchReason("AiError", "RateLimitError", () => Stream.succeed("no")),
            Stream.runCollect,
            Effect.exit
          )
          assertExitFailure(exit, Cause.fail(error))
        }))

      it.effect("handles non-matching reason with orElse", () =>
        Effect.gen(function*() {
          const result = yield* Stream.fail(
            new AiError({ reason: new QuotaExceededError({ limit: 100 }) })
          ).pipe(
            Stream.catchReason(
              "AiError",
              "RateLimitError",
              (reason) => Stream.succeed(`retry: ${reason.retryAfter}`),
              (reason) => Stream.succeed(`quota: ${reason.limit}`)
            ),
            Stream.runCollect
          )
          assert.deepStrictEqual(result, ["quota: 100"])
        }))

      it.effect("ignores non-matching parent tag", () =>
        Effect.gen(function*() {
          const error = new OtherError({ message: "test" })
          const exit = yield* (Stream.fail(error) as Stream.Stream<never, AiError | OtherError>).pipe(
            Stream.catchReason("AiError", "RateLimitError", () => Stream.succeed("no")),
            Stream.runCollect,
            Effect.exit
          )
          assertExitFailure(exit, Cause.fail(error))
        }))

      it.effect("orElse ignores non-matching parent tag", () =>
        Effect.gen(function*() {
          const error = new OtherError({ message: "test" })
          const exit = yield* (Stream.fail(error) as Stream.Stream<never, AiError | OtherError>).pipe(
            Stream.catchReason(
              "AiError",
              "RateLimitError",
              () => Stream.succeed("no"),
              () => Stream.succeed("fallback")
            ),
            Stream.runCollect,
            Effect.exit
          )
          assertExitFailure(exit, Cause.fail(error))
        }))
    })

    describe("catchReasons", () => {
      class RateLimitError extends Data.TaggedError("RateLimitError")<{
        readonly retryAfter: number
      }> {}

      class QuotaExceededError extends Data.TaggedError("QuotaExceededError")<{
        readonly limit: number
      }> {}

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

      it.effect("catches matching reason", () =>
        Effect.gen(function*() {
          const result = yield* Stream.fail(
            new AiError({ reason: new RateLimitError({ retryAfter: 60 }) })
          ).pipe(
            Stream.catchReasons("AiError", {
              RateLimitError: (r) => Stream.succeed(`retry: ${r.retryAfter}`)
            }),
            Stream.runCollect
          )
          assert.deepStrictEqual(result, ["retry: 60"])
        }))

      it.effect("ignores non-matching reason", () =>
        Effect.gen(function*() {
          const reason = new QuotaExceededError({ limit: 100 })
          const error = new AiError({ reason })
          const exit = yield* Stream.fail(error).pipe(
            Stream.catchReasons("AiError", {
              RateLimitError: () => Stream.succeed("no")
            }),
            Stream.runCollect,
            Effect.exit
          )
          assertExitFailure(exit, Cause.fail(error))
        }))

      it.effect("handles non-matching reason with orElse", () =>
        Effect.gen(function*() {
          const result = yield* Stream.fail(
            new AiError({ reason: new RateLimitError({ retryAfter: 60 }) })
          ).pipe(
            Stream.catchReasons(
              "AiError",
              {
                QuotaExceededError: (reason) => Stream.succeed(`quota: ${reason.limit}`)
              },
              (reason) => Stream.succeed(`fallback: ${reason._tag}`)
            ),
            Stream.runCollect
          )
          assert.deepStrictEqual(result, ["fallback: RateLimitError"])
        }))
    })

    describe("ignoreCause", () => {
      type IgnoreCauseOptions = { readonly log?: boolean | LogLevel.Severity }

      const makeTestLogger = () => {
        const capturedLogs: Array<{
          readonly logLevel: LogLevel.LogLevel
          readonly cause: Cause.Cause<unknown>
        }> = []
        const testLogger = Logger.make<unknown, void>((options) => {
          capturedLogs.push({ logLevel: options.logLevel, cause: options.cause })
        })
        return { capturedLogs, testLogger }
      }

      const runIgnoreCause = (
        options?: IgnoreCauseOptions,
        currentLogLevel: LogLevel.Severity = "Info"
      ) =>
        Effect.gen(function*() {
          const stream: Stream.Stream<never, string, never> = Stream.fail("boom")
          const program: Effect.Effect<Array<never>, never, never> = options === undefined
            ? stream.pipe(Stream.ignoreCause, Stream.runCollect)
            : stream.pipe(Stream.ignoreCause(options), Stream.runCollect)
          yield* Effect.scope
          const { capturedLogs, testLogger } = makeTestLogger()
          yield* program.pipe(
            Effect.withLogger(testLogger),
            Effect.provideService(References.MinimumLogLevel, "Trace"),
            Effect.provideService(References.CurrentLogLevel, currentLogLevel)
          )
          return capturedLogs
        })

      it.effect("does not log when log is omitted", () =>
        Effect.gen(function*() {
          const logs = yield* runIgnoreCause()
          assert.strictEqual(logs.length, 0)
        }))

      it.effect("does not log when log is false", () =>
        Effect.gen(function*() {
          const logs = yield* runIgnoreCause({ log: false })
          assert.strictEqual(logs.length, 0)
        }))

      it.effect("logs with the current level when log is true", () =>
        Effect.gen(function*() {
          const logs = yield* runIgnoreCause({ log: true }, "Warn")
          assert.strictEqual(logs.length, 1)
          assert.strictEqual(logs[0].logLevel, "Warn")
          assertCauseFail(logs[0].cause, "boom")
        }))

      it.effect("logs with the provided level when log is a LogLevel", () =>
        Effect.gen(function*() {
          const logs = yield* runIgnoreCause({ log: "Error" }, "Warn")
          assert.strictEqual(logs.length, 1)
          assert.strictEqual(logs[0].logLevel, "Error")
          assertCauseFail(logs[0].cause, "boom")
        }))
    })
  })

  describe("scanning", () => {
    it.effect("scan", () =>
      Effect.gen(function*() {
        const stream = Stream.make(1, 2, 3, 4, 5)
        const { result1, result2 } = yield* Effect.all({
          result1: stream.pipe(Stream.scan(0, (acc, curr) => acc + curr), Stream.runCollect),
          result2: Stream.runCollect(stream).pipe(
            Effect.map((chunk) => Array.scan(chunk, 0, (acc, curr) => acc + curr))
          )
        })
        assert.deepStrictEqual(result1, result2)
      }))

    it.effect("scanEffect", () =>
      Effect.gen(function*() {
        const result = yield* Stream.make(1, 2, 3, 4, 5).pipe(
          Stream.scanEffect(0, (acc, curr) => Effect.succeed(acc + curr)),
          Stream.runCollect
        )

        assert.deepStrictEqual(result, Array.scan([1, 2, 3, 4, 5], 0, (acc, curr) => acc + curr))
      }))
  })

  describe("grouping", () => {
    it.effect("groupBy", () =>
      Effect.gen(function*() {
        const stream = Stream.make(1, 2, 3, 4, 5)
        const grouped = yield* stream.pipe(
          Stream.groupByKey((n) => n % 2 === 0 ? "even" : "odd"),
          Stream.mapEffect(
            Effect.fnUntraced(function*([key, stream]) {
              return [key, yield* Stream.runCollect(stream)] as const
            }),
            { concurrency: "unbounded" }
          ),
          Stream.runCollect
        )
        assert.deepStrictEqual(grouped, [
          ["odd", [1, 3, 5]],
          ["even", [2, 4]]
        ])
      }))
  })

  describe("flattening", () => {
    it.effect("flatten supports dropping parens in pipe", () =>
      Effect.gen(function*() {
        const result = yield* pipe(
          Stream.make(
            Stream.make(1, 2),
            Stream.make(3, 4)
          ),
          Stream.flatten,
          Stream.runCollect
        )
        assert.deepStrictEqual(result, [1, 2, 3, 4])
      }))

    it.effect("flattenEffect supports dropping parens in pipe", () =>
      Effect.gen(function*() {
        const result = yield* pipe(
          Stream.make(
            Effect.succeed(1),
            Effect.succeed(2),
            Effect.succeed(3)
          ),
          Stream.flattenEffect,
          Stream.runCollect
        )
        assert.deepStrictEqual(result, [1, 2, 3])
      }))
  })

  it.effect.prop(
    "rechunk",
    {
      chunks: fc.array(fc.array(fc.integer()), { minLength: 1 }),
      size: fc.integer({ min: 1, max: 100 })
    },
    Effect.fnUntraced(function*({ chunks, size }) {
      const actual = yield* Stream.fromArray(chunks).pipe(
        Stream.filter((chunk): chunk is NonEmptyArray<number> => isReadonlyArrayNonEmpty(chunk)),
        Stream.flattenArray,
        Stream.rechunk(size),
        Stream.chunks,
        Stream.runCollect
      )
      const expected = chunks.flat()
      assert.deepStrictEqual(actual, grouped(expected, size))
    })
  )

  describe("transduce", () => {
    it.effect("no remainder", () =>
      Effect.gen(function*() {
        const result = yield* Stream.make(1, 2, 3, 4).pipe(
          Stream.transduce(Sink.reduceWhile(() => 100, (n) => n % 2 === 0, (acc, n) => acc + n)),
          Stream.runCollect
        )
        assert.deepStrictEqual(result, [101, 105, 104])
      }))

    it.effect("with a sink that always signals more", () =>
      Effect.gen(function*() {
        const result = yield* Stream.make(1, 2, 3).pipe(
          Stream.transduce(Sink.reduceWhile(() => 0, constTrue, (acc, n) => acc + n)),
          Stream.runCollect
        )
        assert.deepStrictEqual(result, [6])
      }))

    it.effect("propagates scope error", () =>
      Effect.gen(function*() {
        const result = yield* Stream.make(1, 2, 3).pipe(
          Stream.transduce(Sink.fail("Woops")),
          Stream.runCollect,
          Effect.result
        )
        assertFailure(result, "Woops")
      }))
  })

  describe("buffer", () => {
    it.effect("maintains elements and ordering", () =>
      Effect.gen(function*() {
        const chunks = Array.make(
          Array.range(0, 3),
          Array.range(2, 5),
          Array.range(3, 7)
        )
        const result = yield* Stream.fromArrays(...chunks).pipe(
          Stream.buffer({ capacity: 2 }),
          Stream.runCollect
        )
        assert.deepStrictEqual(result, chunks.flat())
      }))

    it.effect("stream with a failure", () =>
      Effect.gen(function*() {
        const result = yield* Stream.range(0, 9).pipe(
          Stream.concat(Stream.fail("boom")),
          Stream.buffer({ capacity: 2 }),
          Stream.runCollect,
          Effect.exit
        )
        assertExitFailure(result, Cause.fail("boom"))
      }))

    it.effect("fast producer progresses independently", () =>
      Effect.gen(function*() {
        const arr = Array.empty<number>()
        const latch = yield* Deferred.make<void>()
        const stream = Stream.range(1, 4).pipe(
          Stream.tap(Effect.fnUntraced(function*(n) {
            arr.push(n)
            if (n === 4) {
              yield* Deferred.succeed(latch, void 0)
            }
          })),
          Stream.buffer({ capacity: 2 })
        )
        const result1 = yield* stream.pipe(Stream.take(2), Stream.runCollect)
        yield* Deferred.await(latch)
        assert.deepStrictEqual(result1, [1, 2])
        assert.deepStrictEqual(arr, [1, 2, 3, 4])
      }))
  })

  describe("bufferArray - suspend", () => {
    it.effect("maintains elements and ordering", () =>
      Effect.gen(function*() {
        const chunks = Array.make(
          Array.range(0, 3),
          Array.range(2, 5),
          Array.range(3, 7)
        )
        const result = yield* Stream.fromArrays(...chunks).pipe(
          Stream.bufferArray({ capacity: 2 }),
          Stream.runCollect
        )
        assert.deepStrictEqual(result, chunks.flat())
      }))

    it.effect("stream with a failure", () =>
      Effect.gen(function*() {
        const error = "boom"
        const result = yield* Stream.range(0, 9).pipe(
          Stream.concat(Stream.fail(error)),
          Stream.bufferArray({ capacity: 2 }),
          Stream.runCollect,
          Effect.exit
        )
        assertExitFailure(result, Cause.fail(error))
      }))

    it.effect("fast producer progresses independently", () =>
      Effect.gen(function*() {
        const arr = Array.empty<number>()
        const latch = yield* Deferred.make<void>()
        const stream = Stream.range(1, 4).pipe(
          Stream.tap(Effect.fnUntraced(function*(n) {
            arr.push(n)
            if (n === 4) {
              yield* Deferred.succeed(latch, void 0)
            }
          })),
          Stream.bufferArray({ capacity: 2 })
        )
        const result1 = yield* stream.pipe(Stream.take(2), Stream.runCollect)
        yield* Deferred.await(latch)
        assert.deepStrictEqual(result1, [1, 2])
        assert.deepStrictEqual(arr, [1, 2, 3, 4])
      }))
  })

  describe("bufferArray - dropping", () => {
    it.effect("buffers a stream with a failure", () =>
      Effect.gen(function*() {
        const error = "boom"
        const result = yield* Stream.range(1, 1_000).pipe(
          Stream.concat(Stream.fail(error)),
          Stream.concat(Stream.range(1_001, 2_000)),
          Stream.bufferArray({ capacity: 2, strategy: "dropping" }),
          Stream.runCollect,
          Effect.exit
        )
        assert.deepStrictEqual(result, Exit.fail(error))
      }))

    it.effect("fast producer progress independently", () =>
      Effect.gen(function*() {
        const arr = Array.empty<number>()
        const latch1 = yield* Deferred.make<void>()
        const latch2 = yield* Deferred.make<void>()
        const latch3 = yield* Deferred.make<void>()
        const latch4 = yield* Deferred.make<void>()
        const stream1 = Stream.make(0).pipe(
          Stream.concat(
            Stream.fromEffect(Deferred.await(latch1)).pipe(
              Stream.flatMap(() =>
                Stream.range(1, 16).pipe(
                  Stream.rechunk(1),
                  Stream.ensuring(Deferred.succeed(latch2, void 0))
                )
              )
            )
          )
        )
        const stream2 = Stream.fromEffect(Deferred.await(latch3)).pipe(
          Stream.flatMap(() =>
            Stream.range(17, 24).pipe(
              Stream.rechunk(1),
              Stream.ensuring(Deferred.succeed(latch4, void 0))
            )
          )
        )
        const stream3 = Stream.make(-1)
        const stream = stream1.pipe(
          Stream.concat(stream2),
          Stream.concat(stream3),
          Stream.bufferArray({ capacity: 8, strategy: "dropping" })
        )
        const { result1, result2, result3 } = yield* Stream.toPull(stream).pipe(
          Effect.flatMap((pull) =>
            Effect.gen(function*() {
              const result1 = yield* pull
              yield* Deferred.succeed(latch1, void 0)
              yield* Deferred.await(latch2)
              yield* pull.pipe(
                Effect.tap((chunk) =>
                  Effect.sync(() => {
                    arr.push(...chunk)
                  })
                ),
                Effect.repeat({ times: 7 })
              )
              const result2 = arr.slice()
              yield* Deferred.succeed(latch3, void 0)
              yield* Deferred.await(latch4)
              yield* pull.pipe(
                Effect.tap((chunk) =>
                  Effect.sync(() => {
                    arr.push(...chunk)
                  })
                ),
                Effect.repeat({ times: 7 })
              )
              const result3 = arr.slice()
              return { result1, result2, result3 }
            })
          ),
          Effect.scoped
        )
        const expected1 = Array.of(0)
        const expected2 = [1, 2, 3, 4, 5, 6, 7, 8]
        const expected3 = [1, 2, 3, 4, 5, 6, 7, 8, 17, 18, 19, 20, 21, 22, 23, 24]
        assert.deepStrictEqual(result1, expected1)
        assert.deepStrictEqual(result2, expected2)
        assert.deepStrictEqual(result3, expected3)
      }))

    it.effect("buffers a stream with a failure", () =>
      Effect.gen(function*() {
        const error = "boom"
        const result = yield* pipe(
          Stream.range(1, 1_000),
          Stream.concat(Stream.fail(error)),
          Stream.concat(Stream.range(1_000, 2_000)),
          Stream.buffer({ capacity: 2, strategy: "dropping" }),
          Stream.runCollect,
          Effect.exit
        )
        assert.deepStrictEqual(result, Exit.fail(error))
      }))

    it.effect("fast producer progress independently", () =>
      Effect.gen(function*() {
        const ref = yield* Ref.make(Array.empty<number>())
        const latch1 = yield* Deferred.make<void>()
        const latch2 = yield* Deferred.make<void>()
        const latch3 = yield* Deferred.make<void>()
        const latch4 = yield* Deferred.make<void>()
        const stream1 = pipe(
          Stream.make(0),
          Stream.concat(
            pipe(
              Stream.fromEffect(Deferred.await(latch1)),
              Stream.flatMap(() =>
                pipe(
                  Stream.range(1, 17),
                  Stream.rechunk(1),
                  Stream.ensuring(Deferred.succeed(latch2, void 0))
                )
              )
            )
          )
        )
        const stream2 = pipe(
          Stream.fromEffect(Deferred.await(latch3)),
          Stream.flatMap(() =>
            pipe(
              Stream.range(17, 24),
              Stream.rechunk(1),
              Stream.ensuring(Deferred.succeed(latch4, void 0))
            )
          )
        )
        const stream3 = Stream.make(-1)
        const stream = pipe(
          stream1,
          Stream.concat(stream2),
          Stream.concat(stream3),
          Stream.buffer({ capacity: 8, strategy: "dropping" })
        )
        const { result1, result2, result3 } = yield* pipe(
          Stream.toPull(stream),
          Effect.flatMap((pull) =>
            Effect.gen(function*() {
              const result1 = yield* pull
              yield* Deferred.succeed(latch1, void 0)
              yield* Deferred.await(latch2)
              yield* pipe(
                pull,
                Effect.flatMap((chunk) => Ref.update(ref, Array.appendAll(chunk)))
              )
              const result2 = yield* Ref.get(ref)
              yield* Deferred.succeed(latch3, void 0)
              yield* Deferred.await(latch4)
              yield* pipe(
                pull,
                Effect.flatMap((chunk) => Ref.update(ref, Array.appendAll(chunk)))
              )
              const result3 = yield* (Ref.get(ref))
              return { result1, result2, result3 }
            })
          ),
          Effect.scoped
        )
        const expected1 = [0] as const
        const expected2 = [1, 2, 3, 4, 5, 6, 7, 8]
        const expected3 = [1, 2, 3, 4, 5, 6, 7, 8, 17, 18, 19, 20, 21, 22, 23, 24]
        deepStrictEqual(result1, expected1)
        deepStrictEqual(result2, expected2)
        deepStrictEqual(result3, expected3)
      }))
  })

  describe("bufferArray - sliding", () => {
    it.effect("buffers a stream with a failure", () =>
      Effect.gen(function*() {
        const error = "boom"
        const result = yield* Stream.range(1, 1_000).pipe(
          Stream.concat(Stream.fail(error)),
          Stream.concat(Stream.range(1_001, 2_000)),
          Stream.bufferArray({ capacity: 2, strategy: "sliding" }),
          Stream.runCollect,
          Effect.exit
        )
        assertExitFailure(result, Cause.fail(error))
      }))

    it.effect("fast producer progress independently", () =>
      Effect.gen(function*() {
        const ref = yield* Ref.make(Array.empty<number>())

        const latch1 = yield* Deferred.make<void>()
        const latch2 = yield* Deferred.make<void>()
        const latch3 = yield* Deferred.make<void>()
        const latch4 = yield* Deferred.make<void>()
        const latch5 = yield* Deferred.make<void>()
        const stream1 = Stream.make(0).pipe(
          Stream.concat(
            pipe(
              Stream.fromEffect(Deferred.await(latch1)),
              Stream.flatMap(() =>
                pipe(
                  Stream.range(1, 16),
                  Stream.rechunk(1),
                  Stream.ensuring(Deferred.succeed(latch2, void 0))
                )
              )
            )
          )
        )
        const stream2 = pipe(
          Stream.fromEffect(Deferred.await(latch3)),
          Stream.flatMap(() =>
            pipe(
              Stream.range(17, 25),
              Stream.rechunk(1),
              Stream.ensuring(Deferred.succeed(latch4, void 0))
            )
          )
        )
        const stream3 = pipe(
          Stream.fromEffect(Deferred.await(latch5)),
          Stream.flatMap(() => Stream.make(-1))
        )
        const stream = pipe(
          stream1,
          Stream.concat(stream2),
          Stream.concat(stream3),
          Stream.bufferArray({ capacity: 8, strategy: "sliding" })
        )
        const { result1, result2, result3 } = yield* pipe(
          Stream.toPull(stream),
          Effect.flatMap((pull) =>
            Effect.gen(function*() {
              const result1 = yield* pull
              yield* Deferred.succeed(latch1, void 0)
              yield* Deferred.await(latch2)
              yield* pipe(
                pull,
                Effect.flatMap((chunk) => Ref.update(ref, Array.appendAll(chunk))),
                Effect.repeat({ times: 7 })
              )
              const result2 = yield* Ref.get(ref)
              yield* Deferred.succeed(latch3, void 0)
              yield* Deferred.await(latch4)
              yield* pipe(
                pull,
                Effect.flatMap((chunk) => Ref.update(ref, Array.appendAll(chunk))),
                Effect.repeat({ times: 7 })
              )
              const result3 = yield* (Ref.get(ref))
              return { result1, result2, result3 }
            })
          ),
          Effect.scoped
        )
        const expected1 = [0]
        const expected2 = [9, 10, 11, 12, 13, 14, 15, 16]
        const expected3 = [9, 10, 11, 12, 13, 14, 15, 16, 18, 19, 20, 21, 22, 23, 24, 25]
        assert.deepStrictEqual(Array.fromIterable(result1), expected1)
        assert.deepStrictEqual(Array.fromIterable(result2), expected2)
        assert.deepStrictEqual(Array.fromIterable(result3), expected3)
      }))

    it.effect("buffers a stream with a failure", () =>
      Effect.gen(function*() {
        const error = "boom"
        const result = yield* pipe(
          Stream.range(1, 1_000),
          Stream.concat(Stream.fail(error)),
          Stream.concat(Stream.range(1_001, 2_000)),
          Stream.buffer({ capacity: 2, strategy: "sliding" }),
          Stream.runCollect,
          Effect.exit
        )
        deepStrictEqual(result, Exit.fail(error))
      }))

    it.effect("fast producer progress independently", () =>
      Effect.gen(function*() {
        const ref = yield* Ref.make(Array.empty<number>())
        const latch1 = yield* Deferred.make<void>()
        const latch2 = yield* Deferred.make<void>()
        const latch3 = yield* Deferred.make<void>()
        const latch4 = yield* Deferred.make<void>()
        const stream1 = pipe(
          Stream.make(0),
          Stream.concat(
            pipe(
              Stream.fromEffect(Deferred.await(latch1)),
              Stream.flatMap(() =>
                pipe(
                  Stream.range(1, 16),
                  Stream.rechunk(1),
                  Stream.ensuring(Deferred.succeed(latch2, void 0))
                )
              )
            )
          )
        )
        const stream2 = pipe(
          Stream.fromEffect(Deferred.await(latch3)),
          Stream.flatMap(() =>
            pipe(
              Stream.range(17, 24),
              Stream.rechunk(1),
              Stream.ensuring(Deferred.succeed(latch4, void 0))
            )
          )
        )
        const stream3 = Stream.make(-1)
        const stream = pipe(
          stream1,
          Stream.concat(stream2),
          Stream.concat(stream3),
          Stream.buffer({ capacity: 8, strategy: "sliding" })
        )
        const { result1, result2, result3 } = yield* pipe(
          Stream.toPull(stream),
          Effect.flatMap((pull) =>
            Effect.gen(function*() {
              const result1 = yield* pull
              yield* Deferred.succeed(latch1, void 0)
              yield* Deferred.await(latch2)
              yield* pipe(
                pull,
                Effect.flatMap((chunk) => Ref.update(ref, Array.appendAll(chunk)))
              )
              const result2 = yield* (Ref.get(ref))
              yield* (Deferred.succeed(latch3, void 0))
              yield* (Deferred.await(latch4))
              yield* pipe(
                pull,
                Effect.flatMap((chunk) => Ref.update(ref, Array.appendAll(chunk))),
                Effect.repeat({ times: 7 })
              )
              const result3 = yield* (Ref.get(ref))
              return { result1, result2, result3 }
            })
          ),
          Effect.scoped
        )
        const expected1 = [0] as const
        const expected2 = [9, 10, 11, 12, 13, 14, 15, 16]
        const expected3 = [9, 10, 11, 12, 13, 14, 15, 16, 18, 19, 20, 21, 22, 23, 24, -1]
        deepStrictEqual(result1, expected1)
        deepStrictEqual(result2, expected2)
        deepStrictEqual(result3, expected3)
      }))

    it.effect("propagates defects", () =>
      Effect.gen(function*() {
        const result = yield* pipe(
          Stream.fromEffect(Effect.die("boom")),
          Stream.buffer({ capacity: 1, strategy: "sliding" }),
          Stream.runDrain,
          Effect.exit
        )
        deepStrictEqual(result, Exit.die("boom"))
      }))
  })

  describe("buffer - unbounded", () => {
    it.effect("buffer - buffers the stream", () =>
      Effect.gen(function*() {
        const chunk = Array.range(0, 10)
        const result = yield* pipe(
          Stream.fromIterable(chunk),
          Stream.buffer({ capacity: "unbounded" }),
          Stream.runCollect
        )
        deepStrictEqual(result, chunk)
      }))

    it.effect("buffers a stream with a failure", () =>
      Effect.gen(function*() {
        const error = "boom"
        const result = yield* pipe(
          Stream.range(0, 9),
          Stream.concat(Stream.fail(error)),
          Stream.buffer({ capacity: "unbounded" }),
          Stream.runCollect,
          Effect.exit
        )
        deepStrictEqual(result, Exit.fail(error))
      }))

    it.effect("fast producer progress independently", () =>
      Effect.gen(function*() {
        const ref = yield* Ref.make(Array.empty<number>())
        const latch = yield* Deferred.make<void>()
        const stream = pipe(
          Stream.range(1, 999),
          Stream.tap((n) =>
            pipe(
              Ref.update(ref, Array.append(n)),
              Effect.andThen(pipe(Deferred.succeed(latch, void 0), Effect.when(Effect.sync(() => n === 999))))
            )
          ),
          Stream.rechunk(999),
          Stream.buffer({ capacity: "unbounded" })
        )
        const result1 = yield* pipe(stream, Stream.take(2), Stream.runCollect)
        yield* Deferred.await(latch)
        const result2 = yield* Ref.get(ref)
        deepStrictEqual(result1, [1, 2])
        deepStrictEqual(result2, Array.range(1, 999))
      }))
  })

  describe("share", () => {
    it.effect("sequenced", () =>
      Effect.gen(function*() {
        const sharedStream = yield* Stream.fromSchedule(Schedule.spaced("1 seconds")).pipe(
          Stream.share({ capacity: 16 })
        )

        const firstFiber = yield* sharedStream.pipe(
          Stream.take(1),
          Stream.run(Sink.collect()),
          Effect.forkChild({ startImmediately: true })
        )

        yield* TestClock.adjust("1 seconds")

        const first = yield* Fiber.join(firstFiber)
        deepStrictEqual(first, [0])

        const secondFiber = yield* sharedStream.pipe(
          Stream.take(1),
          Stream.run(Sink.collect()),
          Effect.forkChild({ startImmediately: true })
        )

        yield* TestClock.adjust("1 seconds")

        const second = yield* Fiber.join(secondFiber)
        deepStrictEqual(second, [0])
      }))

    it.effect("sequenced with idleTimeToLive", () =>
      Effect.gen(function*() {
        const sharedStream = yield* Stream.fromSchedule(Schedule.spaced("1 seconds")).pipe(
          Stream.share({
            capacity: 16,
            idleTimeToLive: "1 second"
          })
        )

        const firstFiber = yield* sharedStream.pipe(
          Stream.take(1),
          Stream.run(Sink.collect()),
          Effect.forkChild({ startImmediately: true })
        )

        yield* TestClock.adjust("1 seconds")

        const first = yield* Fiber.join(firstFiber)
        deepStrictEqual(first, [0])

        const secondFiber = yield* sharedStream.pipe(
          Stream.take(1),
          Stream.run(Sink.collect()),
          Effect.forkChild({ startImmediately: true })
        )

        yield* TestClock.adjust("1 seconds")

        const second = yield* Fiber.join(secondFiber)
        deepStrictEqual(second, [1])
      }))

    it.effect("parallel", () =>
      Effect.gen(function*() {
        const sharedStream = yield* Stream.fromSchedule(Schedule.spaced("1 seconds")).pipe(
          Stream.share({ capacity: 16 })
        )

        const fiber1 = yield* sharedStream.pipe(
          Stream.take(1),
          Stream.run(Sink.collect()),
          Effect.forkChild({ startImmediately: true })
        )
        const fiber2 = yield* sharedStream.pipe(
          Stream.take(2),
          Stream.run(Sink.collect()),
          Effect.forkChild({ startImmediately: true })
        )

        yield* TestClock.adjust("2 seconds")

        const [result1, result2] = yield* Fiber.joinAll([fiber1, fiber2])

        deepStrictEqual(result1, [0])
        deepStrictEqual(result2, [0, 1])
      }))
  })

  describe("raceAll", () => {
    it.effect("sync", () =>
      Effect.gen(function*() {
        const result = yield* pipe(
          Stream.raceAll(
            Stream.make(0, 1, 2, 3),
            Stream.make(4, 5, 6, 7),
            Stream.make(7, 8, 9, 10)
          ),
          Stream.runCollect
        )
        deepStrictEqual(result, [0, 1, 2, 3])
      }))

    it.effect("async", () =>
      Effect.gen(function*() {
        const fiber = yield* pipe(
          Stream.raceAll(
            Stream.fromSchedule(Schedule.spaced("1 second")),
            Stream.fromSchedule(Schedule.spaced("2 second"))
          ),
          Stream.take(5),
          Stream.runCollect,
          Effect.forkScoped
        )
        yield* TestClock.adjust("5 second")
        const result = yield* Fiber.join(fiber)
        deepStrictEqual(result, [0, 1, 2, 3, 4])
      }))

    it.effect("combined async + sync", () =>
      Effect.gen(function*() {
        const result = yield* pipe(
          Stream.raceAll(
            Stream.fromSchedule(Schedule.spaced("1 second")),
            Stream.make(0, 1, 2, 3)
          ),
          Stream.runCollect
        )
        deepStrictEqual(result, [0, 1, 2, 3])
      }))

    it.effect("combined sync + async", () =>
      Effect.gen(function*() {
        const result = yield* pipe(
          Stream.raceAll(
            Stream.make(0, 1, 2, 3),
            Stream.fromSchedule(Schedule.spaced("1 second"))
          ),
          Stream.runCollect
        )
        deepStrictEqual(result, [0, 1, 2, 3])
      }))
  })

  it.effect("onStart", () =>
    Effect.gen(function*() {
      let counter = 0
      const result = yield* pipe(
        Stream.make(1, 1),
        Stream.onStart(Effect.sync(() => counter++)),
        Stream.runCollect
      )
      strictEqual(counter, 1)
      deepStrictEqual(result, [1, 1])
    }))

  it.effect("onEnd", () =>
    Effect.gen(function*() {
      let counter = 0
      const result = yield* pipe(
        Stream.make(1, 2, 3),
        Stream.onEnd(Effect.sync(() => counter++)),
        Stream.runCollect
      )
      strictEqual(counter, 1)
      deepStrictEqual(result, [1, 2, 3])
    }))

  describe("groupAdjacentBy", () => {
    it.effect("one big chunk", () =>
      Effect.gen(function*() {
        const result = yield* pipe(
          Stream.fromIterable([
            { code: 1, message: "A" },
            { code: 1, message: "B" },
            { code: 1, message: "D" },
            { code: 2, message: "C" }
          ]),
          Stream.groupAdjacentBy((x) => x.code),
          Stream.runCollect
        )
        deepStrictEqual(
          result.map(([, chunk]) => chunk),
          [
            [
              { code: 1, message: "A" },
              { code: 1, message: "B" },
              { code: 1, message: "D" }
            ],
            [
              { code: 2, message: "C" }
            ]
          ]
        )
      }))

    it.effect("several single element chunks", () =>
      Effect.gen(function*() {
        const result = yield* pipe(
          Stream.fromArrays(
            [{ code: 1, message: "A" }],
            [{ code: 1, message: "B" }],
            [{ code: 1, message: "D" }],
            [{ code: 2, message: "C" }]
          ),
          Stream.groupAdjacentBy((x) => x.code),
          Stream.runCollect
        )
        deepStrictEqual(
          result.map(([, chunk]) => chunk),
          [
            [
              { code: 1, message: "A" },
              { code: 1, message: "B" },
              { code: 1, message: "D" }
            ],
            [
              { code: 2, message: "C" }
            ]
          ]
        )
      }))

    it.effect("group across chunks", () =>
      Effect.gen(function*() {
        const result = yield* pipe(
          Stream.fromArrays(
            [{ code: 1, message: "A" }, { code: 1, message: "B" }],
            [{ code: 1, message: "D" }, { code: 2, message: "C" }]
          ),
          Stream.groupAdjacentBy((x) => x.code),
          Stream.runCollect
        )
        deepStrictEqual(
          result.map(([, chunk]) => chunk),
          [
            [
              { code: 1, message: "A" },
              { code: 1, message: "B" },
              { code: 1, message: "D" }
            ],
            [
              { code: 2, message: "C" }
            ]
          ]
        )
      }))
  })

  describe("aggregateWithin", () => {
    it.effect("groupedWithin does not emit empty arrays when upstream is idle", () =>
      Effect.gen(function*() {
        const fiber = yield* Stream.never.pipe(
          Stream.groupedWithin(25, "50 millis"),
          Stream.take(5),
          Stream.runCollect,
          Effect.forkChild({ startImmediately: true })
        )
        yield* TestClock.adjust("1 second")
        assert.isUndefined(fiber.pollUnsafe())
      }))

    it.effect("groupedWithin flushes final partial batch on upstream end without waiting for the schedule", () =>
      Effect.gen(function*() {
        const result = yield* Stream.make(1, 2, 3, 4, 5).pipe(
          Stream.groupedWithin(3, "1 second"),
          Stream.runCollect
        )
        deepStrictEqual(result, [[1, 2, 3], [4, 5]])
      }))

    it.effect("groupedWithin flushes partial batch when the schedule elapses while upstream is idle", () =>
      Effect.gen(function*() {
        const fiber = yield* Stream.make(1, 2, 3, 4, 5).pipe(
          Stream.concat(Stream.never),
          Stream.groupedWithin(3, "1 second"),
          Stream.take(2),
          Stream.runCollect,
          Effect.forkChild({ startImmediately: true })
        )
        yield* TestClock.adjust("1 second")
        const result = yield* Fiber.join(fiber)
        deepStrictEqual(result, [[1, 2, 3], [4, 5]])
      }))

    it.effect("simple example", () =>
      Effect.gen(function*() {
        const result = yield* pipe(
          Stream.make(1, 1, 1, 1, 2, 2),
          Stream.aggregateWithin(
            pipe(
              Sink.fold(
                () => [[] as Array<number>, true] as readonly [Array<number>, boolean],
                (tuple) => tuple[1],
                ([array], curr: number) =>
                  Effect.succeed<readonly [Array<number>, boolean]>(
                    curr === 1 ? [[curr, ...array], true] : [[curr, ...array], false]
                  )
              ),
              Sink.map((tuple) => tuple[0])
            ),
            Schedule.spaced(Duration.minutes(30))
          ),
          Stream.runCollect
        )
        deepStrictEqual(result, [[2, 1, 1, 1, 1], [2]])
      }))

    it.effect("fails fast", () =>
      Effect.gen(function*() {
        const queue = yield* Queue.unbounded<number>()
        yield* pipe(
          Stream.range(1, 9),
          Stream.tap((n) =>
            pipe(
              Effect.fail("Boom"),
              Effect.when(Effect.sync(() => n === 6)),
              Effect.andThen(Queue.offer(queue, n))
            )
          ),
          Stream.aggregateWithin(
            Sink.foldUntil(constVoid, 5, () => Effect.void),
            Schedule.forever
          ),
          Stream.runDrain,
          Effect.catch(() => Effect.void)
        )
        const result = yield* Queue.takeAll(queue)
        yield* Queue.shutdown(queue)
        deepStrictEqual(result, [1, 2, 3, 4, 5])
      }))

    it.effect("error propagation #1", () =>
      Effect.gen(function*() {
        const error = new Error("Boom")
        const result = yield* pipe(
          Stream.make(1, 1, 1, 1),
          Stream.aggregateWithin(
            Sink.die(error),
            Schedule.spaced(Duration.minutes(30))
          ),
          Stream.runCollect,
          Effect.exit
        )
        deepStrictEqual(result, Exit.die(error))
      }))

    it.effect("error propagation #2", () =>
      Effect.gen(function*() {
        const error = new Error("Boom")
        const result = yield* pipe(
          Stream.make(1, 1),
          Stream.aggregateWithin(
            Sink.fold(() => [], constTrue, () => Effect.die(error)),
            Schedule.spaced(Duration.minutes(30))
          ),
          Stream.runCollect,
          Effect.exit
        )
        deepStrictEqual(result, Exit.die(error))
      }))

    it.effect("interruption propagation #1", () =>
      Effect.gen(function*() {
        const latch = yield* Deferred.make<void>()
        const ref = yield* Ref.make(false)
        const sink = Sink.fold(Array.empty<number>, constTrue, (acc, curr: number) => {
          if (curr === 1) {
            acc.push(curr)
            return Effect.succeed(acc)
          }
          return pipe(
            Deferred.succeed(latch, void 0),
            Effect.andThen(Effect.never),
            Effect.onInterrupt(() => Ref.set(ref, true))
          )
        })
        const fiber = yield* pipe(
          Stream.make(1, 1, 2),
          Stream.aggregateWithin(sink, Schedule.spaced(Duration.minutes(30))),
          Stream.runCollect,
          Effect.forkScoped
        )
        yield* Deferred.await(latch)
        yield* Fiber.interrupt(fiber)
        const result = yield* Ref.get(ref)
        assertTrue(result)
      }))

    it.effect("interruption propagation #2", () =>
      Effect.gen(function*() {
        const ref = yield* Ref.make(false)
        const sink = Sink.fromEffect(
          Effect.never.pipe(
            Effect.onInterrupt(() => Ref.set(ref, true))
          )
        )
        const fiber = yield* Stream.make(1, 1, 2).pipe(
          Stream.aggregateWithin(sink, Schedule.spaced(Duration.minutes(30))),
          Stream.runCollect,
          Effect.forkScoped({ startImmediately: true })
        )
        yield* Fiber.interrupt(fiber)
        const result = yield* Ref.get(ref)
        assertTrue(result)
      }))

    // it.effect("leftover handling", () =>
    //   Effect.gen(function*() {
    //     const input = [1, 2, 2, 3, 2, 3]
    //     const fiber = yield* pipe(
    //       Stream.fromIterable(input),
    //       Stream.aggregateWithin(
    //         Sink.foldWeighted({
    //           initial: Chunk.empty<number>(),
    //           maxCost: 4,
    //           cost: (_, n) => n,
    //           body: (acc, curr) => Chunk.append(acc, curr)
    //         }),
    //         Schedule.spaced(Duration.millis(100))
    //       ),
    //       Stream.flattenIterable,
    //       Stream.runCollect,
    //       Effect.forkScoped
    //     )
    //     yield* TestClock.adjust(Duration.minutes(31))
    //     const result = yield* Fiber.join(fiber)
    //     deepStrictEqual(result, input)
    //   }))
  })

  describe("debounce", () => {
    it.effect("should drop earlier chunks within waitTime", () =>
      Effect.gen(function*() {
        const coordination = yield* chunkCoordination([
          [1],
          [3, 4],
          [5],
          [6, 7]
        ])
        const stream = pipe(
          coordination.stream,
          Stream.chunks,
          Stream.debounce(Duration.seconds(1))
        )
        const fiber = yield* pipe(stream, Stream.runCollect, Effect.forkScoped)
        yield* coordination.offer
        yield* pipe(
          Effect.sleep(Duration.millis(500)),
          Effect.andThen(coordination.offer),
          Effect.forkScoped
        )
        yield* pipe(
          Effect.sleep(Duration.seconds(2)),
          Effect.andThen(coordination.offer),
          Effect.forkScoped
        )
        yield* pipe(
          Effect.sleep(Duration.millis(2500)),
          Effect.andThen(coordination.offer),
          Effect.forkScoped
        )
        yield* TestClock.adjust(Duration.millis(3500))
        const result = yield* Fiber.join(fiber)
        deepStrictEqual(
          result,
          [[3, 4], [6, 7]]
        )
      }))

    it.effect("should take latest chunk within waitTime", () =>
      Effect.gen(function*() {
        const coordination = yield* chunkCoordination([
          [1, 2],
          [3, 4],
          [5, 6]
        ])
        const stream = pipe(
          coordination.stream,
          Stream.chunks,
          Stream.debounce(Duration.seconds(1))
        )
        const fiber = yield* pipe(stream, Stream.runCollect, Effect.forkScoped)
        yield* pipe(
          coordination.offer,
          Effect.andThen(coordination.offer),
          Effect.andThen(coordination.offer)
        )
        yield* TestClock.adjust(Duration.seconds(1))
        const result = yield* Fiber.join(fiber)
        deepStrictEqual(result, [[5, 6]])
      }))

    it.effect("should work properly with parallelization", () =>
      Effect.gen(function*() {
        const coordination = yield* chunkCoordination([
          [1],
          [2],
          [3]
        ])
        const stream = pipe(
          coordination.stream,
          Stream.chunks,
          Stream.debounce(Duration.seconds(1))
        )
        const fiber = yield* pipe(stream, Stream.runCollect, Effect.forkScoped)
        yield* Effect.all([
          coordination.offer,
          coordination.offer,
          coordination.offer
        ], { concurrency: 3, discard: true })
        yield* TestClock.adjust(Duration.seconds(1))
        const result = yield* Fiber.join(fiber)
        deepStrictEqual(result, [[3]])
      }))

    it.effect("should handle empty chunks properly", () =>
      Effect.gen(function*() {
        const fiber = yield* pipe(
          Stream.make(1, 2, 3),
          Stream.tap(() => Effect.sleep(Duration.millis(500))),
          Stream.debounce(Duration.seconds(1)),
          Stream.runCollect,
          Effect.forkScoped({ startImmediately: true })
        )
        yield* TestClock.adjust(Duration.seconds(3))
        const result = yield* Fiber.join(fiber)
        deepStrictEqual(result, [3])
      }))

    it.effect("should fail immediately", () =>
      Effect.gen(function*() {
        const result = yield* pipe(
          Stream.fromEffect(Effect.fail(Option.none())),
          Stream.debounce(Duration.infinity),
          Stream.runCollect,
          Effect.exit
        )
        assertExitFailure(result, Cause.fail(Option.none()))
      }))

    it.effect("should work with empty streams", () =>
      Effect.gen(function*() {
        const result = yield* pipe(
          Stream.empty,
          Stream.debounce(Duration.seconds(5)),
          Stream.runCollect
        )
        assertTrue(result.length === 0)
      }))

    it.effect("should pick last element from every chunk", () =>
      Effect.gen(function*() {
        const fiber = yield* pipe(
          Stream.make(1, 2, 3),
          Stream.debounce(Duration.seconds(1)),
          Stream.runCollect,
          Effect.forkScoped
        )
        yield* TestClock.adjust(Duration.seconds(1))
        const result = yield* Fiber.join(fiber)
        deepStrictEqual(result, [3])
      }))

    it.effect("should interrupt fibers properly", () =>
      Effect.gen(function*() {
        const coordination = yield* chunkCoordination([
          [1],
          [2],
          [3]
        ])
        const fiber = yield* pipe(
          coordination.stream,
          Stream.chunks,
          Stream.tap(() => coordination.proceed),
          Stream.flattenArray,
          Stream.debounce(Duration.millis(200)),
          // Stream.interruptWhen(Effect.never),
          Stream.take(1),
          Stream.runCollect,
          Effect.forkScoped
        )
        yield* pipe(
          coordination.offer,
          Effect.andThen(TestClock.adjust(Duration.millis(100))),
          Effect.andThen(coordination.awaitNext),
          Effect.repeat({ times: 2 })
        )
        yield* TestClock.adjust(Duration.millis(100))
        const result = yield* Fiber.join(fiber)
        deepStrictEqual(result, [3])
      }))

    it.effect("should interrupt children fiber on stream interruption", () =>
      Effect.gen(function*() {
        const ref = yield* (Ref.make(false))
        const fiber = yield* pipe(
          Stream.fromEffect(Effect.void),
          Stream.concat(Stream.fromEffect(pipe(
            Effect.never,
            Effect.onInterrupt(() => Ref.set(ref, true))
          ))),
          Stream.debounce(Duration.millis(800)),
          Stream.runDrain,
          Effect.forkScoped
        )
        yield* TestClock.adjust(Duration.minutes(1))
        yield* Fiber.interrupt(fiber)
        const result = yield* Ref.get(ref)
        assertTrue(result)
      }))
  })

  describe("throttle", () => {
    it.effect("throttleEnforce - free elements", () =>
      Effect.gen(function*() {
        const result = yield* pipe(
          Stream.make(1, 2, 3, 4),
          Stream.throttle({
            cost: () => 0,
            units: 0,
            duration: Duration.infinity,
            strategy: "enforce"
          }),
          Stream.runCollect
        )
        deepStrictEqual(result, [1, 2, 3, 4])
      }))

    it.effect("throttleEnforce - no bandwidth", () =>
      Effect.gen(function*() {
        const result = yield* pipe(
          Stream.make(1, 2, 3, 4),
          Stream.throttle({
            cost: () => 1,
            units: 0,
            duration: Duration.infinity,
            strategy: "enforce"
          }),
          Stream.runCollect
        )
        assertTrue(result.length === 0)
      }))

    it.effect("throttleEnforce - refill bucket tokens", () =>
      Effect.gen(function*() {
        const fiber = yield* pipe(
          Stream.fromSchedule(Schedule.spaced(Duration.millis(100))),
          Stream.take(10),
          Stream.throttle({
            cost: () => 1,
            units: 1,
            duration: Duration.millis(200),
            strategy: "enforce"
          }),
          Stream.runCollect,
          Effect.forkScoped
        )
        yield* TestClock.adjust(Duration.seconds(1))
        const result = yield* Fiber.join(fiber)
        deepStrictEqual(result, [0, 2, 4, 6, 8])
      }))

    it.effect("throttleShape", () =>
      Effect.gen(function*() {
        const queue = yield* Queue.unbounded<number>()
        const fiber = yield* pipe(
          Stream.fromQueue(queue),
          Stream.throttle({
            strategy: "shape",
            cost: (arr) => arr.reduce((a, b) => a + b, 0),
            units: 1,
            duration: Duration.seconds(1)
          }),
          Stream.toPull,
          Effect.flatMap((pull) =>
            Effect.gen(function*() {
              yield* Queue.offer(queue, 1)
              const result1 = yield* pull
              yield* Queue.offer(queue, 2)
              const result2 = yield* pull
              yield* Effect.sleep(Duration.seconds(4))
              yield* Queue.offer(queue, 3)
              const result3 = yield* pull
              return [result1, result2, result3] as const
            })
          ),
          Effect.scoped,
          Effect.forkScoped
        )
        yield* TestClock.adjust(Duration.seconds(8))
        const result = yield* Fiber.join(fiber)
        deepStrictEqual(result, [[1], [2], [3]])
      }))

    it.effect("throttleShape - infinite bandwidth", () =>
      Effect.gen(function*() {
        const queue = yield* Queue.unbounded<number>()
        const result = yield* pipe(
          Stream.fromQueue(queue),
          Stream.throttle({
            strategy: "shape",
            cost: () => 100_000,
            units: 1,
            duration: Duration.zero
          }),
          Stream.toPull,
          Effect.flatMap((pull) =>
            Effect.gen(function*() {
              yield* Queue.offer(queue, 1)
              const result1 = yield* pull
              yield* Queue.offer(queue, 2)
              const result2 = yield* pull
              return [result1, result2] as const
            })
          ),
          Effect.scoped
        )
        deepStrictEqual(result, [[1], [2]])
      }))

    it.effect("throttleShape - with burst", () =>
      Effect.gen(function*() {
        const queue = yield* Queue.unbounded<number>()
        const fiber = yield* pipe(
          Stream.fromQueue(queue),
          Stream.throttle({
            strategy: "shape",
            cost: (arr) => arr.reduce((a, b) => a + b, 0),
            units: 1,
            duration: Duration.seconds(1),
            burst: 2
          }),
          Stream.toPull,
          Effect.flatMap((pull) =>
            Effect.gen(function*() {
              yield* Queue.offer(queue, 1)
              const result1 = yield* pull
              yield* TestClock.adjust(Duration.seconds(2))
              yield* Queue.offer(queue, 2)
              const result2 = yield* pull
              yield* TestClock.adjust(Duration.seconds(4))
              yield* Queue.offer(queue, 3)
              const result3 = yield* pull
              return [result1, result2, result3] as const
            })
          ),
          Effect.scoped,
          Effect.forkScoped
        )
        const result = yield* Fiber.join(fiber)
        deepStrictEqual(result, [[1], [2], [3]])
      }))

    it.effect("throttleShape - free elements", () =>
      Effect.gen(function*() {
        const result = yield* pipe(
          Stream.make(1, 2, 3, 4),
          Stream.throttle({
            strategy: "shape",
            cost: () => 0,
            units: 1,
            duration: Duration.infinity
          }),
          Stream.runCollect
        )
        deepStrictEqual(result, [1, 2, 3, 4])
      }))
  })

  describe("zipping", () => {
    it.effect("zipWith - combines elements point-wise", () =>
      Effect.gen(function*() {
        const stream1 = Stream.make(1, 2, 3)
        const stream2 = Stream.make("a", "b", "c")
        const result = yield* Stream.zipWith(stream1, stream2, (n, s) => `${n}-${s}`).pipe(
          Stream.runCollect
        )
        assert.deepStrictEqual(result, ["1-a", "2-b", "3-c"])
      }))

    it.effect("zipWith - stops when shortest stream ends", () =>
      Effect.gen(function*() {
        const stream1 = Stream.make(1, 2, 3, 4, 5, 6)
        const stream2 = Stream.make("a", "b", "c")
        const result = yield* Stream.zipWith(stream1, stream2, (n, s) => `${n}-${s}`).pipe(
          Stream.runCollect
        )
        assert.deepStrictEqual(result, ["1-a", "2-b", "3-c"])
      }))

    it.effect("zipWith - does not pull too much when one stream ends", () =>
      Effect.gen(function*() {
        const left = Stream.fromArrays([1, 2], [3, 4], [5]).pipe(
          Stream.concat(Stream.fail("boom"))
        )
        const right = Stream.fromArrays(["a", "b"], ["c"])
        const result = yield* Stream.zipWith(left, right, (n, s) => [n, s] as const).pipe(
          Stream.runCollect
        )
        assert.deepStrictEqual(result, [[1, "a"], [2, "b"], [3, "c"]])
      }))

    it.effect("zipWith - prioritizes failures", () =>
      Effect.gen(function*() {
        const result = yield* Stream.zipWith(
          Stream.never,
          Stream.fail("Ouch"),
          (a, b) => [a, b]
        ).pipe(
          Stream.runCollect,
          Effect.exit
        )
        assertExitFailure(result, Cause.fail("Ouch"))
      }))

    it.effect("zipWith - handles exceptions", () =>
      Effect.gen(function*() {
        const error = new Error("Ouch")
        const result = yield* Stream.make(1).pipe(
          Stream.flatMap(() =>
            Stream.sync(() => {
              throw error
            })
          ),
          Stream.zipWith(Stream.make(1), (a, b) => [a, b]),
          Stream.runCollect,
          Effect.exit
        )
        assertExitFailure(result, Cause.die(error))
      }))

    it.effect("zipWith - handles empty streams", () =>
      Effect.gen(function*() {
        const result1 = yield* Stream.zipWith(
          Stream.empty,
          Stream.make(1, 2, 3),
          (a, b) => [a, b]
        ).pipe(Stream.runCollect)

        const result2 = yield* Stream.zipWith(
          Stream.make(1, 2, 3),
          Stream.empty,
          (a, b) => [a, b]
        ).pipe(Stream.runCollect)

        assert.strictEqual(result1.length, 0)
        assert.strictEqual(result2.length, 0)
      }))

    it.effect.prop(
      "zipWith - equivalence with array operations",
      {
        left: fc.array(fc.integer()),
        right: fc.array(fc.integer())
      },
      Effect.fnUntraced(function*({ left, right }) {
        const stream = Stream.zipWith(
          Stream.fromIterable(left),
          Stream.fromIterable(right),
          (a, b) => a + b
        )
        const result = yield* Stream.runCollect(stream)

        const minLength = Math.min(left.length, right.length)
        const expected = Array.empty<number>()
        for (let i = 0; i < minLength; i++) {
          expected.push(left[i] + right[i])
        }

        assert.deepStrictEqual(result, expected)
      })
    )

    describe("zipWithArray", () => {
      it.effect("basic zipping with equal length streams", () =>
        Effect.gen(function*() {
          const stream1 = Stream.make(1, 2, 3)
          const stream2 = Stream.make("a", "b", "c")
          const result = yield* Stream.zipWithArray(stream1, stream2, (left, right) => {
            const minLength = Math.min(left.length, right.length)
            const output = Array.makeBy(minLength, (i: number) => [left[i], right[i]] as const)
            return [output, left.slice(minLength), right.slice(minLength)]
          }).pipe(Stream.runCollect)

          assert.deepStrictEqual(result, [[1, "a"], [2, "b"], [3, "c"]])
        }))

      it.effect("left stream shorter than right", () =>
        Effect.gen(function*() {
          const stream1 = Stream.make(1, 2, 3)
          const stream2 = Stream.make("a", "b", "c", "d", "e")
          const result = yield* Stream.zipWithArray(stream1, stream2, (left, right) => {
            const minLength = Math.min(left.length, right.length)
            const output = Array.makeBy(minLength, (i: number) => [left[i], right[i]] as const)
            return [output, left.slice(minLength), right.slice(minLength)]
          }).pipe(Stream.runCollect)

          assert.deepStrictEqual(result, [[1, "a"], [2, "b"], [3, "c"]])
        }))

      it.effect("right stream shorter than left", () =>
        Effect.gen(function*() {
          const stream1 = Stream.make(1, 2, 3, 4, 5)
          const stream2 = Stream.make("a", "b", "c")
          const result = yield* Stream.zipWithArray(stream1, stream2, (left, right) => {
            const minLength = Math.min(left.length, right.length)
            const output = Array.makeBy(minLength, (i: number) => [left[i], right[i]] as const)
            return [output, left.slice(minLength), right.slice(minLength)]
          }).pipe(Stream.runCollect)

          assert.deepStrictEqual(result, [[1, "a"], [2, "b"], [3, "c"]])
        }))

      it.effect("multiple arrays from each stream", () =>
        Effect.gen(function*() {
          const stream1 = Stream.fromArrays([1, 2], [3, 4], [5, 6])
          const stream2 = Stream.fromArrays(["a", "b"], ["c", "d"], ["e", "f"])
          const result = yield* Stream.zipWithArray(stream1, stream2, (left, right) => {
            const minLength = Math.min(left.length, right.length)
            const output = Array.makeBy(minLength, (i: number) => [left[i], right[i]] as const)
            return [output, left.slice(minLength), right.slice(minLength)]
          }).pipe(Stream.runCollect)

          assert.deepStrictEqual(result, [[1, "a"], [2, "b"], [3, "c"], [4, "d"], [5, "e"], [6, "f"]])
        }))

      it.effect("handles leftover elements correctly", () =>
        Effect.gen(function*() {
          const stream1 = Stream.fromArrays([1, 2, 3], [4, 5])
          const stream2 = Stream.fromArrays(["a", "b"], ["c", "d", "e"])
          let leftLeftovers = 0
          let rightLeftovers = 0

          const result = yield* Stream.zipWithArray(stream1, stream2, (left, right) => {
            const minLength = Math.min(left.length, right.length)
            const output = Array.makeBy(minLength, (i: number) => [left[i], right[i]] as const)
            const leftSlice = left.slice(minLength)
            const rightSlice = right.slice(minLength)
            if (leftSlice.length > 0) {
              leftLeftovers++
            } else if (rightSlice.length > 0) {
              rightLeftovers++
            }
            return [output, leftSlice, rightSlice]
          }).pipe(Stream.runCollect)

          assert.deepStrictEqual(result, [[1, "a"], [2, "b"], [3, "c"], [4, "d"], [5, "e"]])
          assert.isTrue(leftLeftovers > 0 || rightLeftovers > 0)
        }))

      it.effect("error propagation from left stream", () =>
        Effect.gen(function*() {
          const result = yield* Stream.zipWithArray(
            Stream.make(1, 2).pipe(Stream.concat(Stream.fail("boom"))),
            Stream.make("a", "b", "c"),
            (left, right) => {
              const minLength = Math.min(left.length, right.length)
              const output = Array.makeBy(minLength, (i: number) => [left[i], right[i]] as const)
              return [output, [], []]
            }
          ).pipe(Stream.runCollect, Effect.exit)

          assertExitFailure(result, Cause.fail("boom"))
        }))

      // Note: This test is skipped because with sequential pulling (matching zipWith behavior),
      // when the left stream ends, we don't pull from the right stream, so errors after
      // the left stream ends are not encountered. This is correct behavior.
      it.skip("error propagation from right stream", () =>
        Effect.gen(function*() {
          const result = yield* Stream.zipWithArray(
            Stream.make(1, 2, 3),
            Stream.make("a", "b").pipe(Stream.concat(Stream.fail("boom"))),
            (left, right) => {
              const minLength = Math.min(left.length, right.length)
              const output = Array.makeBy(minLength, (i: number) => [left[i], right[i]] as const)
              return [output, [], []]
            }
          ).pipe(Stream.runCollect, Effect.exit)

          assertExitFailure(result, Cause.fail("boom"))
        }))

      it.effect("handles empty streams", () =>
        Effect.gen(function*() {
          const result1 = yield* Stream.zipWithArray(
            Stream.empty,
            Stream.make(1, 2, 3),
            (left, right) => {
              return [Array.of([left[0], right[0]] as const), [], []]
            }
          ).pipe(Stream.runCollect)

          const result2 = yield* Stream.zipWithArray(
            Stream.make(1, 2, 3),
            Stream.empty,
            (left, right) => {
              return [Array.of([left[0], right[0]] as const), [], []]
            }
          ).pipe(Stream.runCollect)

          assert.strictEqual(result1.length, 0)
          assert.strictEqual(result2.length, 0)
        }))

      it.effect("processes arrays of different sizes within streams", () =>
        Effect.gen(function*() {
          const stream1 = Stream.fromArrays([1], [2, 3, 4], [5])
          const stream2 = Stream.fromArrays([10, 20], [30], [40, 50])
          const result = yield* Stream.zipWithArray(stream1, stream2, (left, right) => {
            const minLength = Math.min(left.length, right.length)
            const output = Array.makeBy(minLength, (i: number) => [left[i], right[i]] as const)
            return [output, left.slice(minLength), right.slice(minLength)]
          }).pipe(Stream.runCollect)

          assert.deepStrictEqual(result, [[1, 10], [2, 20], [3, 30], [4, 40], [5, 50]])
        }))

      it.effect("custom array-level logic - take pairs", () =>
        Effect.gen(function*() {
          const stream1 = Stream.make(1, 2, 3, 4)
          const stream2 = Stream.make(10, 20, 30, 40)
          const result = yield* Stream.zipWithArray(stream1, stream2, (left, right) => {
            const pairs = Math.min(Math.floor(left.length / 2), Math.floor(right.length / 2))
            const output = Array.makeBy(pairs, (i: number) => {
              return [left[i * 2], left[i * 2 + 1], right[i * 2], right[i * 2 + 1]] as const
            })

            const leftUsed = pairs * 2
            const rightUsed = pairs * 2
            const leftLeftover = left.slice(leftUsed)
            const rightLeftover = right.slice(rightUsed)

            return [output, leftLeftover, rightLeftover]
          }).pipe(Stream.runCollect)

          assert.deepStrictEqual(result, [[1, 2, 10, 20], [3, 4, 30, 40]])
        }))

      it.effect("does not pull too much when one stream ends", () =>
        Effect.gen(function*() {
          const left = Stream.fromArrays([1, 2], [3, 4], [5]).pipe(
            Stream.concat(Stream.fail("boom"))
          )
          const right = Stream.fromArrays(["a", "b"], ["c"])
          const result = yield* Stream.zipWithArray(left, right, (leftArr, rightArr) => {
            const minLength = Math.min(leftArr.length, rightArr.length)
            const output = Array.makeBy(minLength, (i: number) => [leftArr[i], rightArr[i]] as const)
            return [output, leftArr.slice(minLength), rightArr.slice(minLength)]
          }).pipe(Stream.runCollect)

          assert.deepStrictEqual(result, [[1, "a"], [2, "b"], [3, "c"]])
        }))
    })

    describe("zip", () => {
      it.effect("zips two streams into tuples", () =>
        Effect.gen(function*() {
          const stream1 = Stream.make(1, 2, 3)
          const stream2 = Stream.make("a", "b", "c")
          const result = yield* Stream.zip(stream1, stream2).pipe(Stream.runCollect)

          assert.deepStrictEqual(result, [[1, "a"], [2, "b"], [3, "c"]])
        }))

      it.effect("terminates when left stream ends first", () =>
        Effect.gen(function*() {
          const stream1 = Stream.make(1, 2)
          const stream2 = Stream.make("a", "b", "c", "d")
          const result = yield* Stream.zip(stream1, stream2).pipe(Stream.runCollect)

          assert.deepStrictEqual(result, [[1, "a"], [2, "b"]])
        }))

      it.effect("terminates when right stream ends first", () =>
        Effect.gen(function*() {
          const stream1 = Stream.make(1, 2, 3, 4)
          const stream2 = Stream.make("a", "b")
          const result = yield* Stream.zip(stream1, stream2).pipe(Stream.runCollect)

          assert.deepStrictEqual(result, [[1, "a"], [2, "b"]])
        }))

      it.effect("handles empty left stream", () =>
        Effect.gen(function*() {
          const stream1 = Stream.empty
          const stream2 = Stream.make("a", "b", "c")
          const result = yield* Stream.zip(stream1, stream2).pipe(Stream.runCollect)

          assert.strictEqual(result.length, 0)
        }))

      it.effect("handles empty right stream", () =>
        Effect.gen(function*() {
          const stream1 = Stream.make(1, 2, 3)
          const stream2 = Stream.empty
          const result = yield* Stream.zip(stream1, stream2).pipe(Stream.runCollect)

          assert.strictEqual(result.length, 0)
        }))

      it.effect("propagates errors from left stream", () =>
        Effect.gen(function*() {
          const stream1 = Stream.make(1, 2).pipe(Stream.concat(Stream.fail("boom")))
          const stream2 = Stream.make("a", "b", "c")
          const result = yield* Stream.zip(stream1, stream2).pipe(Stream.runCollect, Effect.exit)

          assertExitFailure(result, Cause.fail("boom"))
        }))

      it.effect("propagates errors from right stream", () =>
        Effect.gen(function*() {
          const stream1 = Stream.make(1, 2, 3)
          const stream2 = Stream.make("a", "b").pipe(Stream.concat(Stream.fail("ouch")))
          const result = yield* Stream.zip(stream1, stream2).pipe(Stream.runCollect, Effect.exit)

          assertExitFailure(result, Cause.fail("ouch"))
        }))

      it.effect("works with pipe syntax", () =>
        Effect.gen(function*() {
          const stream1 = Stream.make(1, 2, 3)
          const stream2 = Stream.make("a", "b", "c")
          const result = yield* stream1.pipe(Stream.zip(stream2), Stream.runCollect)

          assert.deepStrictEqual(result, [[1, "a"], [2, "b"], [3, "c"]])
        }))
    })

    describe("zipLeft", () => {
      it.effect("keeps only left values", () =>
        Effect.gen(function*() {
          const stream1 = Stream.make(1, 2, 3)
          const stream2 = Stream.make("a", "b", "c")
          const result = yield* Stream.zipLeft(stream1, stream2).pipe(Stream.runCollect)

          assert.deepStrictEqual(result, [1, 2, 3])
        }))

      it.effect("terminates when left stream ends first", () =>
        Effect.gen(function*() {
          const stream1 = Stream.make(1, 2)
          const stream2 = Stream.make("a", "b", "c", "d")
          const result = yield* Stream.zipLeft(stream1, stream2).pipe(Stream.runCollect)

          assert.deepStrictEqual(result, [1, 2])
        }))

      it.effect("terminates when right stream ends first", () =>
        Effect.gen(function*() {
          const stream1 = Stream.make(1, 2, 3, 4)
          const stream2 = Stream.make("a", "b")
          const result = yield* Stream.zipLeft(stream1, stream2).pipe(Stream.runCollect)

          assert.deepStrictEqual(result, [1, 2])
        }))

      it.effect("handles empty left stream", () =>
        Effect.gen(function*() {
          const stream1 = Stream.empty
          const stream2 = Stream.make("a", "b", "c")
          const result = yield* Stream.zipLeft(stream1, stream2).pipe(Stream.runCollect)

          assert.strictEqual(result.length, 0)
        }))

      it.effect("handles empty right stream", () =>
        Effect.gen(function*() {
          const stream1 = Stream.make(1, 2, 3)
          const stream2 = Stream.empty
          const result = yield* Stream.zipLeft(stream1, stream2).pipe(Stream.runCollect)

          assert.strictEqual(result.length, 0)
        }))

      it.effect("propagates errors from left stream", () =>
        Effect.gen(function*() {
          const stream1 = Stream.make(1, 2).pipe(Stream.concat(Stream.fail("boom")))
          const stream2 = Stream.make("a", "b", "c")
          const result = yield* Stream.zipLeft(stream1, stream2).pipe(Stream.runCollect, Effect.exit)

          assertExitFailure(result, Cause.fail("boom"))
        }))

      it.effect("propagates errors from right stream", () =>
        Effect.gen(function*() {
          const stream1 = Stream.make(1, 2, 3)
          const stream2 = Stream.make("a", "b").pipe(Stream.concat(Stream.fail("ouch")))
          const result = yield* Stream.zipLeft(stream1, stream2).pipe(Stream.runCollect, Effect.exit)

          assertExitFailure(result, Cause.fail("ouch"))
        }))

      it.effect("works with pipe syntax", () =>
        Effect.gen(function*() {
          const stream1 = Stream.make(1, 2, 3)
          const stream2 = Stream.make("a", "b", "c")
          const result = yield* stream1.pipe(Stream.zipLeft(stream2), Stream.runCollect)

          assert.deepStrictEqual(result, [1, 2, 3])
        }))

      it.effect("consumes right stream elements", () =>
        Effect.gen(function*() {
          let rightPulled = 0
          const stream1 = Stream.make(1, 2, 3)
          const stream2 = Stream.make("a", "b", "c").pipe(
            Stream.tap(() =>
              Effect.sync(() => {
                rightPulled++
              })
            )
          )
          const result = yield* Stream.zipLeft(stream1, stream2).pipe(Stream.runCollect)

          assert.deepStrictEqual(result, [1, 2, 3])
          assert.strictEqual(rightPulled, 3)
        }))
    })

    describe("zipRight", () => {
      it.effect("keeps only right values", () =>
        Effect.gen(function*() {
          const stream1 = Stream.make(1, 2, 3)
          const stream2 = Stream.make("a", "b", "c")
          const result = yield* Stream.zipRight(stream1, stream2).pipe(Stream.runCollect)

          assert.deepStrictEqual(result, ["a", "b", "c"])
        }))

      it.effect("terminates when left stream ends first", () =>
        Effect.gen(function*() {
          const stream1 = Stream.make(1, 2)
          const stream2 = Stream.make("a", "b", "c", "d")
          const result = yield* Stream.zipRight(stream1, stream2).pipe(Stream.runCollect)

          assert.deepStrictEqual(result, ["a", "b"])
        }))

      it.effect("terminates when right stream ends first", () =>
        Effect.gen(function*() {
          const stream1 = Stream.make(1, 2, 3, 4)
          const stream2 = Stream.make("a", "b")
          const result = yield* Stream.zipRight(stream1, stream2).pipe(Stream.runCollect)

          assert.deepStrictEqual(result, ["a", "b"])
        }))

      it.effect("handles empty left stream", () =>
        Effect.gen(function*() {
          const stream1 = Stream.empty
          const stream2 = Stream.make("a", "b", "c")
          const result = yield* Stream.zipRight(stream1, stream2).pipe(Stream.runCollect)

          assert.strictEqual(result.length, 0)
        }))

      it.effect("handles empty right stream", () =>
        Effect.gen(function*() {
          const stream1 = Stream.make(1, 2, 3)
          const stream2 = Stream.empty
          const result = yield* Stream.zipRight(stream1, stream2).pipe(Stream.runCollect)

          assert.strictEqual(result.length, 0)
        }))

      it.effect("propagates errors from left stream", () =>
        Effect.gen(function*() {
          const stream1 = Stream.make(1, 2).pipe(Stream.concat(Stream.fail("boom")))
          const stream2 = Stream.make("a", "b", "c")
          const result = yield* Stream.zipRight(stream1, stream2).pipe(Stream.runCollect, Effect.exit)

          assertExitFailure(result, Cause.fail("boom"))
        }))

      it.effect("propagates errors from right stream", () =>
        Effect.gen(function*() {
          const stream1 = Stream.make(1, 2, 3)
          const stream2 = Stream.make("a", "b").pipe(Stream.concat(Stream.fail("ouch")))
          const result = yield* Stream.zipRight(stream1, stream2).pipe(Stream.runCollect, Effect.exit)

          assertExitFailure(result, Cause.fail("ouch"))
        }))

      it.effect("works with pipe syntax", () =>
        Effect.gen(function*() {
          const stream1 = Stream.make(1, 2, 3)
          const stream2 = Stream.make("a", "b", "c")
          const result = yield* stream1.pipe(Stream.zipRight(stream2), Stream.runCollect)

          assert.deepStrictEqual(result, ["a", "b", "c"])
        }))

      it.effect("consumes left stream elements", () =>
        Effect.gen(function*() {
          let leftPulled = 0
          const stream1 = Stream.make(1, 2, 3).pipe(
            Stream.tap(() =>
              Effect.sync(() => {
                leftPulled++
              })
            )
          )
          const stream2 = Stream.make("a", "b", "c")
          const result = yield* Stream.zipRight(stream1, stream2).pipe(Stream.runCollect)

          assert.deepStrictEqual(result, ["a", "b", "c"])
          assert.strictEqual(leftPulled, 3)
        }))
    })

    describe("zipFlatten", () => {
      it.effect("flattens tuples when zipping", () =>
        Effect.gen(function*() {
          const stream1 = Stream.make([1, "a"] as const, [2, "b"] as const, [3, "c"] as const)
          const stream2 = Stream.make("x", "y", "z")
          const result = yield* Stream.zipFlatten(stream1, stream2).pipe(Stream.runCollect)

          assert.deepStrictEqual(result, [[1, "a", "x"], [2, "b", "y"], [3, "c", "z"]])
        }))

      it.effect("works with single element tuples", () =>
        Effect.gen(function*() {
          const stream1 = Stream.make([1] as const, [2] as const, [3] as const)
          const stream2 = Stream.make("a", "b", "c")
          const result = yield* Stream.zipFlatten(stream1, stream2).pipe(Stream.runCollect)

          assert.deepStrictEqual(result, [[1, "a"], [2, "b"], [3, "c"]])
        }))

      it.effect("works with larger tuples", () =>
        Effect.gen(function*() {
          const stream1 = Stream.make([1, 2, 3] as const, [4, 5, 6] as const)
          const stream2 = Stream.make("a", "b")
          const result = yield* Stream.zipFlatten(stream1, stream2).pipe(Stream.runCollect)

          assert.deepStrictEqual(result, [[1, 2, 3, "a"], [4, 5, 6, "b"]])
        }))

      it.effect("terminates when left stream ends first", () =>
        Effect.gen(function*() {
          const stream1 = Stream.make([1, "a"] as const, [2, "b"] as const)
          const stream2 = Stream.make("x", "y", "z", "w")
          const result = yield* Stream.zipFlatten(stream1, stream2).pipe(Stream.runCollect)

          assert.deepStrictEqual(result, [[1, "a", "x"], [2, "b", "y"]])
        }))

      it.effect("terminates when right stream ends first", () =>
        Effect.gen(function*() {
          const stream1 = Stream.make([1, "a"] as const, [2, "b"] as const, [3, "c"] as const, [4, "d"] as const)
          const stream2 = Stream.make("x", "y")
          const result = yield* Stream.zipFlatten(stream1, stream2).pipe(Stream.runCollect)

          assert.deepStrictEqual(result, [[1, "a", "x"], [2, "b", "y"]])
        }))

      it.effect("handles empty left stream", () =>
        Effect.gen(function*() {
          const stream1 = Stream.empty
          const stream2 = Stream.make("a", "b", "c")
          const result = yield* Stream.zipFlatten(stream1, stream2).pipe(Stream.runCollect)

          assert.strictEqual(result.length, 0)
        }))

      it.effect("handles empty right stream", () =>
        Effect.gen(function*() {
          const stream1 = Stream.make([1, "a"] as const, [2, "b"] as const)
          const stream2 = Stream.empty
          const result = yield* Stream.zipFlatten(stream1, stream2).pipe(Stream.runCollect)

          assert.strictEqual(result.length, 0)
        }))

      it.effect("propagates errors from left stream", () =>
        Effect.gen(function*() {
          const stream1 = Stream.make([1, "a"] as const, [2, "b"] as const).pipe(Stream.concat(Stream.fail("boom")))
          const stream2 = Stream.make("x", "y", "z")
          const result = yield* Stream.zipFlatten(stream1, stream2).pipe(Stream.runCollect, Effect.exit)

          assertExitFailure(result, Cause.fail("boom"))
        }))

      it.effect("propagates errors from right stream", () =>
        Effect.gen(function*() {
          const stream1 = Stream.make([1, "a"] as const, [2, "b"] as const, [3, "c"] as const)
          const stream2 = Stream.make("x", "y").pipe(Stream.concat(Stream.fail("ouch")))
          const result = yield* Stream.zipFlatten(stream1, stream2).pipe(Stream.runCollect, Effect.exit)

          assertExitFailure(result, Cause.fail("ouch"))
        }))

      it.effect("works with pipe syntax", () =>
        Effect.gen(function*() {
          const stream1 = Stream.make([1, "a"] as const, [2, "b"] as const, [3, "c"] as const)
          const stream2 = Stream.make("x", "y", "z")
          const result = yield* stream1.pipe(Stream.zipFlatten(stream2), Stream.runCollect)

          assert.deepStrictEqual(result, [[1, "a", "x"], [2, "b", "y"], [3, "c", "z"]])
        }))
    })

    describe("zipWithIndex", () => {
      it.effect("zips stream with indices starting at 0", () =>
        Effect.gen(function*() {
          const stream = Stream.make("a", "b", "c", "d")
          const result = yield* Stream.zipWithIndex(stream).pipe(Stream.runCollect)

          assert.deepStrictEqual(result, [["a", 0], ["b", 1], ["c", 2], ["d", 3]])
        }))

      it.effect("handles empty stream", () =>
        Effect.gen(function*() {
          const stream = Stream.empty
          const result = yield* Stream.zipWithIndex(stream).pipe(Stream.runCollect)

          assert.strictEqual(result.length, 0)
        }))

      it.effect("handles single element stream", () =>
        Effect.gen(function*() {
          const stream = Stream.make("a")
          const result = yield* Stream.zipWithIndex(stream).pipe(Stream.runCollect)

          assert.deepStrictEqual(result, [["a", 0]])
        }))

      it.effect("preserves order", () =>
        Effect.gen(function*() {
          const stream = Stream.make(5, 4, 3, 2, 1)
          const result = yield* Stream.zipWithIndex(stream).pipe(Stream.runCollect)

          assert.deepStrictEqual(result, [[5, 0], [4, 1], [3, 2], [2, 3], [1, 4]])
        }))

      it.effect("works with numbers", () =>
        Effect.gen(function*() {
          const stream = Stream.make(10, 20, 30)
          const result = yield* Stream.zipWithIndex(stream).pipe(Stream.runCollect)

          assert.deepStrictEqual(result, [[10, 0], [20, 1], [30, 2]])
        }))

      it.effect("propagates errors", () =>
        Effect.gen(function*() {
          const stream = Stream.make(1, 2, 3).pipe(Stream.concat(Stream.fail("boom")))
          const result = yield* Stream.zipWithIndex(stream).pipe(Stream.runCollect, Effect.exit)

          assertExitFailure(result, Cause.fail("boom"))
        }))

      it.effect("works with pipe syntax", () =>
        Effect.gen(function*() {
          const stream = Stream.make("x", "y", "z")
          const result = yield* stream.pipe(Stream.zipWithIndex, Stream.runCollect)

          assert.deepStrictEqual(result, [["x", 0], ["y", 1], ["z", 2]])
        }))

      it.effect("index increments correctly for large streams", () =>
        Effect.gen(function*() {
          const stream = Stream.range(0, 99)
          const result = yield* Stream.zipWithIndex(stream).pipe(Stream.runCollect)

          assert.strictEqual(result.length, 100)
          assert.deepStrictEqual(result[0], [0, 0])
          assert.deepStrictEqual(result[99], [99, 99])
        }))
    })

    describe("zipLatest", () => {
      it.effect("combines streams with latest values", () =>
        Effect.gen(function*() {
          const result = yield* Stream.zipLatest(
            Stream.make(1, 2, 3),
            Stream.make("a", "b", "c", "d")
          ).pipe(Stream.runCollect)

          // Both streams emit, latest values are paired
          assert.isTrue(result.length > 0)
          assert.isTrue(result.every((item) => Array.isArray(item) && item.length === 2))
        }))

      it.effect("waits for both streams to emit before producing output", () =>
        Effect.gen(function*() {
          const queue1 = yield* Queue.unbounded<number>()
          const queue2 = yield* Queue.unbounded<string>()

          const fiber = yield* Effect.forkScoped(
            Stream.zipLatest(
              Stream.fromQueue(queue1),
              Stream.fromQueue(queue2)
            ).pipe(Stream.take(3), Stream.runCollect)
          )

          yield* Queue.offer(queue1, 1)
          yield* Queue.offer(queue2, "a")
          yield* Queue.offer(queue1, 2)
          yield* Queue.offer(queue2, "b")
          yield* Queue.offer(queue1, 3)

          const result = yield* Fiber.join(fiber)

          assert.isTrue(result.length > 0)
        }))

      it.effect("propagates errors from left stream", () =>
        Effect.gen(function*() {
          const result = yield* Stream.zipLatest(
            Stream.make(1, 2).pipe(Stream.concat(Stream.fail("boom"))),
            Stream.make("a", "b", "c")
          ).pipe(Stream.runCollect, Effect.exit)

          assertExitFailure(result, Cause.fail("boom"))
        }))

      it.effect("propagates errors from right stream", () =>
        Effect.gen(function*() {
          const result = yield* Stream.zipLatest(
            Stream.make(1, 2, 3),
            Stream.make("a", "b").pipe(Stream.concat(Stream.fail("ouch")))
          ).pipe(Stream.runCollect, Effect.exit)

          assertExitFailure(result, Cause.fail("ouch"))
        }))

      it.effect("terminates when left stream ends", () =>
        Effect.gen(function*() {
          const result = yield* Stream.zipLatest(
            Stream.make(1, 2),
            Stream.make("a", "b", "c", "d", "e")
          ).pipe(Stream.runCollect)

          // Should terminate when left stream (1, 2) ends
          assert.isTrue(result.length > 0)
        }))

      it.effect("terminates when right stream ends", () =>
        Effect.gen(function*() {
          const result = yield* Stream.zipLatest(
            Stream.make(1, 2, 3, 4, 5),
            Stream.make("a", "b")
          ).pipe(Stream.runCollect)

          // Should terminate when right stream ("a", "b") ends
          assert.isTrue(result.length > 0)
        }))

      it.effect("works with pipe syntax", () =>
        Effect.gen(function*() {
          const result = yield* Stream.make(1, 2, 3).pipe(
            Stream.zipLatest(Stream.make("a", "b", "c")),
            Stream.runCollect
          )

          assert.isTrue(result.length > 0)
        }))

      it.effect("basic synchronous example", () =>
        Effect.gen(function*() {
          const result = yield* Stream.zipLatest(
            Stream.make(1, 2, 3),
            Stream.make("a", "b", "c")
          ).pipe(Stream.runCollect)

          // Both streams emit synchronously, latest values are paired
          assert.isTrue(result.length > 0)
          assert.isTrue(result.every((item) => Array.isArray(item) && item.length === 2))
        }))
    })

    describe("zipLatestWith", () => {
      it.effect("transforms combined values", () =>
        Effect.gen(function*() {
          const result = yield* Stream.zipLatestWith(
            Stream.make(1, 2, 3),
            Stream.make(10, 20, 30),
            (n, m) => n * m
          ).pipe(Stream.runCollect)

          assert.isTrue(result.length > 0)
          assert.isTrue(result.every((item) => typeof item === "number"))
        }))

      it.effect("combines with custom function", () =>
        Effect.gen(function*() {
          const result = yield* Stream.zipLatestWith(
            Stream.make(1, 2, 3),
            Stream.make("a", "b", "c"),
            (n, s) => `${n}-${s}`
          ).pipe(Stream.runCollect)

          assert.isTrue(result.length > 0)
          assert.isTrue(result.every((item) => typeof item === "string" && item.includes("-")))
        }))

      it.effect("propagates errors from left stream", () =>
        Effect.gen(function*() {
          const result = yield* Stream.zipLatestWith(
            Stream.make(1, 2).pipe(Stream.concat(Stream.fail("boom"))),
            Stream.make("a", "b", "c"),
            (n, s) => `${n}-${s}`
          ).pipe(Stream.runCollect, Effect.exit)

          assertExitFailure(result, Cause.fail("boom"))
        }))

      it.effect("propagates errors from right stream", () =>
        Effect.gen(function*() {
          const result = yield* Stream.zipLatestWith(
            Stream.make(1, 2, 3),
            Stream.make("a", "b").pipe(Stream.concat(Stream.fail("ouch"))),
            (n, s) => `${n}-${s}`
          ).pipe(Stream.runCollect, Effect.exit)

          assertExitFailure(result, Cause.fail("ouch"))
        }))

      it.effect("terminates when left stream ends", () =>
        Effect.gen(function*() {
          const result = yield* Stream.zipLatestWith(
            Stream.make(1, 2),
            Stream.make(10, 20, 30, 40),
            (n, m) => n + m
          ).pipe(Stream.runCollect)

          assert.isTrue(result.length > 0)
        }))

      it.effect("terminates when right stream ends", () =>
        Effect.gen(function*() {
          const result = yield* Stream.zipLatestWith(
            Stream.make(1, 2, 3, 4),
            Stream.make(10, 20),
            (n, m) => n + m
          ).pipe(Stream.runCollect)

          assert.isTrue(result.length > 0)
        }))

      it.effect("works with pipe syntax", () =>
        Effect.gen(function*() {
          const result = yield* Stream.make(1, 2, 3).pipe(
            Stream.zipLatestWith(
              Stream.make(10, 20, 30),
              (n, m) => n + m
            ),
            Stream.runCollect
          )

          assert.isTrue(result.length > 0)
        }))

      it.effect("basic synchronous example", () =>
        Effect.gen(function*() {
          const result = yield* Stream.zipLatestWith(
            Stream.make(1, 2, 3),
            Stream.make(10, 20, 30),
            (a, b) => a + b
          ).pipe(Stream.runCollect)

          assert.isTrue(result.length > 0)
          assert.isTrue(result.every((item) => typeof item === "number"))
        }))

      it.effect("string concatenation example", () =>
        Effect.gen(function*() {
          const result = yield* Stream.zipLatestWith(
            Stream.make("Alice", "Bob", "Charlie"),
            Stream.make("Smith", "Jones"),
            (first, last) => `${first} ${last}`
          ).pipe(Stream.runCollect)

          assert.isTrue(result.length > 0)
          assert.isTrue(result.every((item) => typeof item === "string" && item.includes(" ")))
        }))
    })
  })

  describe("tapSink", () => {
    it.effect("sink that is done after stream", () =>
      Effect.gen(function*() {
        const ref = yield* Ref.make(0)
        const sink = Sink.forEach((i: number) => Ref.update(ref, (n) => i + n))
        const result = yield* pipe(
          Stream.make(1, 1, 2, 3, 5, 8),
          Stream.tapSink(sink),
          Stream.runCollect
        )
        const sum = yield* (Ref.get(ref))
        strictEqual(sum, 20)
        deepStrictEqual(result, [1, 1, 2, 3, 5, 8])
      }))

    it.effect("sink that is done before stream", () =>
      Effect.gen(function*() {
        const ref = yield* Ref.make(0)
        const sink = pipe(
          Sink.take<number>(3),
          Sink.map(Array.reduce(0, (x, y) => x + y)),
          Sink.mapEffect((i) => Ref.update(ref, (n) => n + i))
        )
        const result = yield* pipe(
          Stream.make(1, 1, 2, 3, 5, 8),
          Stream.rechunk(1),
          Stream.tapSink(sink),
          Stream.runCollect
        )
        const sum = yield* (Ref.get(ref))
        strictEqual(sum, 4)
        deepStrictEqual(result, [1, 1, 2, 3, 5, 8])
      }))

    it.effect("sink that fails before stream", () =>
      Effect.gen(function*() {
        const sink = Sink.fail("error")
        const result = yield* pipe(
          Stream.make(1, 2, 3),
          Stream.tapSink(sink),
          Stream.runCollect,
          Effect.flip
        )
        strictEqual(result, "error")
      }))

    it.effect("does not read ahead", () =>
      Effect.gen(function*() {
        const ref = yield* Ref.make(0)
        const sink = Sink.forEach((i: number) => Ref.update(ref, (n) => i + n))
        yield* pipe(
          Stream.make(1, 2, 3, 4, 5),
          Stream.rechunk(1),
          Stream.forever,
          Stream.tapSink(sink),
          Stream.take(3),
          Stream.runDrain
        )
        const result = yield* Ref.get(ref)
        strictEqual(result, 6)
      }))
  })

  describe("dropping", () => {
    it.effect("drop - simple example", () =>
      Effect.gen(function*() {
        const n = 2
        const stream = Stream.make(1, 2, 3, 4, 5)
        const { result1, result2 } = yield* (Effect.all({
          result1: pipe(stream, Stream.drop(n), Stream.runCollect),
          result2: pipe(stream, Stream.runCollect, Effect.map(Array.drop(n)))
        }))
        deepStrictEqual(result1, result2)
      }))

    it.effect("drop - does not swallow errors", () =>
      Effect.gen(function*() {
        const result = yield* pipe(
          Stream.fail("Ouch"),
          Stream.concat(Stream.make(1)),
          Stream.drop(1),
          Stream.runDrain,
          Effect.result
        )
        assertFailure(result, "Ouch")
      }))

    it.effect("dropRight - simple example", () =>
      Effect.gen(function*() {
        const n = 2
        const stream = Stream.make(1, 2, 3, 4, 5)
        const { result1, result2 } = yield* (Effect.all({
          result1: pipe(stream, Stream.dropRight(n), Stream.runCollect),
          result2: pipe(stream, Stream.runCollect, Effect.map(Array.dropRight(n)))
        }))
        deepStrictEqual(result1, result2)
      }))

    it.effect("dropRight - does not swallow errors", () =>
      Effect.gen(function*() {
        const result = yield* pipe(
          Stream.make(1),
          Stream.concat(Stream.fail("Ouch")),
          Stream.dropRight(1),
          Stream.runDrain,
          Effect.result
        )
        assertFailure(result, "Ouch")
      }))

    it.effect("dropUntil", () =>
      Effect.gen(function*() {
        const stream = Stream.make(1, 2, 3, 4, 5)
        const f = (n: number) => n < 3
        const { result1, result2 } = yield* (Effect.all({
          result1: pipe(stream, Stream.dropUntil(f), Stream.runCollect),
          result2: pipe(
            Stream.runCollect(stream),
            Effect.map((chunk) => pipe(chunk, Array.dropWhile((n) => !f(n)), Array.drop(1)))
          )
        }))
        deepStrictEqual(result1, result2)
      }))

    it.effect("dropWhile", () =>
      Effect.gen(function*() {
        const stream = Stream.make(1, 2, 3, 4, 5)
        const f = (n: number) => n < 3
        const { result1, result2 } = yield* (Effect.all({
          result1: pipe(stream, Stream.dropWhile(f), Stream.runCollect),
          result2: pipe(stream, Stream.runCollect, Effect.map(Array.dropWhile(f)))
        }))
        deepStrictEqual(result1, result2)
      }))

    it.effect("dropWhileFilter", () =>
      Effect.gen(function*() {
        const result = yield* Stream.make(1, 2, 3, 4, 5).pipe(
          Stream.dropWhileFilter((n) => n < 3 ? Result.succeed(n) : Result.failVoid),
          Stream.runCollect
        )
        deepStrictEqual(result, [3, 4, 5])
      }))

    it.effect("dropWhile - short circuits", () =>
      Effect.gen(function*() {
        const result = yield* pipe(
          Stream.make(1),
          Stream.concat(Stream.fail("Ouch")),
          Stream.take(1),
          Stream.dropWhile(constTrue),
          Stream.runDrain,
          Effect.result
        )
        assertSuccess(result, void 0)
      }))
  })

  describe("haltWhen", () => {
    it.effect("halts after the current element", () =>
      Effect.gen(function*() {
        const ref = yield* Ref.make(false)
        const latch = yield* Deferred.make<void>()
        const halt = yield* Deferred.make<void>()
        yield* pipe(
          Deferred.await(latch),
          Effect.onInterrupt(() => Ref.set(ref, true)),
          Stream.fromEffect,
          Stream.haltWhen(Deferred.await(halt)),
          Stream.runDrain,
          Effect.forkChild
        )
        yield* Deferred.succeed(halt, void 0)
        yield* Deferred.succeed(latch, void 0)
        const result = yield* Ref.get(ref)
        assert.strictEqual(result, false)
      }))

    it.effect("propagates errors", () =>
      Effect.gen(function*() {
        const halt = yield* Deferred.make<void, string>()
        yield* Deferred.fail(halt, "fail")
        const result = yield* pipe(
          Stream.make(0),
          Stream.forever,
          Stream.haltWhen(Deferred.await(halt)),
          Stream.runDrain,
          Effect.flip
        )
        assert.strictEqual(result, "fail")
      }))
  })

  describe("intersperse", () => {
    it.effect("intersperse - several values", () =>
      Effect.gen(function*() {
        const result = yield* pipe(
          Stream.make(1, 2, 3, 4),
          Stream.map(String),
          Stream.intersperse("."),
          Stream.runCollect
        )
        deepStrictEqual(result, ["1", ".", "2", ".", "3", ".", "4"])
      }))

    it.effect("intersperseAffixes - several values", () =>
      Effect.gen(function*() {
        const result = yield* pipe(
          Stream.make(1, 2, 3, 4),
          Stream.map(String),
          Stream.intersperseAffixes({ start: "[", middle: ".", end: "]" }),
          Stream.runCollect
        )
        deepStrictEqual(result, ["[", "1", ".", "2", ".", "3", ".", "4", "]"])
      }))

    it.effect("intersperse - single value", () =>
      Effect.gen(function*() {
        const result = yield* pipe(
          Stream.make(1),
          Stream.map(String),
          Stream.intersperse("."),
          Stream.runCollect
        )
        deepStrictEqual(result, ["1"])
      }))

    it.effect("intersperseAffixes - single value", () =>
      Effect.gen(function*() {
        const result = yield* pipe(
          Stream.make(1),
          Stream.map(String),
          Stream.intersperseAffixes({ start: "[", middle: ".", end: "]" }),
          Stream.runCollect
        )
        deepStrictEqual(result, ["[", "1", "]"])
      }))

    it.effect("intersperse - several from repeat effect (ZIO #3729)", () =>
      Effect.gen(function*() {
        const result = yield* pipe(
          Stream.fromEffectRepeat(Effect.succeed(42)),
          Stream.map(String),
          Stream.take(4),
          Stream.intersperse("."),
          Stream.runCollect
        )
        deepStrictEqual(result, ["42", ".", "42", ".", "42", ".", "42"])
      }))

    it.effect("intersperse - several from repeat effect chunk single element (ZIO #3729)", () =>
      Effect.gen(function*() {
        const result = yield* pipe(
          Stream.fromIterableEffectRepeat(Effect.succeed(Array.of(42))),
          Stream.map(String),
          Stream.intersperse("."),
          Stream.take(4),
          Stream.runCollect
        )
        deepStrictEqual(result, ["42", ".", "42", "."])
      }))
  })

  describe("interleave", () => {
    it.effect("interleave", () =>
      Effect.gen(function*() {
        const stream1 = Stream.make(2, 3)
        const stream2 = Stream.make(5, 6, 7)
        const result = yield* pipe(
          stream1,
          Stream.interleave(stream2),
          Stream.runCollect
        )
        deepStrictEqual(result, [2, 5, 3, 6, 7])
      }))

    it.effect("interleaveWith", () =>
      Effect.gen(function*() {
        const interleave = (
          bools: ReadonlyArray<boolean>,
          numbers1: ReadonlyArray<number>,
          numbers2: ReadonlyArray<number>
        ): ReadonlyArray<number> =>
          pipe(
            Array.head(bools),
            Option.map((head) => {
              if (head) {
                if (Array.isReadonlyArrayNonEmpty(numbers1)) {
                  const head = pipe(numbers1, Array.getUnsafe(0))
                  const tail = pipe(numbers1, Array.drop(1))
                  return pipe(
                    interleave(pipe(bools, Array.drop(1)), tail, numbers2),
                    Array.prepend(head)
                  )
                }
                if (Array.isReadonlyArrayNonEmpty(numbers2)) {
                  return interleave(pipe(bools, Array.drop(1)), Array.empty<number>(), numbers2)
                }
                return Array.empty<number>()
              }
              if (Array.isReadonlyArrayNonEmpty(numbers2)) {
                const head = pipe(numbers2, Array.getUnsafe(0))
                const tail = pipe(numbers2, Array.drop(1))
                return pipe(
                  interleave(pipe(bools, Array.drop(1)), numbers1, tail),
                  Array.prepend(head)
                )
              }
              if (Array.isReadonlyArrayNonEmpty(numbers1)) {
                return interleave(pipe(bools, Array.drop(1)), numbers1, Array.empty<number>())
              }
              return Array.empty<number>()
            }),
            Option.getOrElse(() => Array.empty<number>())
          )
        const boolStream = Stream.make(true, true, false, true, false)
        const stream1 = Stream.make(1, 2, 3, 4, 5)
        const stream2 = Stream.make(4, 5, 6, 7, 8)
        const interleavedStream = yield* pipe(
          stream1,
          Stream.interleaveWith(stream2, boolStream),
          Stream.runCollect
        )
        const bools = yield* (Stream.runCollect(boolStream))
        const numbers1 = yield* (Stream.runCollect(stream1))
        const numbers2 = yield* (Stream.runCollect(stream2))
        const interleavedChunks = interleave(bools, numbers1, numbers2)
        deepStrictEqual(interleavedStream, interleavedChunks)
      }))
  })

  describe("merge", () => {
    it.effect("data-first with options", () =>
      Effect.gen(function*() {
        const result = yield* Stream.runCollect(
          Stream.merge(Stream.make(1), Stream.never, { haltStrategy: "left" })
        )
        deepStrictEqual(result, [1])
      }))

    it.effect("data-last with options", () =>
      Effect.gen(function*() {
        const result = yield* pipe(
          Stream.make(1),
          Stream.merge(Stream.never, { haltStrategy: "left" }),
          Stream.runCollect
        )
        deepStrictEqual(result, [1])
      }))
  })

  describe("partition", () => {
    it.effect("values", () =>
      Effect.gen(function*() {
        const { result1, result2 } = yield* pipe(
          Stream.range(0, 5),
          Stream.partition((n) => n % 2 === 0 ? Result.succeed(n) : Result.fail(n)),
          Effect.flatMap(([odds, evens]) =>
            Effect.all({
              result1: Stream.runCollect(evens),
              result2: Stream.runCollect(odds)
            })
          ),
          Effect.scoped
        )
        deepStrictEqual(result1, [0, 2, 4])
        deepStrictEqual(result2, [1, 3, 5])
      }))

    it.effect("errors", () =>
      Effect.gen(function*() {
        const { result1, result2 } = yield* pipe(
          Stream.make(0),
          Stream.concat(Stream.fail("boom")),
          Stream.partition((n) => n % 2 === 0 ? Result.succeed(n) : Result.fail(n)),
          Effect.flatMap(([odds, evens]) =>
            Effect.all({
              result1: Effect.flip(Stream.runCollect(evens)),
              result2: Effect.flip(Stream.runCollect(odds))
            })
          ),
          Effect.scoped
        )
        assert.strictEqual(result1, "boom")
        assert.strictEqual(result2, "boom")
      }))

    it.effect("backpressure", () =>
      Effect.gen(function*() {
        const { result1, result2, result3 } = yield* pipe(
          Stream.range(0, 5),
          Stream.partition((n) => n % 2 === 0 ? Result.succeed(n) : Result.fail(n), { bufferSize: 1 }),
          Effect.flatMap(([odds, evens]) =>
            Effect.gen(function*() {
              const ref = yield* Ref.make(Array.empty<number>())
              const latch = yield* Deferred.make<void>()
              const fiber = yield* pipe(
                evens,
                Stream.tap((n) =>
                  pipe(
                    Ref.update(ref, Array.prepend(n)),
                    Effect.andThen(
                      pipe(
                        Deferred.succeed(latch, void 0),
                        Effect.when(Effect.sync(() => n === 2))
                      )
                    )
                  )
                ),
                Stream.runDrain,
                Effect.forkChild
              )
              yield* Deferred.await(latch)
              const result1 = yield* Ref.get(ref)
              const result2 = yield* Stream.runCollect(odds)
              yield* Fiber.await(fiber)
              const result3 = yield* Ref.get(ref)
              return { result1, result2, result3 }
            })
          ),
          Effect.scoped
        )
        deepStrictEqual(result1, [2, 0])
        deepStrictEqual(result2, [1, 3, 5])
        deepStrictEqual(result3, [4, 2, 0])
      }))
  })

  describe("partition", () => {
    it.effect("partitionEffect - allows repeated runs without hanging", () =>
      Effect.gen(function*() {
        const stream = pipe(
          Stream.fromIterable(Array.empty<number>()),
          Stream.partitionEffect((n) => Effect.succeed(n % 2 === 0 ? Result.succeed(n) : Result.fail(n))),
          Effect.map(([odds, evens]) => pipe(evens, Stream.mergeResult(odds))),
          Effect.flatMap(Stream.runCollect),
          Effect.scoped
        )
        const result = yield* pipe(
          Effect.all(Array.makeBy(100, () => stream)),
          Effect.as(0)
        )
        strictEqual(result, 0)
      }))

    it.effect("partitionEffect - values", () =>
      Effect.gen(function*() {
        const { result1, result2 } = yield* pipe(
          Stream.range(0, 5),
          Stream.partitionEffect((n) => Effect.succeed(n % 2 === 0 ? Result.succeed(n + 10) : Result.fail(`odd:${n}`))),
          Effect.flatMap(([passes, fails]) =>
            Effect.all({
              result1: Stream.runCollect(passes),
              result2: Stream.runCollect(fails)
            })
          ),
          Effect.scoped
        )
        deepStrictEqual(result1, [10, 12, 14])
        deepStrictEqual(result2, ["odd:1", "odd:3", "odd:5"])
      }))

    it.effect("partitionQueue - values", () =>
      Effect.gen(function*() {
        const { result1, result2 } = yield* pipe(
          Stream.range(0, 5),
          Stream.partitionQueue((n) => n % 2 === 0 ? Result.succeed(n + 1) : Result.fail(`odd:${n}`)),
          Effect.flatMap(([passes, fails]) =>
            Effect.all({
              result1: Stream.runCollect(Stream.fromQueue(passes)),
              result2: Stream.runCollect(Stream.fromQueue(fails))
            })
          ),
          Effect.scoped
        )
        deepStrictEqual(result1, [1, 3, 5])
        deepStrictEqual(result2, ["odd:1", "odd:3", "odd:5"])
      }))

    it.effect("partition - values", () =>
      Effect.gen(function*() {
        const { result1, result2 } = yield* pipe(
          Stream.range(0, 5),
          Stream.partition((n) => n % 2 === 0 ? Result.succeed(n) : Result.fail(n)),
          Effect.flatMap(([odds, evens]) =>
            Effect.all({
              result1: Stream.runCollect(evens),
              result2: Stream.runCollect(odds)
            })
          ),
          Effect.scoped
        )
        deepStrictEqual(result1, [0, 2, 4])
        deepStrictEqual(result2, [1, 3, 5])
      }))

    it.effect("partition - errors", () =>
      Effect.gen(function*() {
        const { result1, result2 } = yield* pipe(
          Stream.make(0),
          Stream.concat(Stream.fail("boom")),
          Stream.partition((n) => n % 2 === 0 ? Result.succeed(n) : Result.fail(n)),
          Effect.flatMap(([odds, evens]) =>
            Effect.all({
              result1: Effect.flip(Stream.runCollect(evens)),
              result2: Effect.flip(Stream.runCollect(odds))
            })
          ),
          Effect.scoped
        )
        assert.strictEqual(result1, "boom")
        assert.strictEqual(result2, "boom")
      }))

    it.effect("partition - backpressure", () =>
      Effect.gen(function*() {
        const { result1, result2, result3 } = yield* pipe(
          Stream.range(0, 5),
          Stream.partition((n) => (n % 2 === 0 ? Result.succeed(n) : Result.fail(n)), { bufferSize: 1 }),
          Effect.flatMap(([odds, evens]) =>
            Effect.gen(function*() {
              const ref = yield* Ref.make(Array.empty<number>())
              const latch = yield* (Deferred.make<void>())
              const fiber = yield* pipe(
                evens,
                Stream.tap((n) =>
                  pipe(
                    Ref.update(ref, Array.prepend(n)),
                    Effect.andThen(
                      pipe(
                        Deferred.succeed(latch, void 0),
                        Effect.when(Effect.sync(() => n === 2))
                      )
                    )
                  )
                ),
                Stream.runDrain,
                Effect.forkChild
              )
              yield* (Deferred.await(latch))
              const result1 = yield* (Ref.get(ref))
              const result2 = yield* (Stream.runCollect(odds))
              yield* (Fiber.await(fiber))
              const result3 = yield* (Ref.get(ref))
              return { result1, result2, result3 }
            })
          ),
          Effect.scoped
        )
        deepStrictEqual(result1, [2, 0])
        deepStrictEqual(result2, [1, 3, 5])
        deepStrictEqual(result3, [4, 2, 0])
      }))

    it.effect("partition with Filter", () =>
      Effect.gen(function*() {
        const { evens, odds } = yield* pipe(
          Stream.range(0, 5),
          Stream.partition((n: number) => n % 2 === 0 ? Result.succeed(n) : Result.fail(n)),
          Effect.flatMap(([fail, pass]) =>
            Effect.all({
              evens: Stream.runCollect(pass),
              odds: Stream.runCollect(fail)
            })
          ),
          Effect.scoped
        )
        deepStrictEqual(evens, [0, 2, 4])
        deepStrictEqual(odds, [1, 3, 5])
      }))
  })

  describe("peel", () => {
    it.effect("peel", () =>
      Effect.gen(function*() {
        const sink = Sink.take<number>(3)
        const [peeled, rest] = yield* pipe(
          Stream.fromArrays(Array.range(1, 3), Array.range(4, 6)),
          Stream.peel(sink),
          Effect.flatMap(([peeled, rest]) =>
            pipe(
              Stream.runCollect(rest),
              Effect.map((rest) => [peeled, rest])
            )
          ),
          Effect.scoped
        )
        deepStrictEqual(peeled, [1, 2, 3])
        deepStrictEqual(rest, [4, 5, 6])
      }))

    it.effect("peel - propagates errors", () =>
      Effect.gen(function*() {
        const stream = Stream.fromEffectRepeat(Effect.fail("fail"))
        const sink = Sink.fold<ReadonlyArray<number>, number>(
          Array.empty,
          constTrue,
          (acc, n) => Effect.succeed(Array.append(acc, n))
        )
        const result = yield* pipe(
          stream,
          Stream.peel(sink),
          Effect.exit,
          Effect.scoped
        )
        deepStrictEqual(result, Exit.fail("fail"))
      }))
  })

  describe("repeat", () => {
    it.effect("repeat", () =>
      Effect.gen(function*() {
        const result = yield* pipe(
          Stream.make(1),
          Stream.repeat(Schedule.recurs(4)),
          Stream.runCollect
        )
        deepStrictEqual(result, [1, 1, 1, 1, 1])
      }))

    // it.effect("tick", () =>
    //   Effect.gen(function*() {
    //     const fiber = yield* pipe(
    //       Stream.tick("10 millis"),
    //       Stream.take(2),
    //       Stream.runCollect,
    //       Effect.fork
    //     )
    //     yield* (TestClock.adjust(Duration.millis(50)))
    //     const result = yield* (Fiber.join(fiber))
    //     deepStrictEqual(Array.from(result), [undefined, undefined])
    //   }))

    it.effect("repeat - short circuits", () =>
      Effect.gen(function*() {
        const ref = yield* Ref.make(Array.empty<number>())
        const fiber = yield* pipe(
          Stream.fromEffect(Ref.update(ref, Array.prepend(1))),
          Stream.repeat(Schedule.spaced(Duration.millis(10))),
          Stream.take(2),
          Stream.runDrain,
          Effect.forkChild
        )
        yield* TestClock.adjust(Duration.millis(50))
        yield* Fiber.join(fiber)
        const result = yield* Ref.get(ref)
        deepStrictEqual(result, [1, 1])
      }))

    // it.effect("repeat - Schedule.CurrentIterationMetadata", () =>
    //   Effect.gen(function*() {
    //     const ref = yield* (Ref.make(Chunk.empty<undefined | Schedule.IterationMetadata>()))
    //     const fiber = yield* pipe(
    //       Stream.fromEffect(
    //         Schedule.CurrentIterationMetadata.pipe(
    //           Effect.flatMap((currentIterationMetadata) => Ref.update(ref, Chunk.append(currentIterationMetadata)))
    //         )
    //       ),
    //       Stream.repeat(Schedule.exponential(Duration.millis(10))),
    //       Stream.runDrain,
    //       Effect.fork
    //     )
    //
    //     yield* (TestClock.adjust(Duration.millis(70)))
    //     yield* (Fiber.interrupt(fiber))
    //     const result = yield* (Ref.get(ref))
    //     deepStrictEqual(Array.from(result), [
    //       {
    //         elapsed: Duration.zero,
    //         elapsedSincePrevious: Duration.zero,
    //         input: undefined,
    //         output: undefined,
    //         now: 0,
    //         recurrence: 0,
    //         start: 0
    //       },
    //       {
    //         elapsed: Duration.zero,
    //         elapsedSincePrevious: Duration.zero,
    //         input: undefined,
    //         output: Duration.millis(10),
    //         now: 0,
    //         recurrence: 1,
    //         start: 0
    //       },
    //       {
    //         elapsed: Duration.millis(10),
    //         elapsedSincePrevious: Duration.millis(10),
    //         input: undefined,
    //         output: Duration.millis(20),
    //         now: 10,
    //         recurrence: 2,
    //         start: 0
    //       },
    //       {
    //         elapsed: Duration.millis(30),
    //         elapsedSincePrevious: Duration.millis(20),
    //         input: undefined,
    //         output: Duration.millis(40),
    //         now: 30,
    //         recurrence: 3,
    //         start: 0
    //       }
    //     ])
    //   }))

    it.effect("repeat - does not swallow errors on a repetition", () =>
      Effect.gen(function*() {
        const ref = yield* (Ref.make(0))
        const result = yield* pipe(
          Stream.fromEffect(pipe(
            Ref.getAndUpdate(ref, (n) => n + 1),
            Effect.flatMap((n) => n <= 2 ? Effect.succeed(n) : Effect.fail("boom"))
          )),
          Stream.repeat(Schedule.recurs(3)),
          Stream.runDrain,
          Effect.exit
        )
        deepStrictEqual(result, Exit.fail("boom"))
      }))

    it.effect("repeatElements - simple", () =>
      Effect.gen(function*() {
        const result = yield* pipe(
          Stream.make("A", "B", "C"),
          Stream.repeatElements(Schedule.recurs(1)),
          Stream.runCollect
        )
        deepStrictEqual(result, ["A", "A", "B", "B", "C", "C"])
      }))

    it.effect("repeatElements - short circuits in a schedule", () =>
      Effect.gen(function*() {
        const result = yield* pipe(
          Stream.make("A", "B", "C"),
          Stream.repeatElements(Schedule.recurs(1)),
          Stream.take(4),
          Stream.runCollect
        )
        deepStrictEqual(result, ["A", "A", "B", "B"])
      }))

    it.effect("repeatElements - short circuits after schedule", () =>
      Effect.gen(function*() {
        const result = yield* pipe(
          Stream.make("A", "B", "C"),
          Stream.repeatElements(Schedule.recurs(1)),
          Stream.take(3),
          Stream.runCollect
        )
        deepStrictEqual(result, ["A", "A", "B"])
      }))
  })

  describe("retry", () => {
    it.effect("retry - retries a failing stream", () =>
      Effect.gen(function*() {
        const ref = yield* Ref.make(0)
        const stream = pipe(
          Stream.fromEffect(Ref.getAndUpdate(ref, (n) => n + 1)),
          Stream.concat(Stream.fail(Option.none()))
        )
        const result = yield* pipe(
          stream,
          Stream.retry(Schedule.forever),
          Stream.take(2),
          Stream.runCollect
        )
        deepStrictEqual(Array.fromIterable(result), [0, 1])
      }))

    it.effect("retry - cleans up resources before restarting the stream", () =>
      Effect.gen(function*() {
        const ref = yield* Ref.make(0)
        const stream = pipe(
          Effect.addFinalizer(() => Ref.getAndUpdate(ref, (n) => n + 1)),
          Effect.as(
            pipe(
              Stream.fromEffect(Ref.get(ref)),
              Stream.concat(Stream.fail(Option.none()))
            )
          ),
          Stream.unwrap
        )
        const result = yield* pipe(
          stream,
          Stream.retry(Schedule.forever),
          Stream.take(2),
          Stream.runCollect
        )
        deepStrictEqual(result, [0, 1])
      }))

    it.effect("retry - retries a failing stream according to a schedule", () =>
      Effect.gen(function*() {
        const ref = yield* Ref.make(Array.empty<number>())
        const stream = pipe(
          Stream.fromEffect(
            pipe(
              Clock.currentTimeMillis,
              Effect.flatMap((n) => Ref.update(ref, Array.prepend(n)))
            )
          ),
          Stream.flatMap(() => Stream.fail(Option.none()))
        )
        const fiber = yield* pipe(
          stream,
          Stream.retry(Schedule.exponential(Duration.seconds(1))),
          Stream.take(3),
          Stream.runDrain,
          Effect.forkChild
        )
        yield* TestClock.adjust(Duration.seconds(1))
        yield* TestClock.adjust(Duration.seconds(2))
        yield* Fiber.interrupt(fiber)
        const result = yield* pipe(Ref.get(ref), Effect.map(Array.map((n) => new Date(n).getSeconds())))
        deepStrictEqual(Array.fromIterable(result), [3, 1, 0])
      }))

    it.effect("retry - reset the schedule after a successful pull", () =>
      Effect.gen(function*() {
        const times = yield* Ref.make(Array.empty<number>())
        const ref = yield* Ref.make(0)
        const effect = pipe(
          Clock.currentTimeMillis,
          Effect.flatMap((time) =>
            pipe(
              Ref.update(times, Array.prepend(time / 1000)),
              Effect.andThen(Ref.updateAndGet(ref, (n) => n + 1))
            )
          )
        )
        const stream = pipe(
          Stream.fromEffect(effect),
          Stream.flatMap((attempt) =>
            attempt === 3 || attempt === 5 ?
              Stream.succeed(attempt) :
              Stream.fail(Option.none())
          ),
          Stream.forever
        )
        const fiber = yield* pipe(
          stream,
          Stream.retry(Schedule.exponential(Duration.seconds(1))),
          Stream.take(2),
          Stream.runDrain,
          Effect.forkChild
        )
        yield* TestClock.adjust(Duration.seconds(1))
        yield* TestClock.adjust(Duration.seconds(2))
        yield* TestClock.adjust(Duration.seconds(1))
        yield* Fiber.join(fiber)
        const result = yield* Ref.get(times)
        deepStrictEqual(result, [4, 3, 3, 1, 0])
      }))

    it.effect("retry - Schedule.CurrentMetadata", () =>
      Effect.gen(function*() {
        const metadata = yield* Ref.make(Array.empty<undefined | Schedule.Metadata>())
        const fiber = yield* pipe(
          Stream.fail(1),
          Stream.tapError(Effect.fnUntraced(function*(_) {
            const currentMeta = yield* Schedule.CurrentMetadata
            yield* Ref.update(metadata, Array.append(currentMeta))
          })),
          Stream.retry(Schedule.exponential(Duration.seconds(1))),
          Stream.runDrain,
          Effect.forkChild({ startImmediately: true })
        )
        yield* TestClock.adjust(Duration.seconds(7))
        yield* Fiber.interrupt(fiber)
        const result = yield* Ref.get(metadata)
        deepStrictEqual(result, [
          {
            elapsed: 0,
            elapsedSincePrevious: 0,
            input: undefined,
            output: undefined,
            now: 0,
            attempt: 0,
            start: 0,
            duration: Duration.zero
          },
          {
            elapsed: 0,
            elapsedSincePrevious: 0,
            input: 1,
            output: Duration.millis(1000),
            now: 0,
            attempt: 1,
            start: 0,
            duration: Duration.millis(1000)
          },
          {
            elapsed: 1000,
            elapsedSincePrevious: 1000,
            input: 1,
            output: Duration.millis(2000),
            now: 1000,
            attempt: 2,
            start: 0,
            duration: Duration.millis(2000)
          },
          {
            elapsed: 3000,
            elapsedSincePrevious: 2000,
            input: 1,
            output: Duration.millis(4000),
            now: 3000,
            attempt: 3,
            start: 0,
            duration: Duration.millis(4000)
          }
        ])
      }))
  })

  describe("schedule", () => {
    it.effect("schedule", () =>
      Effect.gen(function*() {
        const start = yield* Clock.currentTimeMillis
        const fiber = yield* pipe(
          Stream.range(1, 8),
          Stream.schedule(Schedule.fixed(Duration.millis(100))),
          Stream.mapEffect((n) =>
            pipe(
              Clock.currentTimeMillis,
              Effect.map((now) => [n, now - start] as const)
            )
          ),
          Stream.runCollect,
          Effect.forkChild
        )
        yield* TestClock.adjust(Duration.millis(800))
        const result = yield* Fiber.join(fiber)
        deepStrictEqual(result, [
          [1, 100],
          [2, 200],
          [3, 300],
          [4, 400],
          [5, 500],
          [6, 600],
          [7, 700],
          [8, 800]
        ])
      }))
  })

  describe("sliding", () => {
    it.effect("sliding - returns a sliding window", () =>
      Effect.gen(function*() {
        const stream0 = Stream.fromArrays(
          Array.empty<number>(),
          Array.make(1),
          Array.empty<number>(),
          Array.make(2, 3, 4, 5)
        )
        const stream1 = pipe(
          Stream.empty,
          Stream.concat(Stream.make(1)),
          Stream.concat(Stream.empty),
          Stream.concat(Stream.make(2)),
          Stream.concat(Stream.make(3, 4, 5))
        )
        const stream2 = pipe(
          Stream.make(1),
          Stream.concat(Stream.empty),
          Stream.concat(Stream.make(2)),
          Stream.concat(Stream.empty),
          Stream.concat(Stream.make(3, 4, 5))
        )
        const stream3 = pipe(
          Stream.make(1),
          Stream.concat(Stream.make(2)),
          Stream.concat(Stream.make(3, 4, 5))
        )
        const result1 = yield* pipe(
          Stream.make(1, 2, 3, 4, 5),
          Stream.sliding(2),
          Stream.runCollect
        )
        const result2 = yield* pipe(
          stream0,
          Stream.sliding(2),
          Stream.runCollect
        )
        const result3 = yield* pipe(
          stream1,
          Stream.sliding(2),
          Stream.runCollect
        )
        const result4 = yield* pipe(
          stream2,
          Stream.sliding(2),
          Stream.runCollect
        )
        const result5 = yield* pipe(
          stream3,
          Stream.sliding(2),
          Stream.runCollect
        )
        const expected = [[1, 2], [2, 3], [3, 4], [4, 5]]
        deepStrictEqual(result1, expected as any)
        deepStrictEqual(result2, expected as any)
        deepStrictEqual(result3, expected as any)
        deepStrictEqual(result4, expected as any)
        deepStrictEqual(result5, expected as any)
      }))

    it.effect("sliding - returns all elements if chunkSize is greater than the size of the stream", () =>
      Effect.gen(function*() {
        const result = yield* pipe(
          Stream.range(1, 5),
          Stream.sliding(6),
          Stream.runCollect
        )
        deepStrictEqual(result, [[1, 2, 3, 4, 5]])
      }))

    it.effect("sliding - is mostly equivalent to ZStream#grouped when stepSize and chunkSize are equal", () =>
      Effect.gen(function*() {
        const stream = Stream.range(1, 5)
        const { result1, result2 } = yield* (Effect.all({
          result1: pipe(stream, Stream.slidingSize(3, 3), Stream.runCollect),
          result2: pipe(stream, Stream.grouped(3), Stream.runCollect)
        }))
        deepStrictEqual(result1, result2)
      }))

    it.effect("sliding - fails if upstream produces an error", () =>
      Effect.gen(function*() {
        const result = yield* pipe(
          Stream.make(1, 2, 3),
          Stream.concat(Stream.fail("Ouch")),
          Stream.concat(Stream.make(4, 5)),
          Stream.sliding(2),
          Stream.runCollect,
          Effect.flip
        )
        assert.strictEqual(result, "Ouch")
      }))

    it.effect("sliding - should return an empty chunk when the stream is empty", () =>
      Effect.gen(function*() {
        const result = yield* pipe(Stream.empty, Stream.sliding(2), Stream.runCollect)
        deepStrictEqual(result, [])
      }))

    it.effect("sliding - emits elements properly when a failure occurs", () =>
      Effect.gen(function*() {
        const ref = yield* Ref.make(Array.empty<Array.NonEmptyReadonlyArray<number>>())
        const streamChunks = Stream.fromArrays(
          Array.range(1, 4),
          Array.range(5, 7),
          Array.of(8)
        )
        const stream = pipe(
          streamChunks,
          Stream.concat(Stream.fail("Ouch")),
          Stream.slidingSize(3, 3)
        )
        const either = yield* pipe(
          stream,
          Stream.mapEffect((chunk) => Ref.update(ref, Array.append(chunk))),
          Stream.runCollect,
          Effect.flip
        )
        const result = yield* (Ref.get(ref))
        assert.strictEqual(either, "Ouch")
        deepStrictEqual(result, [[1, 2, 3], [4, 5, 6], [7, 8]])
      }))
  })

  describe("split", () => {
    it.effect("split - should split properly", () =>
      Effect.gen(function*() {
        const chunks = Array.make(
          Array.range(1, 2),
          Array.range(3, 4),
          Array.range(5, 6),
          Array.make(7, 8, 9),
          Array.of(10)
        )
        const { result1, result2 } = yield* (Effect.all({
          result1: pipe(
            Stream.range(0, 9),
            Stream.split((n) => n % 4 === 0),
            Stream.runCollect
          ),
          result2: pipe(
            Stream.fromArrays(...chunks),
            Stream.split((n) => n % 3 === 0),
            Stream.runCollect
          )
        }))
        deepStrictEqual(
          result1,
          [[1, 2, 3], [5, 6, 7], [9]]
        )
        deepStrictEqual(
          result2,
          [[1, 2], [4, 5], [7, 8], [10]]
        )
      }))

    it.effect("split - is equivalent to identity when the predicate is not satisfied", () =>
      Effect.gen(function*() {
        const stream = Stream.range(1, 10)
        const { result1, result2 } = yield* (Effect.all({
          result1: pipe(stream, Stream.split((n) => n % 11 === 0), Stream.runCollect),
          result2: pipe(
            Stream.runCollect(stream),
            Effect.map((chunk) => pipe(Array.of(chunk), Array.filter(Array.isArrayNonEmpty)))
          )
        }))
        deepStrictEqual(result1, [Array.range(1, 10)])
        deepStrictEqual(result1, result2)
      }))

    it.effect("split - should output empty chunk when stream is empty", () =>
      Effect.gen(function*() {
        const result = yield* pipe(
          Stream.empty,
          Stream.split((n: number) => n % 11 === 0),
          Stream.runCollect
        )
        deepStrictEqual(result, [])
      }))
  })

  describe("timeout", () => {
    it.effect("timeoutOrElse - succeed", () =>
      Effect.gen(function*() {
        const result = yield* pipe(
          Stream.succeed(1),
          Stream.timeoutOrElse({
            duration: Duration.infinity,
            orElse: () => Stream.succeed(-1)
          }),
          Stream.runCollect
        )
        deepStrictEqual(result, [1])
      }))

    it.effect("timeoutOrElse - should switch streams", () =>
      Effect.gen(function*() {
        const result = yield* pipe(
          Stream.range(0, 4),
          Stream.tap(() => Effect.sleep(Duration.infinity)),
          Stream.timeoutOrElse({
            duration: Duration.zero,
            orElse: () => Stream.succeed(4)
          }),
          Stream.runCollect
        )
        deepStrictEqual(result, [4])
      }))

    it.effect("timeoutOrElse - should not apply timeout after switch", () =>
      Effect.gen(function*() {
        const fiber = yield* pipe(
          Stream.never,
          Stream.timeoutOrElse({
            duration: Duration.zero,
            orElse: () =>
              Stream.succeed(1).pipe(
                Stream.tap(() => Effect.sleep(Duration.seconds(1)))
              )
          }),
          Stream.runCollect,
          Effect.forkChild({ startImmediately: true })
        )
        yield* TestClock.adjust(Duration.seconds(1))
        const result = yield* Fiber.join(fiber)
        deepStrictEqual(result, [1])
      }))

    it.effect("timeout - succeed", () =>
      Effect.gen(function*() {
        const result = yield* pipe(
          Stream.succeed(1),
          Stream.timeout(Duration.infinity),
          Stream.runCollect
        )
        deepStrictEqual(result, [1])
      }))

    it.effect("timeout - should end the stream", () =>
      Effect.gen(function*() {
        const result = yield* pipe(
          Stream.range(0, 4),
          Stream.tap(() => Effect.sleep(Duration.infinity)),
          Stream.timeout(Duration.zero),
          Stream.runCollect
        )
        deepStrictEqual(result, [])
      }))
  })

  describe("when", () => {
    it.effect("when - returns the stream if the condition is satisfied", () =>
      Effect.gen(function*() {
        const stream = Stream.make(1, 2, 3, 4, 5)
        const { result1, result2 } = yield* (Effect.all({
          result1: pipe(stream, Stream.when(Effect.succeed(true)), Stream.runCollect),
          result2: Stream.runCollect(stream)
        }))
        deepStrictEqual(result1, result2)
      }))

    it.effect("when - returns an empty stream if the condition is not satisfied", () =>
      Effect.gen(function*() {
        const result = yield* pipe(
          Stream.make(1, 2, 3, 4, 5),
          Stream.when(Effect.succeed(false)),
          Stream.runCollect
        )
        deepStrictEqual(result, [])
      }))

    it.effect("when - dies if the condition throws an exception", () =>
      Effect.gen(function*() {
        const error = "boom"
        const result = yield* pipe(
          Stream.make(1, 2, 3),
          Stream.when(Effect.sync(() => {
            throw error
          })),
          Stream.runDrain,
          Effect.exit
        )
        deepStrictEqual(result, Exit.die(error))
      }))

    it.effect("when - returns the stream if the effectful condition is satisfied", () =>
      Effect.gen(function*() {
        const result = yield* pipe(
          Stream.make(1, 2, 3, 4, 5),
          Stream.when(Effect.succeed(true)),
          Stream.runCollect
        )
        deepStrictEqual(result, [1, 2, 3, 4, 5])
      }))

    it.effect("when - returns an empty stream if the effectful condition is not satisfied", () =>
      Effect.gen(function*() {
        const result = yield* pipe(
          Stream.make(1, 2, 3, 4, 5),
          Stream.when(Effect.succeed(false)),
          Stream.runCollect
        )
        deepStrictEqual(result, [])
      }))

    it.effect("when - fails if the effectful condition fails", () =>
      Effect.gen(function*() {
        const error = "boom"
        const result = yield* pipe(
          Stream.make(1, 2, 3),
          Stream.when(Effect.fail(error)),
          Stream.runDrain,
          Effect.exit
        )
        deepStrictEqual(result, Exit.fail(error))
      }))
  })

  describe("zipWithNext", () => {
    it.effect("zipWithNext", () =>
      Effect.gen(function*() {
        const result = yield* pipe(
          Stream.make(1, 2, 3),
          Stream.zipWithNext,
          Stream.runCollect
        )
        deepStrictEqual(result, [
          [1, Option.some(2)],
          [2, Option.some(3)],
          [3, Option.none()]
        ])
      }))

    it.effect("zipWithNext - should work with multiple chunks", () =>
      Effect.gen(function*() {
        const result = yield* pipe(
          Stream.fromArrays([1], [2], [3]),
          Stream.zipWithNext,
          Stream.runCollect
        )
        deepStrictEqual(result, [
          [1, Option.some(2)],
          [2, Option.some(3)],
          [3, Option.none()]
        ])
      }))

    it.effect("zipWithNext - should work with an empty stream", () =>
      Effect.gen(function*() {
        const result = yield* pipe(
          Stream.empty,
          Stream.zipWithNext,
          Stream.runCollect
        )
        deepStrictEqual(result, [])
      }))

    it.effect("zipWithPrevious - should zip with previous element for a single chunk", () =>
      Effect.gen(function*() {
        const result = yield* pipe(
          Stream.make(1, 2, 3),
          Stream.zipWithPrevious,
          Stream.runCollect
        )
        deepStrictEqual(result, [
          [Option.none(), 1],
          [Option.some(1), 2],
          [Option.some(2), 3]
        ])
      }))

    it.effect("zipWithPrevious - should work with multiple chunks", () =>
      Effect.gen(function*() {
        const result = yield* pipe(
          Stream.fromArrays([1], [2], [3]),
          Stream.zipWithPrevious,
          Stream.runCollect
        )
        deepStrictEqual(result, [
          [Option.none(), 1],
          [Option.some(1), 2],
          [Option.some(2), 3]
        ])
      }))

    it.effect("zipWithPrevious - should work with an empty stream", () =>
      Effect.gen(function*() {
        const result = yield* pipe(
          Stream.empty,
          Stream.zipWithPrevious,
          Stream.runCollect
        )
        deepStrictEqual(result, [])
      }))

    it("zipWithPrevious - should output same values as first element plus zipping with init", () =>
      fc.assert(fc.asyncProperty(fc.array(fc.array(fc.integer())), async (chunks) => {
        const stream = Stream.fromArrays(...chunks)
        const { result1, result2 } = await Effect.runPromise(Effect.all({
          result1: pipe(
            stream,
            Stream.zipWithPrevious,
            Stream.runCollect
          ),
          result2: pipe(
            Stream.make(Option.none()),
            Stream.concat(pipe(stream, Stream.map(Option.some))),
            Stream.zip(stream),
            Stream.runCollect
          )
        }))
        deepStrictEqual(result1, result2)
      })))

    it.effect("zipWithPreviousAndNext", () =>
      Effect.gen(function*() {
        const result = yield* pipe(
          Stream.make(1, 2, 3),
          Stream.zipWithPreviousAndNext,
          Stream.runCollect
        )
        deepStrictEqual(result, [
          [Option.none(), 1, Option.some(2)],
          [Option.some(1), 2, Option.some(3)],
          [Option.some(2), 3, Option.none()]
        ])
      }))

    it("zipWithPreviousAndNext - should output same values as zipping with both previous and next element", () =>
      fc.assert(fc.asyncProperty(fc.array(fc.array(fc.integer()), { minLength: 0, maxLength: 5 }), async (chunks) => {
        const stream = Stream.fromArrays(...chunks)
        const previous = pipe(
          Stream.make(Option.none()),
          Stream.concat(pipe(stream, Stream.map(Option.some)))
        )
        const next = pipe(
          stream,
          Stream.drop(1),
          Stream.map(Option.some),
          Stream.concat(Stream.make(Option.none()))
        )
        const { result1, result2 } = await pipe(
          Effect.all({
            result1: pipe(
              stream,
              Stream.zipWithPreviousAndNext,
              Stream.runCollect
            ),
            result2: pipe(
              previous,
              Stream.zip(stream),
              Stream.zipFlatten(next),
              Stream.runCollect
            )
          }),
          Effect.runPromise
        )
        deepStrictEqual(result1, result2)
      })))
  })

  describe("broadcastN", () => {
    it.effect("fans out to a fixed number of streams", () =>
      Effect.scoped(Effect.gen(function*() {
        const [left, right] = yield* Stream.make(1, 2, 3).pipe(
          Stream.broadcastN({ n: 2, capacity: 4 })
        )

        const result = yield* Effect.all([
          Stream.runCollect(left),
          Stream.runCollect(right)
        ], { concurrency: "unbounded" })

        assert.deepStrictEqual(result, [[1, 2, 3], [1, 2, 3]])
      })))

    it.effect("propagates failures to all downstream streams", () =>
      Effect.scoped(Effect.gen(function*() {
        const [left, right] = yield* Stream.fail("boom").pipe(
          Stream.broadcastN({ n: 2, capacity: 4 })
        )

        const result = yield* Effect.all([
          Stream.runCollect(left).pipe(Effect.exit),
          Stream.runCollect(right).pipe(Effect.exit)
        ], { concurrency: "unbounded" })

        assert.deepStrictEqual(result, [Exit.fail("boom"), Exit.fail("boom")])
      })))
  })
})

const grouped = <A>(arr: Array<A>, size: number): Array<NonEmptyArray<A>> => {
  const builder: Array<NonEmptyArray<A>> = []
  for (let i = 0; i < arr.length; i = i + size) {
    builder.push(arr.slice(i, i + size) as NonEmptyArray<A>)
  }
  return builder
}