anchor.diff:                  | Message(x) ->
                       let isAcceptable = condition x
                       if isAcceptable then
  -                        nextAgent.Post msg
  -                    return! messageLoop()
  -                | End(replyChannel) ->
  -                    replyChannel.Reply()
  -                | Done(replyChannel) ->
  -                    nextAgent.PostAndReply(Done)
  +                        nextAgent.Tell x
  +                    return! messageLoop()
  +                | End(replyChannel) ->
  +                    replyChannel.Reply()
  +                | Done(replyChannel) ->
  +                    nextAgent.Done()
                       replyChannel.Reply()
                       return! messageLoop()                
                   | Collect(replyChannel) ->
  
positive.code:         Pipe.Pipe(
              fun nextAgent -> 
                  use agent = mapAgent nextAgent id
                  items |> Seq.iter (Message >> agent.Post)
                  agent.PostAndReply(Done)
          )
          
      ///Maps items in the Pipe to a new item.
  
negative.code:         use agent = collectAgent()
          let (Pipe.Pipe previousPipeCall) = pipe
          previousPipeCall agent
          agent.Collect()
      
      ///Only evaluates Pipe once, lazily.
      let Cache (pipe: Pipe<'a>) =
  