Upstream_output.definition_metadata doesn't contain the metadata since 1.7.11 for observable Source Assets [REGRESSION] about dagster HOT 8 CLOSED

def test_input_manager_with_source_assets() -> None:
    fancy_metadata = {"foo": "bar", "baz": 1.23}

    upstream = SourceAsset("upstream", metadata=fancy_metadata)

    @asset(ins={"upstream": AssetIn(input_manager_key="special_io_manager")})
    def downstream(upstream) -> int:
        return upstream + 1

    class MyIOManager(IOManager):
        def load_input(self, context) -> int:
            assert context.upstream_output is not None
            assert context.upstream_output.asset_key == AssetKey(["upstream"])
            assert context.upstream_output.definition_metadata == fancy_metadata

            return 2

        def handle_output(self, context, obj) -> None: ...

    defs = Definitions(
        assets=[upstream, downstream],
        resources={"special_io_manager": IOManagerDefinition.hardcoded_io_manager(MyIOManager())},
    )
    job = defs.get_implicit_job_def_for_assets([downstream.key])
    assert job is not None
    output = job.execute_in_process()

    assert output._get_output_for_handle("downstream", "result") == 3  # noqa: SLF001
    assert False

from dagster import (
    AssetIn,
    AssetKey,
    ConfigurableIOManager,  # noqa
    DataVersion,
    IOManager,  # noqa
    asset,
    materialize,
    observable_source_asset,
)


def test_input_manager_with_source_assets() -> None:
    fancy_metadata = {"foo": "bar", "baz": 1.23}

    @observable_source_asset(metadata=fancy_metadata)
    def upstream():
        return DataVersion('1')
    # upstream = SourceAsset("upstream", metadata=fancy_metadata)

    @asset(ins={"upstream": AssetIn(input_manager_key="special_io_manager")})
    def downstream(upstream) -> int:
        return upstream + 1

    class MyIOManager(IOManager):
        def load_input(self, context) -> int:
            assert context.upstream_output is not None
            assert context.upstream_output.asset_key == AssetKey(["upstream"])
            assert context.upstream_output.definition_metadata == fancy_metadata
            return 2

        def handle_output(self, context, obj) -> None: ...

    materialize(assets=[upstream,downstream], resources={"special_io_manager": MyIOManager()})

AssertionError: assert {'dagster/io_... 'io_manager'} == {'baz': 1.23, 'foo': 'bar'}
  Left contains 1 more item:
  {'dagster/io_manager_key': 'io_manager'}
  Right contains 2 more items:
  {'baz': 1.23, 'foo': 'bar'}
  Use -v to get more diff