Unify K1 discovery ownership and verification across enrollment paths

src/k1link/device_plugins/xgrids_k1/ble/ap_activation.py

@@ -23,6 +23,7 @@ from k1link.device_plugins.xgrids_k1.ble.scanner import (
     discovered_device_selection,
     mark_captured_device_gatt_validated,
     retrieve_connected_device_capture,
+    selected_device_discovery,
 )
 from k1link.device_plugins.xgrids_k1.ble.wifi_provisioning import (
     AP_FALLBACK_IPV4,
@@ -157,6 +158,7 @@ async def _device_ap_activation_session_impl(
     max_without_response: int | None = None
     active_captured_device: CapturedDiscoveredDevice | None = None
     gatt_baseline_validated = False
+    connect_attempted = False
     progress.operation_stage = operation_stage
 
     try:
@@ -228,17 +230,22 @@ async def _device_ap_activation_session_impl(
             raise
 
         async with AsyncExitStack() as client_stack:
-            operation_stage = "connect"
-            progress.operation_stage = operation_stage
             try:
-                client = await client_stack.enter_async_context(
-                    BleakClient(device, timeout=connect_timeout_seconds, pair=False)
-                )
-            except Exception as exc:
-                if active_captured_device is not None:
-                    demote_connected_device_handle_after_gatt_failure(
-                        active_captured_device
+                async with selected_device_discovery(
+                    device, timeout_seconds=connect_timeout_seconds,
+                ):
+                    if (active_captured_device is not None
+                            and captured_device_handle(active_captured_device) is not device):
+                        raise BleakDeviceNotFoundError(
+                            device_macos_uuid, "BLE selection invalidated",
+                        )
+                    operation_stage = "connect"
+                    progress.operation_stage = operation_stage
+                    connect_attempted = True
+                    client = await client_stack.enter_async_context(
+                        BleakClient(device, timeout=connect_timeout_seconds, pair=False)
                     )
+            except Exception as exc:
                 _annotate_ble_operation_error(
                     exc,
                     operation_stage=operation_stage,
@@ -411,10 +418,6 @@ async def _device_ap_activation_session_impl(
                         "outcome": _outcome(baseline, observations, disconnected),
                     }
             except Exception as exc:
-                if active_captured_device is not None and not gatt_baseline_validated:
-                    demote_connected_device_handle_after_gatt_failure(
-                        active_captured_device
-                    )
                 _annotate_ble_operation_error(
                     exc,
                     operation_stage=operation_stage,
@@ -438,7 +441,7 @@ async def _device_ap_activation_session_impl(
         # when that bypasses the normal exception annotator, a captured
         # recovery object that never passed baseline GATT validation must be
         # discarded.
-        if active_captured_device is not None and not gatt_baseline_validated:
+        if connect_attempted and active_captured_device is not None and not gatt_baseline_validated:
             demote_connected_device_handle_after_gatt_failure(active_captured_device)
         frame[:] = b"\x00" * len(frame)
 

src/k1link/device_plugins/xgrids_k1/ble/scanner.py

@@ -17,7 +17,7 @@ from uuid import UUID
 from bleak import BleakScanner
 from bleak.backends.device import BLEDevice
 from bleak.backends.scanner import AdvertisementData
-from bleak.exc import BleakDeviceNotFoundError
+from bleak.exc import BleakDBusError, BleakDeviceNotFoundError
 
 from k1link.artifacts import utc_now_iso
 from k1link.device_plugins.xgrids_k1.ble.runtime_arbiter import (
@@ -840,47 +840,21 @@ async def _retrieve_bluez_device(address: str, details: object = None) -> BLEDev
 
 
 @asynccontextmanager
-async def hold_bluez_device_for_connect(device: BLEDevice):
-    """Keep discovery alive across the cache-check / D-Bus-connect gap.
+async def selected_device_discovery(device: BLEDevice, *, timeout_seconds: float):
+    """One BlueZ discovery session from exact-path resolution through Connect.
 
-    BlueZ can discard an unconnected path after StopDiscovery. Holding one
-    scanner reference until BleakClient connects avoids consuming a path just
-    removed by that cleanup. This neither selects another device nor repeats
-    Connect/GATT writes, and is released before any characteristic is read.
+    All Bleak scanners on this owner loop share a D-Bus client. BlueZ permits
+    one discovery session per client/adapter, so resolution must consume this
+    scanner's observations, never start a nested find_device_by_address scan.
+    The original owner capture remains authoritative; discovery only restores
+    its exact address/path and stays alive until the consuming client connects.
+    CoreBluetooth already carries the native object and needs no extra scan.
     """
     details = getattr(device, "details", None)
     if not sys.platform.startswith("linux") or not isinstance(details, dict):
         yield
         return
     path = details.get("path", "")
-    match = re.fullmatch(r"/org/bluez/(hci[0-9]+)/dev_[0-9A-F_]+", path)
-    if (match is None
-            or not re.fullmatch(r"(?:[0-9a-fA-F]{2}:){5}[0-9a-fA-F]{2}", device.address)
-            or path.rsplit("/", 1)[-1] != "dev_" + device.address.upper().replace(":", "_")):
-        raise BleakDeviceNotFoundError(device.address, "Invalid selected BlueZ transport")
-    runtime = ble_runtime_snapshot()
-    owner = ble_runtime_owner_epoch_for_current_loop()
-    if (owner is None or runtime["owner_epoch"] != owner
-            or runtime["poisoned"] or not runtime["owner_loop_bound"]
-            or runtime["active_operation_kind"] not in {"status-read", "wifi-provision"}):
-        raise BleakDeviceNotFoundError(device.address, "BLE operation owner changed")
-    async with BleakScanner(bluez={"adapter": match[1]}):
-        yield
-
-
-async def ensure_device_for_gatt(device: BLEDevice, *, timeout_seconds: float) -> None:
-    """Restore a vanished BlueZ path before the one admitted GATT connection.
-
-    A BLEDevice stores a D-Bus path, not a native object lease. After the form
-    has been filled in, BlueZ may have removed that path. Observe the same
-    address on the same adapter once, then require that exact path to exist.
-    Keep the original selection/capture; no session pin, public scan generation,
-    GATT connection or write is created here. CoreBluetooth needs no refresh.
-    """
-    details = getattr(device, "details", None)
-    if not sys.platform.startswith("linux") or not isinstance(details, dict):
-        return
-    path = details.get("path", "")
     address = device.address
     match = re.fullmatch(r"/org/bluez/(hci[0-9]+)/dev_[0-9A-F_]+", path)
     if (match is None
@@ -898,30 +872,45 @@ async def ensure_device_for_gatt(device: BLEDevice, *, timeout_seconds: float) -
                 or ble_runtime_owner_epoch_for_current_loop() != owner_epoch
                 or runtime["owner_epoch"] != owner_epoch
                 or not runtime["owner_loop_bound"] or runtime["poisoned"]
-                or operation_kind not in {"status-read", "wifi-provision"}
+                or operation_kind not in {"status-read", "wifi-provision", "ap-enable"}
                 or runtime["active_operation_kind"] != operation_kind):
             raise BleakDeviceNotFoundError(address, "BLE operation owner changed")
 
-    require_owner()
-    current = await _retrieve_bluez_device(address, details)
-    require_owner()
-    if current is not None:
-        return
+    observed = asyncio.Event()
+
+    def observe(candidate: BLEDevice, _advertisement: AdvertisementData) -> None:
+        if (candidate.address.casefold() == address.casefold()
+                and isinstance(candidate.details, dict)
+                and candidate.details.get("path") == path):
+            observed.set()
 
-    candidate = await BleakScanner.find_device_by_address(
-        address,
-        timeout=min(timeout_seconds, 8.0),
-        bluez={"adapter": match[1]},
-    )
     require_owner()
-    if (candidate is None or candidate.address.casefold() != address.casefold()
-            or not isinstance(candidate.details, dict)
-            or candidate.details.get("path") != path):
-        raise BleakDeviceNotFoundError(address, "Selected BlueZ transport unavailable")
-    current = await _retrieve_bluez_device(address, details)
-    require_owner()
-    if current is None:
-        raise BleakDeviceNotFoundError(address, "Selected BlueZ transport disappeared")
+    admitted = False
+    try:
+        async with BleakScanner(
+            detection_callback=observe, bluez={"adapter": match[1]},
+        ):
+            require_owner()
+            if await _retrieve_bluez_device(address, details) is None:
+                await asyncio.wait_for(observed.wait(), timeout=min(timeout_seconds, 8.0))
+                require_owner()
+                if await _retrieve_bluez_device(address, details) is None:
+                    raise BleakDeviceNotFoundError(address, "Selected BlueZ transport disappeared")
+            require_owner()
+            admitted = True
+            yield
+    except Exception as exc:
+        # Describe only resolution failures here. Connect/GATT failures keep
+        # their own stage, and exception text (possibly a payload) stays private.
+        if not admitted:
+            if isinstance(exc, BleakDBusError):
+                exc.reason_code = {
+                    "org.bluez.Error.InProgress": "ble-discovery-busy",
+                    "org.bluez.Error.NotReady": "ble-adapter-unavailable",
+                }.get(exc.dbus_error, "ble-discovery-failed")
+            elif isinstance(exc, (TimeoutError, BleakDeviceNotFoundError)):
+                exc.reason_code = "ble-selected-device-unavailable"
+        raise
 
 
 async def _retrieve_corebluetooth_device(

src/k1link/device_plugins/xgrids_k1/ble/wifi_provisioning.py

@@ -24,11 +24,10 @@ from k1link.device_plugins.xgrids_k1.ble.scanner import (
     demote_connected_device_handle_after_gatt_failure,
     discover_known_device_capture_for_status_read,
     discovered_device_selection,
-    ensure_device_for_gatt,
-    hold_bluez_device_for_connect,
     mark_captured_device_gatt_validated,
     retrieve_connected_device_capture,
     retrieve_known_device_capture_for_status_read,
+    selected_device_discovery,
 )
 
 PROFILE_ID = "xgrids-k1-fw3-wifi-v1"
@@ -308,6 +307,7 @@ async def _read_wifi_status_impl(
 ) -> WifiStatusReadResult:
     active_captured_device: CapturedDiscoveredDevice | None = None
     gatt_baseline_validated = False
+    connect_attempted = False
     try:
         progress.operation_stage = "resolution"
         if captured_device is not None:
@@ -394,12 +394,12 @@ async def _read_wifi_status_impl(
             )
 
         async with AsyncExitStack() as gatt_session:
-            async with hold_bluez_device_for_connect(device):
-                await ensure_device_for_gatt(device, timeout_seconds=timeout_seconds)
+            async with selected_device_discovery(device, timeout_seconds=timeout_seconds):
                 if (active_captured_device is not None
                         and captured_device_handle(active_captured_device) is not device):
                     raise BleakDeviceNotFoundError(device_macos_uuid, "BLE selection invalidated")
                 progress.operation_stage = "connect"
+                connect_attempted = True
                 client = await gatt_session.enter_async_context(
                     BleakClient(device, timeout=timeout_seconds, pair=False)
                 )
@@ -478,7 +478,7 @@ async def _read_wifi_status_impl(
         # Connection/contract/read failure revokes only this transport lease;
         # the process-scoped UUID/session token remains available for another
         # explicit CoreBluetooth retrieval attempt.
-        if active_captured_device is not None and not gatt_baseline_validated:
+        if connect_attempted and active_captured_device is not None and not gatt_baseline_validated:
             demote_connected_device_handle_after_gatt_failure(active_captured_device)
 
 
@@ -566,6 +566,7 @@ async def _provision_wifi_impl(
     max_without_response: int | None = None
     active_captured_device: CapturedDiscoveredDevice | None = None
     gatt_baseline_validated = False
+    connect_attempted = False
 
     try:
         progress.operation_stage = operation_stage
@@ -626,13 +627,13 @@ async def _provision_wifi_impl(
             )
 
         async with AsyncExitStack() as gatt_session:
-            async with hold_bluez_device_for_connect(device):
-                await ensure_device_for_gatt(device, timeout_seconds=timeout_seconds)
+            async with selected_device_discovery(device, timeout_seconds=timeout_seconds):
                 if (active_captured_device is not None
                         and captured_device_handle(active_captured_device) is not device):
                     raise BleakDeviceNotFoundError(device_macos_uuid, "BLE selection invalidated")
                 operation_stage = "connect"
                 progress.operation_stage = operation_stage
+                connect_attempted = True
                 client = await gatt_session.enter_async_context(
                     BleakClient(device, timeout=timeout_seconds, pair=False)
                 )
@@ -767,8 +768,6 @@ async def _provision_wifi_impl(
                 "outcome": _outcome(baseline, observations, disconnected),
             }
     except Exception as exc:
-        if active_captured_device is not None and not gatt_baseline_validated:
-            demote_connected_device_handle_after_gatt_failure(active_captured_device)
         _annotate_ble_operation_error(
             exc,
             operation_stage=operation_stage,
@@ -781,12 +780,14 @@ async def _provision_wifi_impl(
         )
         raise
     finally:
+        # Only an attempted Connect/GATT exchange can revoke this capture.
+        # Discovery failures have not tested the selected transport.
         # Hard-timeout cancellation may bypass ``except Exception``.  The
         # failed live transport lease must still be demoted when the exact
         # captured object never completed the reviewed baseline read; the
         # UUID/session recovery token itself remains available for a later
         # explicit attempt.
-        if active_captured_device is not None and not gatt_baseline_validated:
+        if connect_attempted and active_captured_device is not None and not gatt_baseline_validated:
             demote_connected_device_handle_after_gatt_failure(active_captured_device)
         frame[:] = b"\x00" * len(frame)
 

src/k1link/device_plugins/xgrids_k1/node_bridge.py

@@ -38,6 +38,35 @@ ATTESTATION = {
 }
 
 
+def verification_parameters(state, operation_id, *, requested_device_id=None):
+    """Use the admitted persisted Bridge target before asking for BLE again.
+
+    The facade still checks its ledger, exact identity, route, DeviceInfo and
+    physical reconciliation fences. A saved address alone never grants START.
+    """
+    parameters = {
+        "expected_snapshot_runtime_id": state["snapshot_runtime_id"],
+        "operation_id": operation_id,
+    }
+    policy = (state.get("connection_policy") or {}).get("actions", {})
+    decision = policy.get("observe-configured-device-network") or {}
+    target = decision.get("required_transport_ref")
+    selected = state.get("selected_device_id")
+    if (decision.get("allowed") is True
+            and decision.get("requires_live_gatt_validation") is False
+            and decision.get("required_connection_mode") == "bridge"
+            and isinstance(target, str) and isinstance(selected, str)
+            and target.casefold() == selected.casefold()
+            and (requested_device_id is None
+                 or target.casefold() == requested_device_id.casefold())):
+        parameters.update(
+            device_id=target, source="durable-configured-state",
+            compatibility_attestation=ATTESTATION,
+            expected_mode_revision=state.get("desired_connection_mode_revision"),
+        )
+    return parameters
+
+
 def failure_locations(error: BaseException) -> str:
     """Bounded causal stack, excluding messages, source lines and frame locals.
 
@@ -61,6 +90,37 @@ def failure_locations(error: BaseException) -> str:
     return " <- ".join(chain)
 
 
+def failure_transport_facts(error: BaseException) -> str:
+    """Journal finite native codes and dispatch facts, never error text/payloads."""
+    facts: dict[str, object] = {}
+    current: BaseException | None = error
+    seen: set[int] = set()
+    while current is not None and id(current) not in seen and len(seen) < 6:
+        seen.add(id(current))
+        stage = getattr(current, "operation_stage", None)
+        if isinstance(stage, str) and stage in {
+            "resolution", "exact-uuid-scan", "connect", "gatt-contract",
+            "baseline-read", "gatt-write", "status-poll", "status-read",
+        }:
+            facts["stage"] = stage
+        bluez = getattr(current, "dbus_error", None)
+        if isinstance(bluez, str) and bluez in {
+            "org.bluez.Error.InProgress", "org.bluez.Error.NotReady",
+            "org.bluez.Error.Failed", "org.bluez.Error.NotAuthorized",
+            "org.bluez.Error.NotSupported", "org.bluez.Error.DoesNotExist",
+            "org.freedesktop.DBus.Error.UnknownObject",
+        }:
+            facts["bluez"] = bluez
+        for field in ("device_write_attempted", "device_write_confirmed"):
+            value = getattr(current, field, None)
+            if isinstance(value, bool):
+                facts[field] = value
+        current = current.__cause__ or (
+            None if current.__suppress_context__ else current.__context__
+        )
+    return " ".join(f"{key}={value}" for key, value in facts.items()) or "unavailable"
+
+
 class NodeEnrollmentRejected(ValueError):
     """A host admission failure proven to precede any device invocation."""
 
@@ -186,12 +246,15 @@ class NodeBridge:
         except Exception as error:
             # Journalled device failures are returned as normal operation
             # results, so the HTTP exception logger never sees them. Record
-            # only source locations here; exception text may contain a frame.
+            # only source locations and finite transport facts; exception text
+            # may contain a frame.
             logging.getLogger(__name__).warning(
-                "K1 journalled invocation failed: action=%s operation=%s exception=%s chain=%s",
+                "K1 journalled invocation failed: action=%s operation=%s "
+                "exception=%s facts=%s chain=%s",
                 action,
                 identifier,
                 type(error).__name__,
+                failure_transport_facts(error),
                 failure_locations(error),
             )
             # Read the exact result after a failure; never repeat an invocation
@@ -234,7 +297,8 @@ class NodeBridge:
         async with self.lock:
             if datetime.fromisoformat(command["deadline_at"]) <= datetime.now(UTC):
                 raise NodeEnrollmentRejected("command-expired")
-            state = await self.state()
+            snapshot = await self.invoke("state.read", {}, identifier + "-admission")
+            state = self.project(snapshot)
             if command.get("runtime_id") != state["runtime_id"]:
                 raise NodeEnrollmentRejected("runtime-changed")
             if action == "networks":
@@ -274,6 +338,10 @@ class NodeBridge:
                         ssid=parameters.get("ssid"),
                         password=parameters.get("password"),
                     )
+                else:
+                    payload.update(verification_parameters(
+                        snapshot, plugin_identifier, requested_device_id=parameters["device_id"],
+                    ))
                 try:
                     result = await self.invoke_journalled(
                         "network.provision" if action == "connect" else "connection.verify",

src/k1link/device_plugins/xgrids_k1/node_sensor.py

@@ -12,7 +12,7 @@ from .facade import (
     ViewerSettingsRequest,
     normalize_project_name,
 )
-from .node_bridge import ATTESTATION, plugin_operation_id
+from .node_bridge import ATTESTATION, plugin_operation_id, verification_parameters
 
 PHYSICAL_ACCEPTANCE_KEYS = (
     "operator_present",
@@ -107,33 +107,6 @@ def project_sensor(snapshot, node_id):
     }
 
 
-def verification_parameters(state, operation_id):
-    """Use the admitted persisted Bridge target before asking for BLE again.
-
-    The facade still checks its ledger, exact identity, route, DeviceInfo and
-    physical reconciliation fences. A saved address alone never grants START.
-    """
-    parameters = {
-        "expected_snapshot_runtime_id": state["snapshot_runtime_id"],
-        "operation_id": operation_id,
-    }
-    policy = (state.get("connection_policy") or {}).get("actions", {})
-    decision = policy.get("observe-configured-device-network") or {}
-    target = decision.get("required_transport_ref")
-    selected = state.get("selected_device_id")
-    if (decision.get("allowed") is True
-            and decision.get("requires_live_gatt_validation") is False
-            and decision.get("required_connection_mode") == "bridge"
-            and isinstance(target, str) and isinstance(selected, str)
-            and target.casefold() == selected.casefold()):
-        parameters.update(
-            device_id=target, source="durable-configured-state",
-            compatibility_attestation=ATTESTATION,
-            expected_mode_revision=state.get("desired_connection_mode_revision"),
-        )
-    return parameters
-
-
 class NodeK1Sensor:
     def __init__(self, bridge, peers):
         self.bridge, self.peers = bridge, peers