fix(device-pool): scope primary key by host and thread through capability_tags

- pooled_devices primary key changed from device_id alone to
  (host_id, device_id), so two hosts reporting the same local
  device_id no longer crash sync_host_devices() with an uncaught
  sqlite3.IntegrityError.
- Device gained a capability_tags field so DevicePool.sync_host_devices()
  can actually populate PooledDevice.capability_tags from a real host
  sync instead of always falling back to an empty list.

openspec: device-pool capability, archived change cloud-runtime

cloud/pool.py

@@ -106,7 +106,7 @@ class DevicePool:
         synced_at: datetime,
     ) -> PooledDevice:
         raw_status = device.status if device.status in _HOST_REPORTED_STATUSES else "idle"
-        tags = list(getattr(device, "capability_tags", []) or [])
+        tags = list(device.capability_tags or [])
         return PooledDevice(
             device_id=device.id,
             host_id=host_id,

cloud/store.py

@@ -301,12 +301,13 @@ class CloudStore:
             connection.execute(
                 """
                 create table if not exists pooled_devices (
-                    device_id text primary key,
+                    device_id text not null,
                     host_id text not null,
                     driver_type text not null,
                     status text not null,
                     capability_tags_json text not null,
-                    synced_at text
+                    synced_at text,
+                    primary key (host_id, device_id)
                 )
                 """
             )

core/models.py

@@ -58,6 +58,7 @@ class Device:
     name: str | None = None
     driver_type: str = "wda"
     connection_info: dict[str, Any] = field(default_factory=dict)
+    capability_tags: list[str] = field(default_factory=list)
 
     def to_dict(self) -> dict[str, Any]:
         return {
@@ -66,6 +67,7 @@ class Device:
             "status": self.status,
             "driver_type": self.driver_type,
             "connection_info": dict(self.connection_info),
+            "capability_tags": list(self.capability_tags),
         }
 
 

device/manager.py

@@ -33,6 +33,7 @@ class DeviceManager:
         driver_type: str = "wda",
         connection_info: dict[str, object] | None = None,
         status: DeviceStatus = "idle",
+        capability_tags: list[str] | None = None,
     ) -> Device:
         with self._lock:
             device = Device(
@@ -41,6 +42,7 @@ class DeviceManager:
                 status=status,
                 driver_type=driver_type,
                 connection_info=dict(connection_info or {}),
+                capability_tags=list(capability_tags or []),
             )
             self._devices[device_id] = device
             self._factories[device_id] = driver_factory

tests/test_cloud_store.py

@@ -125,3 +125,40 @@ def test_capability_tags_round_trip(tmp_path) -> None:
     device = store.get_device("dev-1")
     assert device is not None
     assert device.capability_tags == ["ios", "physical"]
+
+
+def test_two_hosts_with_colliding_local_device_id_do_not_crash(tmp_path) -> None:
+    """Two hosts legitimately reporting the same local device_id string are
+    different devices in the pool; syncing both must not violate a uniqueness
+    constraint and must keep both rows independently addressable by host."""
+    store = CloudStore(tmp_path / "cloud.sqlite3")
+    from datetime import UTC, datetime
+
+    store.upsert_host("host-a", address="a", last_seen_at=datetime(2026, 1, 1, tzinfo=UTC))
+    store.upsert_host("host-b", address="b", last_seen_at=datetime(2026, 1, 1, tzinfo=UTC))
+
+    store.replace_host_devices("host-a", [_pooled("dev-1", "host-a", status="idle")])
+    # Same local device_id "dev-1", different host -- must not raise.
+    store.replace_host_devices("host-b", [_pooled("dev-1", "host-b", status="busy")])
+
+    devices = store.list_devices()
+    assert len(devices) == 2
+    by_host = {d.host_id: d for d in devices}
+    assert set(by_host) == {"host-a", "host-b"}
+    assert by_host["host-a"].device_id == "dev-1"
+    assert by_host["host-b"].device_id == "dev-1"
+    assert by_host["host-a"].status == "idle"
+    assert by_host["host-b"].status == "busy"
+
+    # Lookup by id alone must still work (finds a device on some host) without crashing.
+    found = store.get_device("dev-1")
+    assert found is not None
+    assert found.host_id in {"host-a", "host-b"}
+
+    # Re-syncing host-a alone must not disturb host-b's colliding-id row.
+    store.replace_host_devices("host-a", [_pooled("dev-1", "host-a", status="error")])
+    devices = store.list_devices()
+    assert len(devices) == 2
+    by_host = {d.host_id: d for d in devices}
+    assert by_host["host-a"].status == "error"
+    assert by_host["host-b"].status == "busy"

tests/test_console_api.py

@@ -72,6 +72,7 @@ def test_console_status_endpoints_cover_empty_and_populated_states(tmp_path) ->
             "status": "idle",
             "driver_type": "wda",
             "connection_info": {},
+            "capability_tags": [],
         }
     ]
     assert [task["id"] for task in client.get("/console/tasks").json()] == [

tests/test_device_pool.py

@@ -23,8 +23,19 @@ def _config(**overrides) -> CloudConfig:
     return CloudConfig(**base)
 
 
-def _device(device_id: str, *, status: str = "idle", driver_type: str = "wda") -> Device:
-    return Device(id=device_id, status=status, driver_type=driver_type)  # type: ignore[arg-type]
+def _device(
+    device_id: str,
+    *,
+    status: str = "idle",
+    driver_type: str = "wda",
+    capability_tags: list[str] | None = None,
+) -> Device:
+    return Device(
+        id=device_id,
+        status=status,  # type: ignore[arg-type]
+        driver_type=driver_type,
+        capability_tags=list(capability_tags or []),
+    )
 
 
 def test_new_host_sync_creates_registration_and_devices(tmp_path) -> None:
@@ -140,3 +151,37 @@ def test_two_hosts_aggregate_into_one_listing(tmp_path) -> None:
         "b-dev-1": "host-b",
         "b-dev-2": "host-b",
     }
+
+
+def test_two_hosts_syncing_same_local_device_id_do_not_crash(tmp_path) -> None:
+    """Two hosts legitimately reporting the same local device_id are distinct
+    devices in the pool; syncing both must succeed and yield two entries."""
+    pool = DevicePool(CloudStore(tmp_path / "cloud.sqlite3"), _config())
+    pool.sync_host_devices("host-a", [_device("dev-1", status="idle")], address="a")
+    pool.sync_host_devices("host-b", [_device("dev-1", status="busy")], address="b")
+
+    devices = pool.list_devices()
+    assert len(devices) == 2
+    by_host = {d.host_id: d for d in devices}
+    assert set(by_host) == {"host-a", "host-b"}
+    assert by_host["host-a"].device_id == "dev-1"
+    assert by_host["host-b"].device_id == "dev-1"
+    assert by_host["host-a"].status == "idle"
+    assert by_host["host-b"].status == "busy"
+
+
+def test_capability_tags_flow_from_device_to_pooled_device(tmp_path) -> None:
+    pool = DevicePool(CloudStore(tmp_path / "cloud.sqlite3"), _config())
+    pool.sync_host_devices(
+        "host-a",
+        [_device("dev-1", capability_tags=["ios", "physical"])],
+        address="a",
+    )
+
+    devices = pool.list_devices()
+    assert len(devices) == 1
+    assert devices[0].capability_tags == ["ios", "physical"]
+
+    fetched = pool.get_device("dev-1")
+    assert fetched is not None
+    assert fetched.capability_tags == ["ios", "physical"]