fix: release 1.5.2 with resilient Windows task state writes

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",

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",

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)

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 = [

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)

uv.lock

@@ -206,7 +206,7 @@ wheels = [
 
 [[package]]
 name = "artemis"
-version = "1.5.1"
+version = "1.5.2"
 source = { editable = "." }
 dependencies = [
     { name = "adbutils" },