louthy/language-ext · error · InvalidOperationException

closed

Error message

closed

What it means

PipeTYield is an intermediate node of the PipeT composition tree: it holds a value that still needs to be paired with a consumer/producer and reduced to a runnable K<M, A>. Its Run() is intentionally unimplemented and throws InvalidOperationException("closed"). The library throws it when you call Run() (directly or via an API like PipeT.run) on a pipe value that is still in the 'yield' state rather than a fully interpreted pipe, i.e. the pipe has already been consumed/closed into a terminal form but is being evaluated as a stream program.

Solutions

1. Drive the pipe through the intended interpreter instead of Run(): use the Producer/Consumer/Pipe run entry points (e.g. Producer.run, Consumer.run, or PipeT combinators that supply a producer for awaits) so every await/yield is paired before evaluation.
2. If you just have a final value, use PipeT.pure or ensure the composition ends in PipeTPure; a pipe that ends in a yield has no meaning without a consumer.
3. When writing a custom fold/interpreter over PipeT, handle the PipeTYield case explicitly instead of delegating to Run().
4. Check LanguageExt version: PipeT (LanguageExt.Streaming) is a new API; consult its samples for the correct run pattern rather than calling internal Run().

Example fix

// before
var pipe = from u in PipeT.yield<IN, OUT, M>(value) select u;
var result = pipe.Run(); // throws InvalidOperationException("closed")

// after
var result = Producer.run(source, pipe.Map(_ => done)); // interpret the pipe against a real producer/consumer

// Ensure the pipe is interpreted via the producer/consumer runners, never .Run():
// if you hold a PipeT that came from yield/bind, hand it to Producer.run / Consumer.run.
var interpreted = Producer.run(source, pipe); // instead of pipe.Run()

static bool IsRunnable<A>(K<PipeT<IN, OUT, M>, A> p) =>
    p is not PipeTYield<IN, OUT, M, A> and not PipeTAwait<IN, OUT, M, A>;

try { result = Producer.run(source, pipe); }
catch (InvalidOperationException e) when (e.Message == "closed")
{
    // pipe was run in an unpaired (yield) state; rebuild with a producer/consumer
}

When it happens

Trigger: Calling .Run() on a PipeTYield<IN, OUT, M, A> — e.g. invoking PipeT.run / Run on the result of a `yield` step (or a Bind/Map chain ending in a yield) instead of driving the pipe through a producer with the interpreter (PipeT.run with producer/consumer, e.g. via Producer/Consumer/Pipe run helpers).

Common situations: Treating PipeT like a monad you can 'execute' after composing yields; misusing an intermediate value from Bind continuation; refactoring code from older LanguageExt Pipes (Pipe) to the new PipeT streaming API where run was implicit; writing custom interpreters that fall through to the default Run() on yield nodes.

Understand the failure class

Background: "Invalid state transition" errors: "status must be X, actually Y", "already rejected/charging/uninstalled", "cannot ... while running" — what they mean when a library rejects your call — this error's family across 31 libraries.

Related errors

• Ord attribute should have a struct type that derives from…
• Hashable attribute should have a struct type that derives…
• Don't use Equals - use either RecordType
• Don't use Equals - use either RecordType
• s

AI-assisted analysis of louthy/language-ext@2f0e362824 (2026-09-15). Data as JSON: /api/errors/ceae1044e427c748. Report an issue: GitHub.

Appendix: source

Thrown at LanguageExt.Streaming/Pipes/PipeT/PipeT.DSL.cs:279

        new PipeTYield<IN, OUT, M, B>(Value, _ => Next(default).Action(fb));

    public override PipeT<IN, OUT, M, B> Bind<B>(Func<A, PipeT<IN, OUT, M, B>> f) => 
        new PipeTYield<IN, OUT, M, B>(Value, _ => Next(default).Bind(f));

    internal override PipeT<IN1, OUT, M, A> ReplaceAwait<IN1>(Func<PipeT<IN1, OUT, M, IN>> producer) =>
        new PipeTYield<IN1, OUT, M, A>(Value, _ => Next(default).ReplaceAwait(producer));
    
    internal override PipeT<IN, OUT1, M, A> ReplaceYield<OUT1>(Func<OUT, PipeT<IN, OUT1, M, Unit>> consumer) =>
        consumer(Value).Bind(_ => Next(default).ReplaceYield(consumer));

    internal override PipeT<IN1, OUT, M, A> PairEachAwaitWithYield<IN1>(Func<Unit, PipeT<IN1, IN, M, A>> producer) =>
        new PipeTYield<IN1, OUT, M, A>(Value, _ => Next(default).PairEachAwaitWithYield(producer));

    internal override PipeT<IN, OUT1, M, A> PairEachYieldWithAwait<OUT1>(Func<OUT, PipeT<OUT, OUT1, M, A>> consumer) =>
        consumer(Value).PairEachAwaitWithYield(_ => Next(default));

    internal override K<M, A> Run() => 
        throw new InvalidOperationException("closed");
}

record PipeTAwait<IN, OUT, M, A>(Func<IN, PipeT<IN, OUT, M, A>> Await) : PipeT<IN, OUT, M, A>
    where M : MonadIO<M>
{
    public override PipeT<IN, OUT, M, B> Map<B>(Func<A, B> f) => 
        new PipeTAwait<IN, OUT, M, B>(x => Await(x).Map(f));

    public override PipeT<IN, OUT, M, B> MapM<B>(Func<K<M, A>, K<M, B>> f) => 
        new PipeTAwait<IN, OUT, M, B>(x => Await(x).MapM(f));

    public override PipeT<IN, OUT, M, B> ApplyBack<B>(PipeT<IN, OUT, M, Func<A, B>> ff) => 
        new PipeTAwait<IN, OUT, M, B>(x => Await(x).ApplyBack(ff));

    public override PipeT<IN, OUT, M, B> Action<B>(PipeT<IN, OUT, M, B> fb) => 
        new PipeTAwait<IN, OUT, M, B>(x => Await(x).Action(fb));

    public override PipeT<IN, OUT, M, B> Bind<B>(Func<A, PipeT<IN, OUT, M, B>> f) => 

View on GitHub (pinned to 2f0e362824)