Unexpected behaviour from Elixir streams

stream_behaviour_test.ex

defmodule StreamBehaviourTest do                                                                                     
  use ExUnit.Case, async: true

  test "will not pull from upstream after halt" do
    called = fn key, x -> Process.put(key, [x|Process.get(key, [])]) end 

    results = 1..10
              |> Stream.each(&called.(:upstream, &1))
              |> Stream.take(1)
              |> Stream.each(&called.(:downstream, &1))
              |> Enum.to_list

    assert results == [1]                       # as expected
    assert Process.get(:downstream, []) == [1]  # as expected
    assert Process.get(:upstream, []) == [1]    # if fails
  end 
end

stream_behaviour_test.output

  1) test will not pull from upstream after halt (StreamBehaviour)
     test/paco_stream_test.exs:148
     Assertion with == failed
     code: Process.get(:upstream, []) == [1]
     lhs:  [2, 1]
     rhs:  [1]
     stacktrace:
       test/paco_stream_test.exs:159


Finished in 0.1 seconds (0.09s on load, 0.02s on tests)
1 tests, 1 failures

Randomized with seed 848524