feat: move syscalls off UI thread; refactor mach write protection

.github/FUNDING.yml

@@ -0,0 +1,3 @@
+# These are supported funding model platforms
+
+github: JeanExtreme002

PyMemoryEditor/app/cheat_poll_worker.py

@@ -62,12 +62,16 @@ class _CheatPollWorker(QThread):
 
     values_ready = Signal(object)  # list[tuple[int, type, int, Any]]
     freeze_failed = Signal(object)  # dict[_EntryKey, str] — current failing freezes
+    write_failed = Signal(object)  # tuple[int, type, int, str] — a manual write that failed
 
     def __init__(self, process: AbstractProcess, parent=None):
         super().__init__(parent)
         self._process = process
         self._mutex = QMutex()
         self._snapshot: List[Tuple[int, type, int, Any, bool]] = []
+        # One-shot manual writes queued from the UI (inline value edits). Drained
+        # at the top of each tick so the syscall runs here, not on the UI thread.
+        self._pending_writes: List[Tuple[int, type, int, Any]] = []
         self._stop = False
         # Entries whose freeze write is currently failing → last error string.
         # Touched only by the worker thread (run() / _poll_once), so no lock.
@@ -87,29 +91,83 @@ def update_snapshot(
         with QMutexLocker(self._mutex):
             self._snapshot = list(snapshot)
 
+    def request_write(
+        self, address: int, pytype: type, length: int, value: Any
+    ) -> None:
+        """Queue a one-shot manual write to be performed on the worker thread.
+
+        The cheat table's inline value edits used to call
+        ``write_process_memory`` directly on the UI thread, freezing the UI when
+        the target was slow or the page faulted. Routing them here keeps the
+        syscall off the UI thread; a write that fails comes back via the
+        ``write_failed`` signal. Latency is at most one tick (~100 ms), which is
+        imperceptible for a manual edit.
+        """
+        with QMutexLocker(self._mutex):
+            self._pending_writes.append((address, pytype, length, value))
+
     def stop(self) -> None:
         with QMutexLocker(self._mutex):
             self._stop = True
 
+    def _drain_pending_writes(self) -> List[Tuple[int, type, int, str]]:
+        """Perform and clear every queued manual write.
+
+        Returns one ``(address, pytype, length, message)`` tuple per write that
+        failed, so ``run`` can surface it via ``write_failed``. Reads and clears
+        the queue under the lock, then runs the syscalls outside it so a slow
+        write never blocks a UI thread enqueuing the next edit.
+        """
+        with QMutexLocker(self._mutex):
+            if not self._pending_writes:
+                return []
+            pending = self._pending_writes
+            self._pending_writes = []
+
+        failures: List[Tuple[int, type, int, str]] = []
+        for address, pytype, length, value in pending:
+            try:
+                self._process.write_process_memory(address, pytype, length, value)
+            except Exception as exc:  # noqa: BLE001
+                message = "%s: %s" % (type(exc).__name__, exc)
+                failures.append((address, pytype, length, message))
+                _LOG.warning(
+                    "Cheat-table write failed at 0x%X (%s, %dB): %s",
+                    address, pytype.__name__, length, message,
+                )
+        return failures
+
     def run(self) -> None:
         while True:
             with QMutexLocker(self._mutex):
                 if self._stop:
                     return
                 snapshot = list(self._snapshot)
 
-            if snapshot:
-                results = self._poll_once(snapshot)
-                if results:
-                    self.values_ready.emit(results)
-
-            # Emit the freeze-failure state only when it changed since the last
-            # tick — a frozen page that keeps failing shouldn't fire the signal
-            # 10×/second, but the UI must learn the moment one starts or stops
-            # failing (including recovering back to an empty map).
-            if self._freeze_failures != self._last_emitted_failures:
-                self._last_emitted_failures = dict(self._freeze_failures)
-                self.freeze_failed.emit(dict(self._freeze_failures))
+            # Never let an exception escape the loop body: a QThread whose
+            # run() raises dies silently, and the table would then stop
+            # auto-updating forever while the rest of the app (separate
+            # workers) keeps working. Catch, log the cause, and keep ticking.
+            try:
+                # Apply queued manual writes before reading, so a just-typed
+                # value lands and is read back as the new current value this tick.
+                for failure in self._drain_pending_writes():
+                    self.write_failed.emit(failure)
+
+                if snapshot:
+                    results = self._poll_once(snapshot)
+                    if results:
+                        self.values_ready.emit(results)
+
+                # Emit the freeze-failure state only when it changed since the
+                # last tick — a frozen page that keeps failing shouldn't fire the
+                # signal 10×/second, but the UI must learn the moment one starts
+                # or stops failing (including recovering back to an empty map).
+                if self._freeze_failures != self._last_emitted_failures:
+                    self._last_emitted_failures = dict(self._freeze_failures)
+                    self.freeze_failed.emit(dict(self._freeze_failures))
+            except Exception:  # noqa: BLE001
+                _LOG.exception("Cheat poll tick failed; continuing")
 
             QThread.msleep(TICK_INTERVAL_MS)
 

PyMemoryEditor/app/cheat_table.py

@@ -17,7 +17,6 @@
 """
 import copy
 import json
-import logging
 from typing import Dict, List, Optional, Tuple
 
 from PySide6.QtCore import Qt, QTimer, Signal
@@ -56,9 +55,6 @@
 # poll-interval constant from this module before the split.
 _TICK_INTERVAL_MS = TICK_INTERVAL_MS
 
-# Child of "PyMemoryEditor" — the Log Console captures these via propagation.
-_LOG = logging.getLogger(__name__)
-
 
 class CheatTable(QWidget):
     """Bottom pane: saved addresses, freezing, manual edits."""
@@ -85,6 +81,7 @@ def __init__(self, process: AbstractProcess, parent=None):
         self._poller = _CheatPollWorker(process, self)
         self._poller.values_ready.connect(self._on_values_ready)
         self._poller.freeze_failed.connect(self._on_freeze_failed)
+        self._poller.write_failed.connect(self._on_write_failed)
         self._poller.start()
 
         # A short cadence to push fresh entry snapshots into the worker. This
@@ -314,23 +311,15 @@ def _on_cell_changed(self, row: int, column: int) -> None:
                 self._suspend_signals = False
                 return
 
-            try:
-                self._process.write_process_memory(
-                    entry.address, entry.spec.pytype, entry.length, value
-                )
-            except Exception as exc:  # noqa: BLE001
-                QMessageBox.critical(
-                    self, "Write Failed", f"{type(exc).__name__}: {exc}"
-                )
-                _LOG.warning(
-                    "Cheat-table write failed at 0x%X (%s, %dB): %s: %s",
-                    entry.address,
-                    entry.spec.pytype.__name__,
-                    entry.length,
-                    type(exc).__name__,
-                    exc,
-                )
-                return
+            # Route the write through the poll worker so the syscall never runs
+            # on the UI thread — a slow target or a page fault would otherwise
+            # freeze the interface. The local update below is optimistic: the
+            # next poll tick reads the value back and corrects the cell if the
+            # write didn't take, and an outright failure returns via
+            # write_failed → _on_write_failed.
+            self._poller.request_write(
+                entry.address, entry.spec.pytype, entry.length, value
+            )
 
             entry.last_value = value
             if entry.frozen:
@@ -424,6 +413,17 @@ def _on_freeze_failed(self, failures: dict) -> None:
         finally:
             self._suspend_signals = False
 
+    def _on_write_failed(self, failure) -> None:
+        """Surface a manual value write that failed on the worker thread.
+
+        The write is now async (queued onto the poll worker), so its failure
+        arrives here via the ``write_failed`` signal rather than inline. The
+        optimistic cell update made when the edit was queued is corrected by the
+        next poll tick, which reads back the unchanged value.
+        """
+        _address, _pytype, _length, message = failure
+        QMessageBox.critical(self, "Write Failed", message)
+
     def _editing_row(self) -> int:
         """Return the row currently being edited, or -1 if none."""
         if self._table.state() != QAbstractItemView.EditingState:

PyMemoryEditor/app/main_window.py

@@ -344,21 +344,16 @@ def _on_first_scan(self, request: ScanRequest) -> None:
         if self._worker is not None:
             return
 
-        # Build a cached region snapshot the first time the user asks for one.
-        if self._scanner.use_snapshot_cache() and self._region_snapshot is None:
-            try:
-                self._region_snapshot = self._process.snapshot_memory_regions()
-            except Exception as exc:  # noqa: BLE001
-                QMessageBox.warning(
-                    self,
-                    "Memory regions",
-                    f"Could not cache memory regions ({exc}). Continuing without cache.",
-                )
-                self._region_snapshot = None
-
-        request.memory_regions = (
-            self._region_snapshot if self._scanner.use_snapshot_cache() else None
-        )
+        # Decide how the worker sources its region list. When the user wants the
+        # snapshot cache and we don't have one yet, let the worker build it off
+        # the UI thread (region enumeration can be slow on a large target) and
+        # hand it back via snapshot_ready for reuse by the refine scans.
+        if self._scanner.use_snapshot_cache():
+            request.memory_regions = self._region_snapshot
+            request.build_snapshot = self._region_snapshot is None
+        else:
+            request.memory_regions = None
+            request.build_snapshot = False
 
         self._results_model.clear()
         self._results_model.set_value_spec(request.spec)
@@ -377,6 +372,9 @@ def _on_first_scan(self, request: ScanRequest) -> None:
         worker.status.connect(self._status.showMessage)
         worker.error.connect(self._on_worker_error)
         worker.finished_ok.connect(self._on_first_scan_done)
+        # The worker builds the region snapshot off the UI thread; cache it here
+        # for the refine/update scans that reuse it.
+        worker.snapshot_ready.connect(self._on_snapshot_ready)
         # Connection order matters: _cleanup_worker must clear self._worker
         # before _fill_initial_values runs, otherwise the busy guard in
         # _on_update_values rejects the auto-refresh.
@@ -695,6 +693,10 @@ def _open_hex_viewer_with_size(self, address: int, size: int) -> None:
         self._hex_viewers.append(viewer)
         viewer.show()
 
+    def _on_snapshot_ready(self, snapshot) -> None:
+        """Cache the region snapshot the FirstScanWorker built off-thread."""
+        self._region_snapshot = snapshot
+
     def _refresh_region_snapshot(self) -> None:
         try:
             self._region_snapshot = self._process.snapshot_memory_regions()

PyMemoryEditor/app/scan_worker.py

@@ -71,6 +71,11 @@ class ScanRequest:
     # Optional cached snapshot of memory regions, reused across scans to skip
     # the region enumeration step. Pass None to let the backend enumerate.
     memory_regions: Optional[Sequence[MemoryRegion]] = None
+    # When True and no snapshot is supplied, FirstScanWorker builds one itself
+    # (off the UI thread) and emits it via ``snapshot_ready`` so the caller can
+    # cache it for later scans. Building it here keeps the (potentially slow)
+    # region enumeration off the UI thread.
+    build_snapshot: bool = False
 
 
 def build_scan_request(
@@ -172,13 +177,34 @@ def cancel(self) -> None:
 class FirstScanWorker(_BaseWorker):
     """Performs the very first scan, finding every address that matches."""
 
+    # Emitted (off the UI thread) with the freshly built MemoryRegionSnapshot
+    # when ``request.build_snapshot`` was set, so the owner can cache it.
+    snapshot_ready = Signal(object)
+
     def __init__(self, process: AbstractProcess, request: ScanRequest, parent=None):
         super().__init__(process, parent)
         self._request = request
 
     def run(self) -> None:
         req = self._request
         try:
+            # Build the region snapshot here rather than on the UI thread: the
+            # enumeration can be slow on a large target and would otherwise
+            # stall the scan dialog. Emit it so the owner can cache it for the
+            # refine/update scans that follow.
+            if req.memory_regions is None and req.build_snapshot:
+                try:
+                    req.memory_regions = self._process.snapshot_memory_regions()
+                    self.snapshot_ready.emit(req.memory_regions)
+                except Exception as exc:  # noqa: BLE001
+                    # Snapshot is an optimization; fall back to per-scan
+                    # enumeration rather than failing the whole scan.
+                    _LOG.warning(
+                        "Region snapshot failed; scanning without cache: %s: %s",
+                        type(exc).__name__,
+                        exc,
+                    )
+                    req.memory_regions = None
             # Pattern path: req.value is the IDA-style string or a bytes regex,
             # routed through search_by_pattern. req.length carries byte_length —
             # ignored for IDA strings (inferred from the token count), required