# grouped

Partitions the stream into non-empty arrays of size `n`. The final array
may be smaller if there are leftover elements.
The size must be a positive safe integer. A group can retain up to `n`
values, so callers own the memory policy for valid sizes. Invalid sizes
fail with `Cause.IllegalArgumentError`.

Matches Effect `Stream.grouped`.

## Signatures

```ts
export declare const grouped: {
    (n: number): <A, E, R>(self: Fx<A, E, R>) => Fx<NonEmptyReadonlyArray<A>, E | Cause.IllegalArgumentError, R>;
    <A, E, R>(self: Fx<A, E, R>, n: number): Fx<NonEmptyReadonlyArray<A>, E | Cause.IllegalArgumentError, R>;
};
```

## Why

`grouped` exposes fixed-size batching without changing source order. Full groups contain exactly
`n` values. Because failure is delivered to the Sink while `Fx.run` remains infallible, any
partial group is flushed after the source run returns even if a failure was delivered first.

## Ownership and lifetime

Each run retains at most `n` values and releases its buffer after the final flush or interruption.
Invalid sizes deliver failure before source acquisition. A terminal observer may interrupt before
the post-failure flush, but a Sink that handles the Cause can receive the partial group afterward.

## Examples

```ts
import { Cause, Effect, Ref } from "effect"
import { Fx, Sink } from "@typed/fx"
const program = Effect.gen(function* () {
  const deliveries = yield* Ref.make<Array<string>>([])
  const source = Fx.make<number, string>((sink) =>
    sink.onSuccess(1).pipe(Effect.andThen(sink.onFailure(Cause.fail("boom"))))
  )
  yield* Fx.grouped(source, 2).run(Sink.make(
    () => Ref.update(deliveries, (xs) => [...xs, "failure"]),
    (group) => Ref.update(deliveries, (xs) => [...xs, `group:${group.join(",")}`])
  ))
  return yield* Ref.get(deliveries) // ["failure", "group:1"]
})
```