fix: pipe: Tag slice with run number on input

Previously, the run number was taken at the time of output, from the
concurrent control  structure, which can be out of sync with the
original data