# mapAccumEffect

Maps over the output (Fx) side of a `Push` with an effectful accumulator: for each emitted value `b`,
runs `f(state, b)` to get `[nextState, emitted]` and emits the second element.
The adapter does not serialize callbacks. Calling `onSuccess` invokes `f`
immediately with the current seed and constructs its Effect; overlapping calls
can therefore observe the same seed. Successful completion commits the returned
seed and emits in completion order, so later completion may overwrite newer
state. Reducer failure is sent to the output Sink, emits nothing, restores that
call's previous seed, and completes normally so a continuing producer can send
later values. The input Sink is unchanged.

## Signatures

```ts
export declare const mapAccumEffect: {
    <S, B, C, E3, R3>(initial: S, f: (s: S, b: B) => Effect.Effect<readonly [
        S,
        C
    ], E3, R3>): <A, E, R, E2, R2>(push: Push<A, E, R, B, E2, R2>) => Push<A, E, R, C, E2 | E3, R2 | R3>;
    <A, E, R, B, E2, R2, S, C, E3, R3>(push: Push<A, E, R, B, E2, R2>, initial: S, f: (s: S, b: B) => Effect.Effect<readonly [
        S,
        C
    ], E3, R3>): Push<A, E, R, C, E2 | E3, R2 | R3>;
};
```

## Why

Use an effectful accumulator when each transition needs services or can fail,
while keeping state local to observation rather than global application state.

## Ownership and lifetime

Each output run owns one mutable seed, but concurrent producer callbacks may
race over it; callers needing serialized state transitions must serialize the
upstream deliveries. Each callback fiber runs and interrupts its own reducer
Effect. `E3` and `R3` join the output channels. State is discarded when the
subscription ends.

## Examples

```ts
import { Effect } from "effect"
import * as Fx from "@typed/fx/Fx"
import * as Push from "@typed/fx/Push"
import * as Sink from "@typed/fx/Sink"

const concurrent = Fx.make<number>((sink) =>
  Effect.all([sink.onSuccess(1), sink.onSuccess(2)], {
    concurrency: "unbounded",
    discard: true
  })
)
const push = Push.make(Sink.make(() => Effect.void, (_: string) => Effect.void), concurrent)
const totals = Push.mapAccumEffect(push, 0, (seed, value) =>
  Effect.sleep(value === 1 ? "20 millis" : "1 millis").pipe(
    Effect.as([seed + value, seed + value] as const)
  )
)
const program = Fx.collectAll(totals).pipe(Effect.scoped)
// Both reducers see seed 0; Effect.runPromise(program) resolves to [2, 1].

const values: Array<number> = []
let failures = 0
const sequential = Push.make(
  Sink.make(() => Effect.void, (_: string) => Effect.void),
  Fx.fromIterable([1, 2, 3])
)
const continued = Push.mapAccumEffect(sequential, 0, (seed, value) =>
  value === 2
    ? Effect.fail("rejected" as const)
    : Effect.succeed([seed + value, seed + value] as const)
)
const recovery = continued.run(Sink.make(
  () => Effect.sync(() => { failures += 1 }),
  (value) => Effect.sync(() => values.push(value))
))
// Effect.runPromise(recovery) leaves values [1, 4] and failures 1.
```