fix(config_flow): D3 — await the discovery-dedup coroutine (t37)

`ConfigFlow._async_handle_discovery_without_unique_id` is an `async def` in core
(verified: `inspect.iscoroutinefunction` is True) and core awaits it. t32 called
it without `await`, so the coroutine body never ran: no `already_configured`
abort when the handler already had entries, no
`DEFAULT_DISCOVERY_UNIQUE_ID` claim, no `already_in_progress` abort — HA's
discovery dedup was dead, Python warned "coroutine ... was never awaited", and
the host-equality loop that used to sit here was the only accidental dedup.

The call is awaited and the now-unreachable host loop is gone (the real helper
is stricter: any entry for the domain, or any in-flight flow). The harness
mirrors the real contract instead of a sync no-op:

* `entity_stublib`'s hub stub makes the helper a coroutine (a sync stub would
  turn the production `await` into a TypeError);
* `test_standalone_setup`'s stub models the real semantics — a coroutine raising
  `DiscoveryDedupAbortError` (core raises AbortFlow), driven by the shared
  holder's entry list and in-progress list — and its tests now assert all three
  outcomes: awaited-and-proceeds (counter + claimed discovery unique id),
  `already_configured` when entries exist, `already_in_progress` when another
  flow runs. A fourth test collects the un-awaited-coroutine warning, and the
  prefill test in the UX suite asserts the awaited counter.

Red proof: with the pre-fix config_flow.py the repo-owned sweep reports
`config_flow.py:425: missing-await: self._async_handle_discovery_without_unique_id()`.

Verified: 289 passed, 2 skipped (D3 stage); ruff clean.

tests/test_standalone_setup.py

@@ -8,8 +8,10 @@ migration validity for v2 entries without the field.
 """
 
 import asyncio  # noqa: F401  (async steps need the event loop)
+import gc
 import sys
 import types
+import warnings
 from pathlib import Path  # noqa: E402 (F1 test key staging)
 from types import SimpleNamespace
 from unittest import mock
@@ -42,6 +44,19 @@ _STUB_NAMES = (
 )
 
 
+class DiscoveryDedupAbortError(Exception):
+    """Stands in for data_entry_flow.AbortFlow inside the discovery stub.
+
+    HA's `ConfigFlow._async_handle_discovery_without_unique_id` RAISES on a
+    duplicate (the flow manager turns that into an abort result); modelling it
+    as a raise is what makes an un-awaited call fail loudly here.
+    """
+
+    def __init__(self, reason):
+        super().__init__(reason)
+        self.reason = reason
+
+
 def _install_stubs() -> None:
     """Import-time stand-ins (idempotent; same shape as the pairing tests).
 
@@ -101,11 +116,23 @@ def _install_stubs() -> None:
         def _async_current_entries(self):
             return _zc_holder.entries
 
-        # t32 F2: the audited helper for uniqueness-less discovery — a
-        # recording no-op here; the flow's own host guard does the stacking
-        # prevention the acceptance pins.
-        def _async_handle_discovery_without_unique_id(self):
+        def _async_in_progress(self, include_uninitialized=False, **kwargs):
+            return _zc_holder.in_progress
+
+        # t37 D3: HA's own helper, modelled faithfully — a COROUTINE that
+        # dedups uniqueness-less discovery by raising. An un-awaited call
+        # therefore does nothing at all, which is exactly the production bug
+        # this replaces.
+        async def _async_handle_discovery_without_unique_id(self):
             _zc_holder.discovery_without_unique_id_calls += 1
+            if getattr(self, "_unique_id", None) is not None:
+                return
+            if self._async_current_entries():
+                raise DiscoveryDedupAbortError("already_configured")
+            await self.async_set_unique_id(_zc_holder.discovery_unique_id)
+            self._abort_if_unique_id_configured()
+            if self._async_in_progress(include_uninitialized=True):
+                raise DiscoveryDedupAbortError("already_in_progress")
 
     class OptionsFlow:
         def async_show_form(self, *, step_id, data_schema=None, errors=None):
@@ -140,7 +167,12 @@ def _install_stubs() -> None:
 
     # Shared holder so tests can swap the current-entry list through the
     # module object (the stub closure reads this same object).
