The on_stream and on_stream_end closures in _dispatch hardcoded
their metadata dicts, dropping channel-specific fields like
message_thread_id. Copy msg.metadata first, then add internal
streaming flags, matching the pattern already used by _bus_progress.
