Just an idea:

I think the algorithm could be easier to read if sinks and subscriptions were organized a bit differently. We could define a class called Deduplicator, which holds sinks and subs for a given request key. It would have three states: ACTIVE, PASSIVE and DONE. Any deduplicator starts as ACTIVE, and there is no more than one active deduplicator for a given request key. It is the one that currently deduplicates requests. This is how we can change states:

ACTIVE => PASSIVE
PASSIVE => PASSIVE
  event: we received a response from delegate

ACTIVE => DONE
  event: delegate stream is closed
  event: last subscriber has left deduplicator
  action: cleanup

PASSIVE => DONE
  event: delegate stream is closed
  event: last subscriber has left deduplicator
  action: cleanup

This would capture the "before data" case as a deduplicator can only be ACTIVE if it hasn't received any responses from the delegate. It also doesn't require tracking of first response as receiving a response only takes an active deduplicator to passive and it simply stays passive from then on.