-    _zc_holder = types.SimpleNamespace(entries=[], discovery_without_unique_id_calls=0)
+    _zc_holder = types.SimpleNamespace(
+        entries=[],
+        in_progress=[],
+        discovery_unique_id="default_discovery_unique_id",
+        discovery_without_unique_id_calls=0,
+    )
     sys.modules[__name__]._zc_holder = _zc_holder
 
     class AbortIfConfiguredError(Exception):
@@ -856,23 +888,27 @@ class TestMacLessDiscoveryF2:
             ip_address=host, host=host, properties=props or {}
         )
 
-    async def test_mac_less_records_the_helper(self, tmp_path):
+    def _holder(self):
+        return sys.modules[__name__]._zc_holder
+
+    async def test_mac_less_discovery_awaits_the_helper(self, tmp_path):
         module = _form_module()
         flow = module.NvidiaShieldConfigFlow()
         flow.hass = FakeHass(str(tmp_path))
-        sys.modules[__name__]._zc_holder.discovery_without_unique_id_calls = 0
+        self._holder().discovery_without_unique_id_calls = 0
 
         result = await flow.async_step_zeroconf(self._discovery())
 
         assert result["type"] == "form"  # prefill still offered once
-        assert (
-            sys.modules[__name__]._zc_holder.discovery_without_unique_id_calls
-            == 1
-        )
+        # t37 D3: only an AWAITED coroutine body runs — before the fix this
+        # counter stayed at 0 and Python warned "coroutine ... was never
+        # awaited", while HA's dedup never happened
+        assert self._holder().discovery_without_unique_id_calls == 1
+        # the helper claims HA's discovery sentinel for the flow
+        assert flow._unique_id == self._holder().discovery_unique_id
 
-    async def test_mac_less_rediscovery_does_not_stack(self, tmp_path):
-        """A second MAC-less discovery of an already-configured host
-        aborts — one-entry outcome."""
+    async def test_mac_less_discovery_with_entries_aborts(self, tmp_path):
+        """Real contract: any entry for the handler → already_configured."""
         module = _form_module()
         flow = module.NvidiaShieldConfigFlow()
         flow.hass = FakeHass(str(tmp_path))
@@ -882,13 +918,45 @@ class TestMacLessDiscoveryF2:
             options={CONF_PROTOCOL_HOST: "192.168.1.50"},
             title="Shield",
         )
-        sys.modules[__name__]._zc_holder.entries[:] = [entry]
+        self._holder().entries[:] = [entry]
 
-        result = await flow.async_step_zeroconf(self._discovery())
+        with pytest.raises(DiscoveryDedupAbortError) as raised:
+            await flow.async_step_zeroconf(self._discovery())
 
-        assert result == {"type": "abort", "reason": "already_configured"}
+        assert raised.value.reason == "already_configured"
         # the entries list is untouched: one-entry outcome
-        assert len(sys.modules[__name__]._zc_holder.entries) == 1
+        assert len(self._holder().entries) == 1
+
+    async def test_mac_less_discovery_while_another_flow_runs_aborts(self, tmp_path):
+        """The other half of the real contract: one flow at a time."""
+        module = _form_module()
+        flow = module.NvidiaShieldConfigFlow()
+        flow.hass = FakeHass(str(tmp_path))
+        self._holder().in_progress[:] = [{"flow_id": "other"}]
+
+        with pytest.raises(DiscoveryDedupAbortError) as raised:
+            await flow.async_step_zeroconf(self._discovery())
+
+        assert raised.value.reason == "already_in_progress"
+
+    async def test_mac_less_discovery_leaves_no_un_awaited_coroutine(
+        self, tmp_path
+    ):
+        """t37 D3: the helper must never be left un-awaited."""
+        module = _form_module()
+        flow = module.NvidiaShieldConfigFlow()
+        flow.hass = FakeHass(str(tmp_path))
+
+        with warnings.catch_warnings(record=True) as caught:
+            warnings.simplefilter("always")
+            await flow.async_step_zeroconf(self._discovery())
+            del flow
+            gc.collect()
+
+        never_awaited = [
+            str(w.message) for w in caught if "never awaited" in str(w.message)
+        ]
+        assert never_awaited == []
 
 
 class TestSeedBlockShapeF4: