fix(automation): atomic state file writing and Windows PermissionError retry (T-20261107-192001888)

src/codex_logdatenbank_wartung/automation_control.py

@@ -185,26 +185,46 @@ def _read_automation(path: Path, *, match_folder: bool = True) -> tomlkit.TOMLDo
     return data
 
 
-def _write_toml(path: Path, document: tomlkit.TOMLDocument) -> None:
+def _atomic_write_bytes(
+    path: Path,
+    payload: bytes,
+    *,
+    retries: int = 5,
+    retry_delay: float = 0.05,
+) -> None:
     """Schreibt atomar: erst vollstaendig kodieren, dann tmp im selben Ordner, fsync, ``os.replace``.
 
     Bricht der Vorgang ab, bleibt die Originaldatei unveraendert und die tmp-Datei wird entfernt (auch eine
     schreibgeschuetzte); der urspruengliche Fehler wird nicht durch einen Aufraeumfehler ersetzt.
-    ``mkstemp`` legt die Datei mit 0600 an: die Rechte des Originals kommen vor dem Austausch auf die tmp-Datei.
+    ``mkstemp`` legt die Datei mit 0600 an: die Rechte des Originals kommen vor dem Austausch auf die tmp-Datei,
+    sofern das Original bereits existiert.
     Ist ``path`` ein Link, wird sein Ziel ersetzt und der Link bleibt bestehen.
+    Unter Windows kann ``os.replace`` bei transienten Dateisperren (z. B. Virenscanner, Backup, Indexer) einen
+    ``PermissionError`` werfen. Dieser wird mit kurzem Retry und exponentiellem Backoff abgefangen.
     """
-    payload = tomlkit.dumps(document).encode("utf-8")
+    path.parent.mkdir(parents=True, exist_ok=True)
     target = Path(os.path.realpath(path))
-    mode = stat.S_IMODE(target.stat().st_mode)
+    try:
+        mode: int | None = stat.S_IMODE(target.stat().st_mode)
+    except OSError:
+        mode = None
     descriptor, tmp_name = tempfile.mkstemp(dir=target.parent, prefix=f"{target.name}.", suffix=".tmp")
     temporary = Path(tmp_name)
     try:
         with os.fdopen(descriptor, "wb") as handle:
             handle.write(payload)
             handle.flush()
             os.fsync(handle.fileno())
-        os.chmod(temporary, mode)
-        os.replace(temporary, target)
+        if mode is not None:
+            os.chmod(temporary, mode)
+        for attempt in range(max(1, retries)):
+            try:
+                os.replace(temporary, target)
+                break
+            except PermissionError:
+                if attempt >= retries - 1:
+                    raise
+                time.sleep(retry_delay * (2 ** attempt))
     except BaseException:
         # Die tmp-Datei kann das Nur-Lesen-Attribut des Originals tragen (Windows): ohne chmod scheitert unlink,
         # laesst die Datei liegen und ueberdeckt den eigentlichen Fehler. Dessen Text bleibt der, der zaehlt.
@@ -216,6 +236,25 @@ def _write_toml(path: Path, document: tomlkit.TOMLDocument) -> None:
         raise
 
 
+def _atomic_write_json(
+    path: Path,
+    data: object,
+    *,
+    indent: int = 2,
+    retries: int = 5,
+    retry_delay: float = 0.05,
+) -> None:
+    """Schreibt ein JSON-Objekt atomar via ``_atomic_write_bytes``."""
+    payload = (json.dumps(data, ensure_ascii=False, indent=indent) + "\n").encode("utf-8")
+    _atomic_write_bytes(path, payload, retries=retries, retry_delay=retry_delay)
+
+
+def _write_toml(path: Path, document: tomlkit.TOMLDocument) -> None:
+    """Schreibt atomar: erst vollstaendig kodieren, dann tmp im selben Ordner, fsync, ``os.replace``."""
+    payload = tomlkit.dumps(document).encode("utf-8")
+    _atomic_write_bytes(path, payload)
+
+
 def _quoted_string(value: object) -> str:
     return str(value or "")
 
@@ -343,13 +382,12 @@ def load_carecenter_paused_ids(config: MaintenanceConfig) -> list[str]:
 
 def _save_carecenter_paused_ids(config: MaintenanceConfig, ids: set[str]) -> None:
     path = control_state_path(config)
-    path.parent.mkdir(parents=True, exist_ok=True)
     payload = {
         "tool": "CareCenter for Codex",
         "updated_at": _now_iso(),
         "paused_by_carecenter_ids": sorted(ids),
     }
-    path.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
+    _atomic_write_json(path, payload, indent=2)
 
 
 def _append_event(config: MaintenanceConfig, result: AutomationControlResult) -> None:
@@ -361,6 +399,7 @@ def _append_event(config: MaintenanceConfig, result: AutomationControlResult) ->
     }
     with path.open("a", encoding="utf-8") as handle:
         handle.write(json.dumps(payload, ensure_ascii=False) + "\n")
+        handle.flush()
 
 
 def _set_status(record: AutomationRecord, status: str) -> bool:
@@ -438,7 +477,8 @@ def pause_active_automations(config: MaintenanceConfig) -> AutomationControlResu
         status="ok",
         target_count=len(active_records),
     )
-    carecenter_ids = set(load_carecenter_paused_ids(config))
+    initial_ids = set(load_carecenter_paused_ids(config))
+    carecenter_ids = set(initial_ids)
     for record in active_records:
         try:
             if _set_status(record, PAUSED):
@@ -451,7 +491,8 @@ def pause_active_automations(config: MaintenanceConfig) -> AutomationControlResu
     _record_unparseable(result, unparseable)
     if result.errors:
         result.status = "partial" if result.changed_ids else "failed"
