From 33c4a0d112f0740f98e3bdb9b1cb7627ec60d410 Mon Sep 17 00:00:00 2001 From: somew1nd Date: Tue, 1 Sep 2026 11:50:14 -0700 Subject: [PATCH] refactor(device): remove single-device state and test branches --- apps/admin_console/routers/system.py | 6 +- apps/admin_console/routers/tasks.py | 9 ++- .../services/task_queue_service.py | 40 ++++++------- artemis/core/diagnostics/emulator_manager.py | 7 ++- artemis/core/diagnostics/engine.py | 24 +++----- artemis/runtime/awake_service.py | 17 ++++-- artemis/runtime/device_pool.py | 34 +++++++---- .../test_task_device_readiness.py | 8 +-- .../admin_console/test_task_queue_service.py | 58 ++++++++++++++----- tests/unit/core/test_diagnostics.py | 10 ++-- tests/unit/runtime/test_awake_service.py | 23 +++++++- tests/unit/runtime/test_device_pool.py | 27 +++++++++ 12 files changed, 175 insertions(+), 88 deletions(-) diff --git a/apps/admin_console/routers/system.py b/apps/admin_console/routers/system.py index 6345ea5..a2370ea 100644 --- a/apps/admin_console/routers/system.py +++ b/apps/admin_console/routers/system.py @@ -111,7 +111,7 @@ async def select_active_device(request: SelectDeviceRequest): if not serial: raise HTTPException(status_code=400, detail="Device serial cannot be empty.") - readiness_engine.set_active_device_serial(serial) + readiness_engine.set_probe_target_serial(serial) # Return updated readiness report = await readiness_engine.run_all(force_refresh=True) return { @@ -192,7 +192,7 @@ async def connect_adb_server(payload: ConnectAdbServerRequest, request: Request) response: dict[str, object] = {"connection_result": connection_result} if connection_result["success"]: - readiness_engine.set_active_device_serial(None) + readiness_engine.set_probe_target_serial(None) readiness_engine.invalidate_cache() response["report"] = await readiness_engine.run_all(force_refresh=True) return response @@ -214,7 +214,7 @@ async def use_local_adb_server(request: Request, persist: bool = True): """Restore the standard local ADB server without touching a remote daemon.""" _require_local_admin_request(request) connection_result = await adb_server_connection.use_local_server(persist=persist) - readiness_engine.set_active_device_serial(None) + readiness_engine.set_probe_target_serial(None) readiness_engine.invalidate_cache() updated_report = await readiness_engine.run_all(force_refresh=True) return { diff --git a/apps/admin_console/routers/tasks.py b/apps/admin_console/routers/tasks.py index cc66ac4..fd18f35 100644 --- a/apps/admin_console/routers/tasks.py +++ b/apps/admin_console/routers/tasks.py @@ -140,7 +140,10 @@ async def run_task(request: RunRequest): # polling intervals cannot start through a stale Ready state. Use the # bounded submission probe: the full diagnostics path also scans packages, # emulator installations, Android version, and screen size. - target_serial = request.device_serial or readiness_engine.get_active_device_serial() + # With no explicit serial the probe itself resolves a live target (it + # prefers the diagnostics target preference, then any unlocked ready + # device); the verified serial is bound below. + target_serial = request.device_serial device_probe = await readiness_engine.run_device_submission_probe( target_serial=target_serial ) @@ -164,10 +167,6 @@ async def run_task(request: RunRequest): # invalid, enqueue_tasks rejects the submission with a clear error. if verified_serial and not request.device_serial: target_serial = verified_serial - # The probe verified a live device either way: keep the global active - # serial in sync even when the caller pinned the target explicitly. - if verified_serial and readiness_engine.get_active_device_serial() != verified_serial: - readiness_engine.set_active_device_serial(verified_serial) return await task_queue_service.enqueue_tasks( incoming_goals, diff --git a/apps/admin_console/services/task_queue_service.py b/apps/admin_console/services/task_queue_service.py index 0737e3e..4941a9a 100644 --- a/apps/admin_console/services/task_queue_service.py +++ b/apps/admin_console/services/task_queue_service.py @@ -767,13 +767,6 @@ class TaskQueueService: for i, goal in enumerate(goals): sess_id = (session_id if (session_id and len(goals) == 1) else str(uuid.uuid4())) assigned_serial = device_serial - if not assigned_serial: - try: - from artemis.core.diagnostics import readiness_engine - - assigned_serial = readiness_engine.get_active_device_serial() - except Exception: - assigned_serial = None if not assigned_serial: try: from artemis.runtime import device_pool @@ -949,21 +942,18 @@ class TaskQueueService: owner = dev_owner break if owner is None: + # Covers scoped/legacy lock records that get_active_owners cannot + # key by session or device. A stale owner for a different session + # is never adopted: with no live owner, a session-targeted stop + # falls through to the queue/session-repository fallbacks below. fallback_owner = DeviceExecutionLock.get_active_owner(target_device) - if fallback_owner: - if not target_sid or ( - fallback_owner.session_id and str(fallback_owner.session_id) == target_sid - ): - owner = fallback_owner - elif not active_owners: - # In unit tests, get_active_owners is unmocked (empty), while get_active_owner is mocked - owner = fallback_owner + if fallback_owner and ( + not target_sid + or (fallback_owner.session_id and str(fallback_owner.session_id) == target_sid) + ): + owner = fallback_owner - owner_record_exists = ( - DeviceExecutionLock.has_owner_record(target_device) - if hasattr(DeviceExecutionLock, "has_owner_record") - else False - ) + owner_record_exists = DeviceExecutionLock.has_owner_record(target_device) owner_pid = owner.pid if owner else None # Resolve the locally-managed run for this target (concurrent scheduling @@ -981,8 +971,16 @@ class TaskQueueService: None, ) elif len(state.active_runs) == 1: + # Untargeted stop (legacy single-device gesture): with exactly one + # live run the target is unambiguous. With several runs no run is + # resolved -- callers must target a session or device explicitly. local_run = next(iter(state.active_runs.values())) - local_proc = (local_run or {}).get("process") or state.current_process + local_proc = (local_run or {}).get("process") + if local_proc is None and not state.active_runs: + # Legacy fallback for the window without per-run bookkeeping. With + # live active_runs, current_process belongs to the newest run and + # must not stand in for an unresolved target. + local_proc = state.current_process local_pid = getattr(local_proc, "pid", None) is_local_owner = bool(owner_pid and local_pid and owner_pid == local_pid) diff --git a/artemis/core/diagnostics/emulator_manager.py b/artemis/core/diagnostics/emulator_manager.py index fc53092..2a54f4d 100644 --- a/artemis/core/diagnostics/emulator_manager.py +++ b/artemis/core/diagnostics/emulator_manager.py @@ -404,9 +404,12 @@ class EmulatorManager: try: from artemis.core.diagnostics.engine import readiness_engine - readiness_engine.set_active_device_serial(detected_serial) + # Focus the readiness report on the freshly booted + # emulator; task routing still resolves its own + # target from the request or the device pool. + readiness_engine.set_probe_target_serial(detected_serial) except Exception as e: - logger.debug(f"Failed to auto-select device serial: {e}") + logger.debug(f"Failed to focus probes on the new emulator: {e}") return else: # Gradual progress visual indicator diff --git a/artemis/core/diagnostics/engine.py b/artemis/core/diagnostics/engine.py index 2153ff1..37b010c 100644 --- a/artemis/core/diagnostics/engine.py +++ b/artemis/core/diagnostics/engine.py @@ -52,7 +52,6 @@ class ReadinessEngine: def __init__(self): self._probes: dict[str, BaseProbe] = {} - self._active_device_serial: str | None = None self._python_probe = PythonRuntimeProbe() self._config_probe = SystemConfigProbe() self._toolchain_probe = ToolchainProbe() @@ -81,24 +80,19 @@ class ReadinessEngine: """Remove a diagnostic probe.""" self._probes.pop(probe_id, None) - def set_active_device_serial(self, serial: str | None) -> None: - """Set user-selected active target device serial.""" - self._active_device_serial = serial + def set_probe_target_serial(self, serial: str | None) -> None: + """Set the diagnostics probes' preferred device serial. + + This is a probe/report preference only (which device the readiness + report highlights and verifies first). Task routing never reads it + back: execution targets come from explicit request serials or the + device pool. Pass None to clear the preference, e.g. after switching + ADB server endpoints. + """ self.invalidate_cache() if hasattr(self._adb_probe, "set_target_serial"): self._adb_probe.set_target_serial(serial) - def get_active_device_serial(self) -> str | None: - """Get currently selected active device serial.""" - if self._report_cache and self._report_cache.active_device: - if self._report_cache.active_device.is_locked is False: - return self._report_cache.active_device.serial - if self._active_device_serial: - return self._active_device_serial - if self._report_cache and self._report_cache.active_device: - return self._report_cache.active_device.serial - return None - async def run_probe(self, probe_id: str) -> ProbeResult | None: """Execute a single specific probe by ID.""" probe = self._probes.get(probe_id) diff --git a/artemis/runtime/awake_service.py b/artemis/runtime/awake_service.py index 0d87149..7c9c864 100644 --- a/artemis/runtime/awake_service.py +++ b/artemis/runtime/awake_service.py @@ -150,7 +150,7 @@ def _configure_usb_stay_awake(device_id: str) -> str | None: def _discover_connected_device_ids() -> list[str]: - """Return the selected or all connected local Android devices.""" + """Return the explicitly targeted device, or the connected devices Artemis manages.""" if not _awake_enabled() or os.environ.get("ARTEMIS_CLOUD_MODE") == "1": return [] @@ -171,9 +171,18 @@ def _discover_connected_device_ids() -> list[str]: if target: return [target] if target in serials else [] - # Artemis has a single-device execution policy. Do not modify unrelated - # devices that happen to share the ADB server. - return serials[:1] + # Keep awake only the devices Artemis actually manages: serials claimed in + # the device pool by an active execution lock or a live queue reservation. + # Other devices sharing the same ADB server belong to unrelated tools or + # users, and their power policy must not be touched. + try: + from artemis.runtime.device_pool import device_pool + + claimed = device_pool.get_claimed_serials() + except Exception as exc: + logger.debug(f"Could not query claimed devices for the awake policy: {exc}") + return [] + return [serial for serial in serials if serial in claimed] class ScreenAwakeService: diff --git a/artemis/runtime/device_pool.py b/artemis/runtime/device_pool.py index 10dc17c..a861d1c 100644 --- a/artemis/runtime/device_pool.py +++ b/artemis/runtime/device_pool.py @@ -398,6 +398,30 @@ class DevicePool: """Return ready devices that are not currently holding an execution lock.""" return [d for d in self.list_devices() if d.state == "device" and not d.is_busy] + def get_claimed_serials(self) -> set[str]: + """Return device serials Artemis currently claims on the ADB server. + + A device is claimed while any Artemis process holds its execution lock + or has a live pending queue reservation targeting it. Placeholder + reservations not yet bound to a concrete device ("pending"/"any") + claim nothing. Read-only: computed purely from cross-process lock and + queue metadata, without any adb traffic. + """ + claimed: set[str] = set() + try: + for owner in DeviceExecutionLock.get_active_owners().values(): + if owner.device_id: + claimed.add(str(owner.device_id)) + for item in DeviceExecutionLock.get_queued_tasks(): + serial = item.get("device_serial") + if serial: + claimed.add(str(serial)) + except Exception as exc: + logger.debug(f"Could not compute claimed device serials: {exc}") + claimed.discard("pending") + claimed.discard("any") + return claimed + def select_device(self, preferred_serial: str | None = None) -> str | None: """Select a target device serial for task execution. @@ -419,16 +443,6 @@ class DevicePool: # If not detected by ADB, still return it so ADB can attempt connection return preferred_serial - # Prefer active ready device from readiness engine if available - try: - from artemis.core.diagnostics import readiness_engine - - active_serial = readiness_engine.get_active_device_serial() - if active_serial and any(d.serial == active_serial and d.state == "device" for d in all_devs): - return active_serial - except Exception: - pass - # Pick first idle device for dev in ready_devs: if not dev.is_busy: diff --git a/tests/unit/admin_console/test_task_device_readiness.py b/tests/unit/admin_console/test_task_device_readiness.py index 74b2b46..9b34f2f 100644 --- a/tests/unit/admin_console/test_task_device_readiness.py +++ b/tests/unit/admin_console/test_task_device_readiness.py @@ -78,8 +78,8 @@ async def test_run_task_enqueues_when_device_is_unlocked(monkeypatch): @pytest.mark.asyncio -async def test_run_task_switches_to_unlocked_device_and_syncs_engine(monkeypatch): - """When default device is locked, run_task should enqueue on verified fallback and update engine.""" +async def test_run_task_binds_probe_verified_device_when_no_serial_requested(monkeypatch): + """Without an explicit serial, run_task enqueues on the probe-verified live device.""" unlocked_probe = ProbeResult( id="android_adb", category=ProbeCategory.DEVICE, @@ -100,9 +100,6 @@ async def test_run_task_switches_to_unlocked_device_and_syncs_engine(monkeypatch tasks.readiness_engine, "run_device_submission_probe", run_probe ) monkeypatch.setattr(tasks.task_queue_service, "enqueue_tasks", enqueue_tasks) - monkeypatch.setattr(tasks.readiness_engine, "get_active_device_serial", lambda: "device-locked-1") - set_active_mock = AsyncMock() - monkeypatch.setattr(tasks.readiness_engine, "set_active_device_serial", set_active_mock) result = await tasks.run_task(RunRequest(goal="Run on any ready device")) @@ -111,7 +108,6 @@ async def test_run_task_switches_to_unlocked_device_and_syncs_engine(monkeypatch # Check that device_serial passed to enqueue_tasks is the verified unlocked device _, kwargs = enqueue_tasks.call_args assert kwargs.get("device_serial") == "emulator-5556" - set_active_mock.assert_called_once_with("emulator-5556") @pytest.mark.asyncio diff --git a/tests/unit/admin_console/test_task_queue_service.py b/tests/unit/admin_console/test_task_queue_service.py index f1466ff..5c7fbcc 100644 --- a/tests/unit/admin_console/test_task_queue_service.py +++ b/tests/unit/admin_console/test_task_queue_service.py @@ -222,10 +222,6 @@ async def test_queue_worker_execution_lifecycle(): patch("asyncio.create_subprocess_exec", side_effect=fake_subprocess_exec), patch("apps.admin_console.services.task_queue_service.session_repo") as mock_repo, patch("apps.admin_console.services.task_queue_service.media_service"), - patch( - "artemis.core.diagnostics.readiness_engine.get_active_device_serial", - return_value=None, - ), patch( "artemis.runtime.device_pool.device_pool.select_device", return_value="emulator-5554", @@ -272,10 +268,6 @@ async def test_queue_worker_cmd_construction(): patch("asyncio.create_subprocess_exec", side_effect=fake_subprocess_exec), patch("apps.admin_console.services.task_queue_service.session_repo") as mock_repo, patch("apps.admin_console.services.task_queue_service.media_service"), - patch( - "artemis.core.diagnostics.readiness_engine.get_active_device_serial", - return_value=None, - ), patch( "artemis.runtime.device_pool.device_pool.select_device", return_value="emulator-5554", @@ -489,6 +481,48 @@ def test_stop_tasks_does_not_kill_stale_reused_pid(): terminate_verified.assert_not_called() +def test_stop_tasks_session_target_never_adopts_mismatched_fallback_owner(): + """No live owners is a legal state: a session-targeted stop must not + terminate a fallback owner that belongs to a different session.""" + other_owner = DeviceLockOwner( + pid=24680, + process_created_at=1234.5, + token="other-owner-token", + device_id="emulator-5554", + description="Task for another session", + acquired_at="2026-08-24T00:00:00+00:00", + session_id="other-session", + ingress="cli", + ) + with ( + patch( + "apps.admin_console.services.task_queue_service.DeviceExecutionLock.get_active_owners", + return_value={}, + ), + patch( + "apps.admin_console.services.task_queue_service.DeviceExecutionLock.get_active_owner", + return_value=other_owner, + ), + patch( + "apps.admin_console.services.task_queue_service.DeviceExecutionLock.is_active_owner", + return_value=True, + ), + patch( + "apps.admin_console.services.task_queue_service.process_supervisor.terminate_tree_verified" + ) as terminate_verified, + patch( + "apps.admin_console.services.task_queue_service.session_repo.update_session_status" + ), + patch( + "apps.admin_console.services.task_queue_service.session_repo.get_session_by_id", + return_value=None, + ), + ): + task_queue_service.stop_tasks(clear_all=False, session_id="wanted-session") + + terminate_verified.assert_not_called() + + @pytest.mark.asyncio async def test_status_reports_external_global_owner_without_ipc_connection(): external_owner = DeviceLockOwner( @@ -548,10 +582,6 @@ async def test_cancel_task_triggers_next_pending_task(): patch("asyncio.create_subprocess_exec", side_effect=fake_subprocess_exec), patch("apps.admin_console.services.task_queue_service.session_repo") as mock_repo, patch("apps.admin_console.services.task_queue_service.media_service"), - patch( - "artemis.core.diagnostics.readiness_engine.get_active_device_serial", - return_value=None, - ), patch( "artemis.runtime.device_pool.device_pool.select_device", return_value="emulator-5554", @@ -603,10 +633,6 @@ async def test_immediate_cancel_ignores_stale_ipc_and_runs_next_task(): patch("asyncio.create_subprocess_exec", side_effect=fake_subprocess_exec), patch("apps.admin_console.services.task_queue_service.session_repo") as mock_repo, patch("apps.admin_console.services.task_queue_service.media_service"), - patch( - "artemis.core.diagnostics.readiness_engine.get_active_device_serial", - return_value=None, - ), patch( "artemis.runtime.device_pool.device_pool.select_device", return_value="emulator-5554", diff --git a/tests/unit/core/test_diagnostics.py b/tests/unit/core/test_diagnostics.py index b212e74..6cd1ca4 100644 --- a/tests/unit/core/test_diagnostics.py +++ b/tests/unit/core/test_diagnostics.py @@ -392,11 +392,13 @@ async def test_toolchain_probe_structure(): @pytest.mark.asyncio -async def test_active_device_selection(): - """Verify selecting a device updates engine state.""" +async def test_probe_target_serial_forwards_to_adb_probe(): + """Verify the probe target preference reaches the ADB probe and can be cleared.""" engine = ReadinessEngine() - engine.set_active_device_serial("test-emulator-1234") - assert engine.get_active_device_serial() == "test-emulator-1234" + engine.set_probe_target_serial("test-emulator-1234") + assert engine._adb_probe._target_serial == "test-emulator-1234" + engine.set_probe_target_serial(None) + assert engine._adb_probe._target_serial is None @pytest.mark.asyncio diff --git a/tests/unit/runtime/test_awake_service.py b/tests/unit/runtime/test_awake_service.py index 311ba79..1c29043 100644 --- a/tests/unit/runtime/test_awake_service.py +++ b/tests/unit/runtime/test_awake_service.py @@ -156,11 +156,30 @@ def test_disconnect_reconciliation_stops_heartbeat_and_allows_reenrollment( assert configure.call_count == 2 +@patch("artemis.runtime.device_pool.device_pool.get_claimed_serials") @patch("artemis.runtime.awake_service.AdbClient") -def test_discovery_honors_single_device_policy(adb_client): +def test_discovery_keeps_only_pool_claimed_devices(adb_client, claimed): + """The pool manages two devices while the ADB server lists three: only the + two claimed devices are kept awake; the unrelated one is left untouched.""" + adb_client.return_value.device_list.return_value = [ + MagicMock(serial="device-1"), + MagicMock(serial="device-2"), + MagicMock(serial="device-3"), + ] + claimed.return_value = {"device-1", "device-2"} + + assert _discover_connected_device_ids() == ["device-1", "device-2"] + + +@patch( + "artemis.runtime.device_pool.device_pool.get_claimed_serials", + return_value=set(), +) +@patch("artemis.runtime.awake_service.AdbClient") +def test_discovery_returns_nothing_when_pool_claims_no_device(adb_client, _claimed): adb_client.return_value.device_list.return_value = [ MagicMock(serial="device-1"), MagicMock(serial="device-2"), ] - assert _discover_connected_device_ids() == ["device-1"] + assert _discover_connected_device_ids() == [] diff --git a/tests/unit/runtime/test_device_pool.py b/tests/unit/runtime/test_device_pool.py index 6fff225..0a12399 100644 --- a/tests/unit/runtime/test_device_pool.py +++ b/tests/unit/runtime/test_device_pool.py @@ -91,6 +91,33 @@ def test_device_pool_list_devices_and_busy_status(monkeypatch): lock.release() +def test_get_claimed_serials_reflects_locks_and_reservations(): + pool = DevicePool() + assert pool.get_claimed_serials() == set() + + lock = DeviceExecutionLock("emulator-5554", "claimed by test") + lock.acquire() + bound_ticket = DeviceExecutionLock.reserve( + description="queued task", + device_id="pixel-8-pro", + session_id="queued-session", + ) + unbound_ticket = DeviceExecutionLock.reserve( + description="unbound task", + device_id="pending", + session_id="pending-session", + ) + try: + # Locked and reserved serials are claimed; the unbound placeholder is not. + assert pool.get_claimed_serials() == {"emulator-5554", "pixel-8-pro"} + finally: + DeviceExecutionLock.cancel_reservation(bound_ticket) + DeviceExecutionLock.cancel_reservation(unbound_ticket) + lock.release() + + assert pool.get_claimed_serials() == set() + + RAW_DEVICE = ("emulator-5554", "device", "Emulator", "sdk")