fix: serialize watchtower state saves

obliteratus/watchtower.py

@@ -212,9 +212,9 @@ class Watchtower:
                     "scan_count": self._scan_count,
                     "models": {mid: m.to_dict() for mid, m in self._models.items()},
                 }
-            # Atomic write via temp file
-            tmp.write_text(json.dumps(data, indent=2, default=str), encoding="utf-8")
-            tmp.replace(self.state_file)
+                # Serialize the shared temporary path as well as its snapshot.
+                tmp.write_text(json.dumps(data, indent=2, default=str), encoding="utf-8")
+                tmp.replace(self.state_file)
         except Exception as e:
             try:
                 tmp.unlink(missing_ok=True)
@@ -445,10 +445,12 @@ class Watchtower:
             by_status = {}
             for m in self._models.values():
                 by_status[m.status] = by_status.get(m.status, 0) + 1
+            last_scan = self._last_scan
+            scan_count = self._scan_count
         return {
             "total_tracked": total,
-            "last_scan": self._last_scan,
-            "scan_count": self._scan_count,
+            "last_scan": last_scan,
+            "scan_count": scan_count,
             "by_status": by_status,
         }
 

tests/test_watchtower_contracts.py

@@ -202,6 +202,54 @@ def test_state_save_failure_cleans_up_without_masking_operation(
     assert "failed to save state" in caplog.text
 
 
+def test_state_writes_serialize_snapshot_and_shared_temporary_path(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    state = tmp_path / "watchtower.json"
+    temporary = state.with_suffix(".tmp")
+    watchtower = Watchtower(state, clock=lambda: NOW, fetch_models=lambda: [])
+    first_write_started = threading.Event()
+    release_first_write = threading.Event()
+    overlapping_write = threading.Event()
+    active_writes = 0
+    active_lock = threading.Lock()
+    original_write_text = Path.write_text
+
+    def controlled_write(path: Path, data: str, **kwargs) -> int:
+        nonlocal active_writes
+        if path != temporary:
+            return original_write_text(path, data, **kwargs)
+        with active_lock:
+            active_writes += 1
+            if active_writes > 1:
+                overlapping_write.set()
+        first_write_started.set()
+        assert release_first_write.wait(timeout=1)
+        try:
+            return original_write_text(path, data, **kwargs)
+        finally:
+            with active_lock:
+                active_writes -= 1
+
+    monkeypatch.setattr(Path, "write_text", controlled_write)
+    first = threading.Thread(target=watchtower._save_state)
+    second = threading.Thread(target=watchtower._save_state)
+    first.start()
+    assert first_write_started.wait(timeout=1)
+    second.start()
+
+    overlapped = overlapping_write.wait(timeout=0.05)
+    release_first_write.set()
+    first.join(timeout=1)
+    second.join(timeout=1)
+
+    assert not overlapped
+    assert not first.is_alive()
+    assert not second.is_alive()
+    assert json.loads(state.read_text(encoding="utf-8"))["scan_count"] == 0
+
+
 @pytest.mark.parametrize(
     "bad_field,bad_value",
     [