deterministic cleanup for local query outputs
I fixed local one-shot reads and streams retaining internal listeners after they had finished or been dropped. They now clean themselves up immediately, including error paths, while shared streams stay alive until their last owner leaves; regression tests cover these cases.