diff --git a/apps/showcase_ui/package-lock.json b/apps/showcase_ui/package-lock.json index 3c29c38..ee4e647 100644 --- a/apps/showcase_ui/package-lock.json +++ b/apps/showcase_ui/package-lock.json @@ -1,12 +1,12 @@ { "name": "artemis-showcase-ui", - "version": "1.5.1", + "version": "1.5.2", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "artemis-showcase-ui", - "version": "1.5.1", + "version": "1.5.2", "dependencies": { "@angular/cdk": "^22.1.5", "@angular/common": "^22.1.4", diff --git a/apps/showcase_ui/package.json b/apps/showcase_ui/package.json index 4a681a4..8fe1da7 100644 --- a/apps/showcase_ui/package.json +++ b/apps/showcase_ui/package.json @@ -1,6 +1,6 @@ { "name": "artemis-showcase-ui", - "version": "1.5.1", + "version": "1.5.2", "scripts": { "ng": "ng", "start": "ng serve --proxy-config proxy.conf.json", diff --git a/artemis/runtime/task_control.py b/artemis/runtime/task_control.py index f67389d..45bffab 100644 --- a/artemis/runtime/task_control.py +++ b/artemis/runtime/task_control.py @@ -1,7 +1,7 @@ """Cooperative, session-scoped task guidance and device lease parking.""" import asyncio -from collections.abc import Awaitable +from collections.abc import Awaitable, Callable import json import logging import os @@ -15,6 +15,26 @@ from artemis.runtime.device_lock import DeviceExecutionLock logger = logging.getLogger(__name__) +# Windows readers may briefly deny rename/delete even when directory permissions +# allow writes. Bound the total wait to 1.55s; other IO failures remain immediate. +_WINDOWS_ACCESS_RETRY_DELAYS = (0.05, 0.1, 0.2, 0.4, 0.8) + + +def _retry_windows_access[T](operation: Callable[[], T]) -> T: + delays = iter(_WINDOWS_ACCESS_RETRY_DELAYS) + while True: + try: + return operation() + except OSError as exc: + delay = next(delays, None) + if delay is None or getattr(exc, "winerror", None) not in {5, 32, 33}: + raise + time.sleep(delay) + + +def _unlink_control_file(path: Path) -> None: + _retry_windows_access(lambda: path.unlink(missing_ok=True)) + def control_dir(session_id: str) -> Path: if not session_id or not all(c.isalnum() or c in "-_" for c in session_id): @@ -25,8 +45,21 @@ def control_dir(session_id: str) -> Path: def _write_atomic(path: Path, data: dict) -> None: path.parent.mkdir(parents=True, exist_ok=True) temporary = path.with_name(f".{path.name}.{uuid.uuid4().hex}.tmp") - temporary.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8") - os.replace(temporary, path) + try: + temporary.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8") + _retry_windows_access(lambda: os.replace(temporary, path)) + except OSError: + logger.exception("Cannot write task control file %s", path) + raise + finally: + try: + _unlink_control_file(temporary) + except OSError: + # Cleanup must never replace the original write error or turn a + # successful atomic write into a failed task/control submission. + logger.warning( + "Could not remove task control temporary file %s", temporary, exc_info=True + ) def read_state(session_id: str) -> dict: @@ -39,7 +72,7 @@ def read_state(session_id: str) -> dict: def request_pause(session_id: str) -> None: - (control_dir(session_id) / "resume.request").unlink(missing_ok=True) + _unlink_control_file(control_dir(session_id) / "resume.request") _write_atomic(control_dir(session_id) / "pause.request", {"created_at": time.time()}) @@ -138,7 +171,7 @@ class TaskControl: pause = self.directory / "pause.request" if not pause.exists(): return False - pause.unlink(missing_ok=True) + _unlink_control_file(pause) if on_pause is not None: await on_pause() self.device_lock.release() @@ -146,7 +179,7 @@ class TaskControl: resume = self.directory / "resume.request" while not resume.exists(): await asyncio.sleep(0.2) - resume.unlink(missing_ok=True) + _unlink_control_file(resume) self._state("resuming") previous = self.device_lock self.device_lock = DeviceExecutionLock( @@ -208,4 +241,10 @@ class TaskControl: def release(self) -> None: self.device_lock.release() - (self.directory / "task_control_state.json").unlink(missing_ok=True) + state_path = self.directory / "task_control_state.json" + try: + _unlink_control_file(state_path) + except OSError: + # The lease is already released. An ephemeral status file left + # behind must not override the task's result or block the queue. + logger.warning("Could not remove task control state file %s", state_path, exc_info=True) diff --git a/pyproject.toml b/pyproject.toml index 72c2012..e137100 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -14,7 +14,7 @@ [project] name = "artemis" -version = "1.5.1" +version = "1.5.2" description = "Autonomous AI Assistant for Mobile Automation" authors = [ diff --git a/tests/unit/runtime/test_task_control.py b/tests/unit/runtime/test_task_control.py index 107c6c9..404137b 100644 --- a/tests/unit/runtime/test_task_control.py +++ b/tests/unit/runtime/test_task_control.py @@ -1,12 +1,256 @@ """Paused tasks yield their device and resume with a fresh lease.""" import asyncio +import errno +import json +import logging +import os +from pathlib import Path +import threading import pytest import artemis.runtime.task_control as task_control +def _windows_access_error(code: int) -> PermissionError: + error = PermissionError(errno.EACCES, "state file is in use") + error.winerror = code + return error + + +@pytest.mark.parametrize("winerror", [5, 32, 33]) +def test_running_state_recovers_from_temporary_windows_file_access(tmp_path, monkeypatch, winerror): + monkeypatch.setattr(task_control, "TRACES_PATH", tmp_path) + path = task_control.control_dir("task-123") / "task_control_state.json" + task_control._write_atomic(path, {"status": "queued"}) + replace = os.replace + attempts = [] + delays = [] + + def busy_replace(source, target): + attempts.append(source) + if len(attempts) <= 2: + assert json.loads(path.read_text(encoding="utf-8")) == {"status": "queued"} + raise _windows_access_error(winerror) + return replace(source, target) + + monkeypatch.setattr(task_control.os, "replace", busy_replace) + monkeypatch.setattr(task_control.time, "sleep", delays.append) + task_control.TaskControl("task-123", object())._state("running") + + assert task_control.read_state("task-123")["status"] == "running" + assert delays and sum(delays) < 2 + assert not list(path.parent.glob("*.tmp")) + + +def test_persistent_windows_access_error_is_bounded_and_preserves_old_state(tmp_path, monkeypatch): + path = tmp_path / "task_control_state.json" + task_control._write_atomic(path, {"status": "queued"}) + error = _windows_access_error(5) + attempts = [] + delays = [] + + def denied_replace(source, target): + attempts.append(source) + raise error + + monkeypatch.setattr(task_control.os, "replace", denied_replace) + monkeypatch.setattr(task_control.time, "sleep", delays.append) + with pytest.raises(PermissionError) as raised: + task_control._write_atomic(path, {"status": "running"}) + + assert raised.value is error + assert 1 < len(attempts) < 20 + assert sum(delays) < 2 + assert json.loads(path.read_text(encoding="utf-8")) == {"status": "queued"} + assert not list(tmp_path.glob("*.tmp")) + + +@pytest.mark.parametrize( + "error", [PermissionError(errno.EACCES, "denied"), OSError(errno.ENOSPC, "disk full")] +) +def test_unrelated_file_errors_fail_without_retry_and_clean_temporary_file( + tmp_path, monkeypatch, error +): + path = tmp_path / "task_control_state.json" + task_control._write_atomic(path, {"status": "queued"}) + attempts = [] + delays = [] + + def failed_replace(source, target): + attempts.append(source) + raise error + + monkeypatch.setattr(task_control.os, "replace", failed_replace) + monkeypatch.setattr(task_control.time, "sleep", delays.append) + with pytest.raises(OSError) as raised: + task_control._write_atomic(path, {"status": "running"}) + + assert raised.value is error + assert len(attempts) == 1 + assert delays == [] + assert json.loads(path.read_text(encoding="utf-8")) == {"status": "queued"} + assert not list(tmp_path.glob("*.tmp")) + + +def test_temporary_write_failure_cleans_partial_file(tmp_path, monkeypatch): + path = tmp_path / "task_control_state.json" + task_control._write_atomic(path, {"status": "queued"}) + write_text = Path.write_text + error = OSError(errno.ENOSPC, "disk full") + + def partial_write(target, content, **kwargs): + write_text(target, "{", **kwargs) + raise error + + monkeypatch.setattr(Path, "write_text", partial_write) + with pytest.raises(OSError) as raised: + task_control._write_atomic(path, {"status": "running"}) + + assert raised.value is error + assert json.loads(path.read_text(encoding="utf-8")) == {"status": "queued"} + assert not list(tmp_path.glob("*.tmp")) + + +def test_busy_temporary_cleanup_preserves_original_write_error(tmp_path, monkeypatch, caplog): + path = tmp_path / "task_control_state.json" + task_control._write_atomic(path, {"status": "queued"}) + error = OSError(errno.ENOSPC, "disk full") + unlink = Path.unlink + + def failed_replace(source, target): + raise error + + def busy_unlink(target, **kwargs): + if target.suffix == ".tmp": + raise _windows_access_error(32) + return unlink(target, **kwargs) + + monkeypatch.setattr(task_control.os, "replace", failed_replace) + monkeypatch.setattr(Path, "unlink", busy_unlink) + monkeypatch.setattr(task_control.time, "sleep", lambda _: None) + try: + with caplog.at_level(logging.WARNING, logger=task_control.__name__): + with pytest.raises(OSError) as raised: + task_control._write_atomic(path, {"status": "running"}) + assert raised.value is error + assert "Could not remove task control temporary file" in caplog.text + assert json.loads(path.read_text(encoding="utf-8")) == {"status": "queued"} + finally: + for temporary in tmp_path.glob("*.tmp"): + unlink(temporary) + + +@pytest.mark.parametrize( + "send_request,filename", + [ + (task_control.request_pause, "pause.request"), + (task_control.request_resume, "resume.request"), + ], +) +def test_repeated_control_requests_recover_from_file_sharing( + tmp_path, monkeypatch, send_request, filename +): + monkeypatch.setattr(task_control, "TRACES_PATH", tmp_path) + send_request("task-123") + replace = os.replace + attempts = [] + + def busy_replace(source, target): + attempts.append(target) + if len(attempts) == 1: + raise _windows_access_error(32) + return replace(source, target) + + monkeypatch.setattr(task_control.os, "replace", busy_replace) + monkeypatch.setattr(task_control.time, "sleep", lambda _: None) + send_request("task-123") + + assert ( + json.loads((task_control.control_dir("task-123") / filename).read_text(encoding="utf-8"))[ + "created_at" + ] + > 0 + ) + assert not list(task_control.control_dir("task-123").glob("*.tmp")) + + +@pytest.mark.parametrize("persistent", [False, True]) +def test_release_does_not_fail_task_when_state_cleanup_is_busy( + tmp_path, monkeypatch, caplog, persistent +): + monkeypatch.setattr(task_control, "TRACES_PATH", tmp_path) + released = [] + + class Lock: + def release(self): + released.append(True) + + control = task_control.TaskControl("task-123", Lock()) + control._state("running") + path = control.directory / "task_control_state.json" + unlink = Path.unlink + attempts = [] + + def busy_unlink(target, **kwargs): + if target == path: + attempts.append(target) + if persistent or len(attempts) == 1: + raise _windows_access_error(32) + return unlink(target, **kwargs) + + monkeypatch.setattr(Path, "unlink", busy_unlink) + monkeypatch.setattr(task_control.time, "sleep", lambda _: None) + with caplog.at_level(logging.WARNING, logger=task_control.__name__): + control.release() + + assert released == [True] + assert 1 < len(attempts) < 20 + assert path.exists() == persistent + if persistent: + assert path.name in caplog.text + + +@pytest.mark.skipif(os.name != "nt", reason="Windows file sharing semantics") +def test_running_state_survives_real_windows_reader_handle(tmp_path, monkeypatch): + monkeypatch.setattr(task_control, "TRACES_PATH", tmp_path) + control = task_control.TaskControl("task-123", object()) + control._state("queued") + path = control.directory / "task_control_state.json" + reader = path.open(encoding="utf-8") + blocked = threading.Event() + errors = [] + replace = os.replace + + def release_reader(): + if blocked.wait(2): + reader.close() + + def observe_replace(source, target): + try: + return replace(source, target) + except OSError as exc: + errors.append(exc.winerror) + blocked.set() + raise + + releaser = threading.Thread(target=release_reader) + releaser.start() + monkeypatch.setattr(task_control.os, "replace", observe_replace) + try: + control._state("running") + finally: + blocked.set() + releaser.join(timeout=3) + reader.close() + + assert not releaser.is_alive() + assert errors and set(errors).issubset({5, 32, 33}) + assert task_control.read_state("task-123")["status"] == "running" + assert not list(control.directory.glob("*.tmp")) + + @pytest.mark.asyncio async def test_pause_releases_device_and_resume_waits_for_new_acquisition(tmp_path, monkeypatch): monkeypatch.setattr(task_control, "TRACES_PATH", tmp_path) diff --git a/uv.lock b/uv.lock index 0a27126..f56add0 100644 --- a/uv.lock +++ b/uv.lock @@ -206,7 +206,7 @@ wheels = [ [[package]] name = "artemis" -version = "1.5.1" +version = "1.5.2" source = { editable = "." } dependencies = [ { name = "adbutils" },