I couldn't use StateT to manage state because the IO () function passed in has to be able to update the state and I don't want to make it use StateT as well so an IORef does the job nicely.

Stateful Controller

runController :: (MonadIO m) => Controller a -> Producer a m ()
runController (Controller m) = do
  input <- liftIO m
  let loop = do
        xM <- liftIO (atomically $ recv input)
        maybe (return ()) (\x -> yield x >> loop) xM
  loop

type StateFn s = (Maybe s -> s) -> IO s

runStateful :: Controller (StateFn s -> IO ()) -> IO ()
runStateful c = runSafeT . runEffect $ runController c >-> manageState
  where
    manageState = do
      v <- liftIO $ newIORef Nothing
      let stateF f =
            liftIO $ atomicModifyIORef v (\s -> let s' = f s in (Just s', s'))
          go f = bracket (async $ f stateF) cancel (\_ -> await) >>= go
      await >>= go

runStateful' :: Controller (StateFn s -> Producer a IO ()) -> Controller a
runStateful' c = Controller $ do
  (output, input) <- liftIO $ spawn Unbounded
  let stateF v f = atomicModifyIORef v (\s -> let s' = f s in (Just s', s'))
      put = do
        ok <- await >>= liftIO . atomically . send output
        when ok put
      manageState = do
        v <- liftIO $ newIORef Nothing
        let go f =
              bracket (async . runEffect $ f (liftIO . stateF v) >-> put)
                (\a -> cancel a >> performGC)
                (\_ -> await) >>= go
        await >>= go
  liftIO $ runSafeT . runEffect $ runController c >-> manageState
  return input

runStateful' seems to work but it doesn't stream in a natural way - it buffers up all its output and dumps it in one go. I'm not sure why yet.