UpdateLocalContextProblem.scala

import cats.data.ReaderT
import cats.effect.{Concurrent, IO, Sync}
import cats.mtl.ApplicativeLocal
import fs2.Stream

object UpdateLocalContextProblem extends App {
  type Ctx = String
  type Header = String

  type ReaderIO[A] = ReaderT[IO, Ctx, A]
  case class Message(payload: String, header: Header)

  def toCtx(header: Header): Ctx = s"$header-updated"

  case class Consumer[F[_] : Concurrent](messages: Seq[Message])(implicit val local: ApplicativeLocal[F, Ctx]) {
    def consume(): Stream[F, Ctx] = {
      Stream.emits(messages).flatMap[F, Message] {
        msg =>
          val ctx = toCtx(msg.header)
          val fun: F[Message] => F[Message] = local.scope[Message](ctx)
          Stream.eval(fun(Sync[F].pure(msg)))
      }.evalMap {
        _ => local.ask // returns ctx, not ctx-updated
      }
    }
  }

  implicit val ioCs = IO.contextShift(scala.concurrent.ExecutionContext.global)
  import cats.mtl.implicits._

  val result = Consumer[ReaderIO](Seq(Message("value1", "header1"))).consume().compile.toList.run("ctx").unsafeRunSync()
  println(result) // prints List(ctx), but I want to print it ctx-updated
}