-    _save_carecenter_paused_ids(config, carecenter_ids)
+    if result.changed_ids or carecenter_ids != initial_ids:
+        _save_carecenter_paused_ids(config, carecenter_ids)
     _append_event(config, result)
     return result
 

tests/test_automation_control.py

@@ -373,8 +373,9 @@ def spy_replace(src: object, dst: object) -> None:
 
     pause_active_automations(config)
 
-    assert [name for name, _ in events] == ["fsync", "replace"]
+    assert [name for name, _ in events] == ["fsync", "replace", "fsync", "replace"]
     assert events[1][1] == (path.parent, path)
+    assert events[3][1] == (control_state_dir(config), control_state_path(config))
     assert read_status(path) == "PAUSED"
 
 
@@ -537,7 +538,7 @@ def spy_replace(src: object, dst: object) -> None:
 
     pause_active_automations(config)
 
-    assert calls == [("chmod", original_mode, True), ("replace", 0, False)]
+    assert calls[:2] == [("chmod", original_mode, True), ("replace", 0, False)]
 
 
 @pytest.mark.skipif(os.name != "nt", reason="nur unter Windows scheitert os.replace an einer schreibgeschuetzten Datei")
@@ -615,3 +616,105 @@ def fake_realpath(value: object, *args: object, **kwargs: object) -> str:
     assert read_status(target) == "PAUSED"
     assert link_path.read_bytes() == link_before, "der Link-Pfad selbst bleibt unberuehrt"
     assert sorted(item.name for item in real_dir.iterdir()) == ["job.toml"]
+
+
+# --- Atomare Zustandsdateien & PermissionError-Retry (T-20261107-192001888) -------------------------
+
+
+def test_save_carecenter_paused_ids_is_atomic_and_syncs_in_same_folder(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    config = make_config(tmp_path)
+    state_file = control_state_path(config)
+    state_dir = control_state_dir(config)
+    events: list[tuple[str, object]] = []
+    real_fsync, real_replace = os.fsync, os.replace
+
+    def spy_fsync(fd: int) -> None:
+        events.append(("fsync", fd))
+        real_fsync(fd)
+
+    def spy_replace(src: object, dst: object) -> None:
+        events.append(("replace", (Path(str(src)).parent, Path(str(dst)))))
+        real_replace(src, dst)
+
+    monkeypatch.setattr(automation_control.os, "fsync", spy_fsync)
+    monkeypatch.setattr(automation_control.os, "replace", spy_replace)
+
+    automation_control._save_carecenter_paused_ids(config, {"job-1", "job-2"})
+
+    assert [name for name, _ in events] == ["fsync", "replace"]
+    assert events[1][1] == (state_dir, state_file)
+    assert load_carecenter_paused_ids(config) == ["job-1", "job-2"]
+    # Es darf keine verwaiste temporaere Datei im Verzeichnis verbleiben
+    assert sorted(item.name for item in state_dir.iterdir()) == ["paused-by-carecenter.json"]
+
+
+def test_save_carecenter_paused_ids_retries_on_transient_permission_error(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    config = make_config(tmp_path)
+    attempts = 0
+    real_replace = os.replace
+
+    def flaky_replace(src: object, dst: object) -> None:
+        nonlocal attempts
+        attempts += 1
+        if attempts < 3:
+            raise PermissionError("Access is denied (transient lock)")
+        real_replace(src, dst)
+
+    monkeypatch.setattr(automation_control.os, "replace", flaky_replace)
+    monkeypatch.setattr(automation_control.time, "sleep", lambda _s: None)
+
+    automation_control._save_carecenter_paused_ids(config, {"resilient-job"})
+
+    assert attempts == 3
+    assert load_carecenter_paused_ids(config) == ["resilient-job"]
+    assert sorted(item.name for item in control_state_dir(config).iterdir()) == ["paused-by-carecenter.json"]
+
+
+def test_save_carecenter_paused_ids_fails_cleanly_on_permanent_permission_error(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    config = make_config(tmp_path)
+    state_file = control_state_path(config)
+    attempts = 0
+
+    def blocked_replace(src: object, dst: object) -> None:
+        nonlocal attempts
+        attempts += 1
+        raise PermissionError("Access is denied (permanent lock)")
+
+    monkeypatch.setattr(automation_control.os, "replace", blocked_replace)
+    monkeypatch.setattr(automation_control.time, "sleep", lambda _s: None)
+
+    with pytest.raises(PermissionError, match="permanent lock"):
+        automation_control._save_carecenter_paused_ids(config, {"failed-job"})
+
+    assert attempts == 5
+    # Tmp-Datei muss aufgeraeumt sein, kein Muell
+    state_dir = control_state_dir(config)
+    assert not state_file.exists()
+    assert list(state_dir.iterdir()) == []
+
+
+def test_save_carecenter_paused_ids_interrupted_leaves_previous_state_intact(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    config = make_config(tmp_path)
+    automation_control._save_carecenter_paused_ids(config, {"initial-job"})
+    previous_bytes = control_state_path(config).read_bytes()
+
+    def crash_replace(src: object, dst: object) -> None:
+        raise OSError("unexpected crash during replace")
+
+    monkeypatch.setattr(automation_control.os, "replace", crash_replace)
+
+    with pytest.raises(OSError, match="unexpected crash"):
+        automation_control._save_carecenter_paused_ids(config, {"overwritten-job"})
+
+    assert control_state_path(config).read_bytes() == previous_bytes
+    assert load_carecenter_paused_ids(config) == ["initial-job"]
+    assert sorted(item.name for item in control_state_dir(config).iterdir()) == ["paused-by-carecenter.json"]
+