fix(tests): pin glibc's mmap threshold in the browser test server (#516)

Each tests/browser/server.py grew from about 13 MB to 0.6-1 GB RSS over
a page-workflow shard while its Python heap stayed near 12 MB. glibc
raises its mmap threshold (up to 32 MiB) after a large free, so the
seeder's image-sized buffers came from per-thread malloc arenas, which
fragmented and kept their pages. The server now fixes the threshold at
glibc's 128 KiB start value (mallopt M_MMAP_THRESHOLD) before it starts,
so large blocks go back to the OS on free. No memory limit is set; off
glibc it does nothing.

Closes #516

.github/workflows/portability.yml

@@ -26,9 +26,6 @@ env:
 # Measured 2026-10-29 on main runs 40935337099 (dd247e3), 40938407252 (dd247e3), 40941240624 (dd247e3):
 #   slowest leg: 7m24s (py3.12 windows) / 7m31s (py3.13 windows) / 7m32s (py3.14 windows) -> 15 min
 #     (windows 4m56s-7m32s, macos 4m39s-6m14s)
-#   native-trash (#499) has no green main runs yet: its probe, run 40949370943 (2026-10-29), took
-#     about 1 minute per macOS leg, so ceil(1.5 x 2) + 3 gives it 6 minutes (step 4). Recalibrate
-#     after three green runs.
 jobs:
   test:
     name: py${{ matrix.python-version }} on ${{ matrix.os }}
@@ -79,28 +76,3 @@ jobs:
         env:
           PM_TEST_SLOWEST: "40"
         run: python -m tests.run_suite -v
-
-  # The real Trash on both macOS CPUs (#499 design §10.2). It runs apart from the suite, so a red
-  # suite never skips the native proof.
-  native-trash:
-    name: native trash on ${{ matrix.os }}
-    runs-on: ${{ matrix.os }}
-    timeout-minutes: 6
-    strategy:
-      fail-fast: false
-      matrix:
-        os: [macos-latest, macos-15-intel]
-
-    steps:
-      - uses: actions/checkout@v7
-
-      - name: Set up Python
-        uses: actions/setup-python@v7
-        with:
-          python-version: "3.12"
-
-      - name: Native trash proof
-        timeout-minutes: 4
-        env:
-          PM_NATIVE_TRASH: macos
-        run: python -m unittest tests.test_trash_macos_native tests.test_trash_macos -v

TESTING.md

@@ -2487,30 +2487,14 @@ files to the system trash. **No code path unlinks or overwrites a media file.**
   freedesktop Trash specification (home trash on the same mount, else the mount's
   `.Trash/$uid` or `.Trash-$uid`), reserve the name with `O_EXCL` and move with
   `renameat2(RENAME_NOREPLACE)`; without it, across devices, on a read-only mount or with
-  a trash inside an output root the move is refused (`no-trash`, `trash-in-root`). macOS
-  (#499, `utils/trash_macos.py`) calls `-[NSFileManager trashItemAtURL:resultingItemURL:error:]`
-  through `ctypes` (ten fixed `objc_msgSend` prototypes, nil checks before every send, one
-  autorelease pool per call, loaded on first use). It answers in the review dialog, without
-  writing: `network` (not a local volume), `no-trash` (read-only, or no trash folder
-  macOS would use), `cloud-sync` (iCloud Drive, `~/Library/CloudStorage`, or ubiquitous),
-  `trash-in-root` (the trash folder is inside an output root, compared by `(st_dev,
-  st_ino)` of every ancestor) and `no-permission` (folder not writable, an immutable or
-  append flag, a foreign file in a sticky folder). A fresh volume's `.Trashes/<uid>` is
-  predicted (`URLForDirectory` answers 3328/`ENOENT` until macOS makes it). A refused move
-  keeps the file and logs the `NSError`; a move macOS reports done but whose place cannot be
-  shown (no result URL, or one inside an output root) goes under `errors` as
-  `trash-unconfirmed`, keeps its records and stops the job. Windows
+  a trash inside an output root the move is refused (`no-trash`, `trash-in-root`). Windows
   never uses the Recycle Bin (#500, maintainer ruling (a)): files larger than the bin's
   capacity are deleted permanently and IFileOperation's `PreDeleteItem` guard cannot detect
   it (native probe runs 40951037575, 40953347904), so every file answers `windows-recycle`
   ("Moving files to the Recycle Bin isn't available on Windows: files larger than the bin's
-  capacity would be deleted permanently."). Library-only removal works everywhere. Tests use `tests/fake_trash.py`; the
-  browser test server installs it, so no test touches a real trash. The native proof,
-  `tests/test_trash_macos_native.py`, runs only with `PM_NATIVE_TRASH=macos` in
-  portability.yml's `native-trash` job (`macos-latest` and `macos-15-intel`): the startup
-  volume, a name clash, fresh APFS and FAT32 disk images, an image used as an output root,
-  a read-only image and an unwritable folder (design and probe:
-  `docs/egm/499-macos-trash/`, local).
+  capacity would be deleted permanently."). macOS reports `no-trash` until #499 lands;
+  library-only removal works everywhere. Tests use `tests/fake_trash.py`; the browser test server installs it, so no
+  test touches a real trash.
 
 ### Prompt library cards and recorded deviations (#249 PR E)
 

py/api/image_delete.py

@@ -65,9 +65,6 @@ TITLES = {
     "trash": "Delete images (move to the trash)",
 }
 MAX_DETAIL_ROWS = 50
-# The system moved the file but its place in a trash could not be shown (#499).
-TRASH_UNCONFIRMED = "trash-unconfirmed"
-RECORDS_FAILED_CODE = "records-failed"
 
 # Reason codes (design §4.4) and the sentence each shows.
 REASONS = {
@@ -86,17 +83,6 @@ REASONS = {
     trash.TRASH_IN_ROOT: "This drive's trash is inside an output folder.",
     trash.LOCKED: "The file is open in another program. Close it and try again.",
     trash.FAILED: "The file could not be moved to the trash.",
-    trash.NO_PERMISSION: (
-        "PromptManager isn't allowed to move this file (no permission, or it is "
-        "locked in Finder)."
-    ),
-    trash.CLOUD_SYNC: (
-        "Files in iCloud Drive or a cloud-synced folder can't be moved to the "
-        "trash here."
-    ),
-    TRASH_UNCONFIRMED: (
-        "Moved by macOS, but its place in the trash could not be confirmed."
-    ),
     trash.WINDOWS_RECYCLE: (
         "Moving files to the Recycle Bin isn't available on Windows: files larger "
         "than the bin's capacity would be deleted permanently."
@@ -341,7 +327,7 @@ class ImageDeleteMixin:
         job.update(total=len(targets), status_line=TITLES[mode])
         roots = [str(root) for root in self._get_all_output_dirs()]
         done_group = "trashed" if mode == "trash" else "removed"
-        changed, failed = 0, None
+        changed, failed = 0, False
         for done, target in enumerate(targets, start=1):
             if failed:
                 self._record_delete(job, "not_changed", target, "not-attempted")
@@ -350,8 +336,7 @@ class ImageDeleteMixin:
                 self._delete_one_image, db, target, mode, roots
             )
             if group == "errors":
-                failed = code or RECORDS_FAILED_CODE
-                text = REASONS[code] if code else RECORDS_FAILED
+                failed = True
                 job.record(
                     "errors",
                     {
@@ -359,9 +344,9 @@ class ImageDeleteMixin:
                         "root": target["root_label"],
                         "root_index": target["root_index"],
                         "root_scope": target["root_scope"],
-                        "detail": text,
+                        "detail": RECORDS_FAILED,
                     },
-                    f"! {target['root_label']}/{target['rel']} · {text}",
+                    f"! {target['root_label']}/{target['rel']} · {RECORDS_FAILED}",
                 )
             elif group == done_group:
                 changed += 1
@@ -380,10 +365,8 @@ class ImageDeleteMixin:
         job.log_line(line)
         job.update(progress=100, status_line=f"{line} · {backup_note(backup)}")
         if failed:
-            # "records-failed" names every stop at an errors row (the page's
-            # "Stopped after N of M" toast); the row says which error it was.
-            job.scope["failure"] = RECORDS_FAILED_CODE
-            job.publish({"type": "error", "code": failed, "message": line})
+            job.scope["failure"] = "records-failed"
+            job.publish({"type": "error", "code": "records-failed", "message": line})
             job.finish("failed")
 
     @staticmethod
@@ -422,19 +405,7 @@ class ImageDeleteMixin:
                 label = trash.move_to_trash(target["abs"], roots)
             except trash.TrashRefused as refused:
                 code = refused.code if refused.code in REASONS else trash.FAILED
-                self.logger.info(
-                    "Not moved to the trash (%s): %s %s",
-                    code,
-                    target["rel"],
-                    refused.detail,
-                )
                 return "not_changed", code, None
-            except trash.TrashUnconfirmed as unconfirmed:
-                # The file left its folder: keep its records, stop the job.
-                self.logger.warning(
-                    "Trash move not confirmed: %s %s", target["rel"], unconfirmed.detail
-                )
-                return "errors", TRASH_UNCONFIRMED, None
             mark = None
         else:
             mark = (target["size"], target["mtime_ns"])

tests/browser/malloc_tuning.py

@@ -0,0 +1,31 @@
+"""Keep the browser test server's large frees going back to the OS (#516).
+
+glibc serves an allocation of at least the mmap threshold (128 KiB at start) with its
+own mmap and unmaps it on free. Freeing such a block also raises the threshold, up to
+32 MiB, so the server's later image-sized buffers (PIL pixel data, PNG encoding,
+whole-file reads in the executor threads) come from per-thread malloc arenas instead.
+Those arenas fragment and keep their pages: a server grew from 13 MB to 0.6-1 GB RSS
+over one page-workflow shard while its Python heap stayed near 12 MB. Setting
+M_MMAP_THRESHOLD fixes the threshold at its start value and turns the raising off.
+"""
+
+import ctypes
+import sys
+
+M_MMAP_THRESHOLD = -3
+MMAP_THRESHOLD_BYTES = 128 * 1024
+
+
+def pin_mmap_threshold(libc=None) -> bool:
+    """Fix glibc's mmap threshold; False where there is no glibc ``mallopt``."""
+    if not sys.platform.startswith("linux"):
+        return False
+    if libc is None:
+        try:
+            libc = ctypes.CDLL(None)
+        except OSError:
+            return False
+    mallopt = getattr(libc, "mallopt", None)
+    if mallopt is None:
+        return False
+    return bool(mallopt(M_MMAP_THRESHOLD, MMAP_THRESHOLD_BYTES))

tests/browser/server.py

@@ -20,6 +20,7 @@ from PIL import Image
 from PIL.PngImagePlugin import PngInfo
 
 from tests.browser.folder_trash import close_trash, replace_folder
+from tests.browser.malloc_tuning import pin_mmap_threshold
 from tests.browser.request_hold import register_request_holds
 from tests.browser import slow_reset_watch
 from tests.browser.fast_reset import SEED_ON_LIVE_ENV, ScratchSeeder
@@ -755,6 +756,8 @@ def main(argv: list[str] | None = None) -> None:
         help="shut down when stdin closes (the launcher's parent-death guard)",
     )
     args = parser.parse_args(argv)
+    # Before any thread allocates (#516): large frees then return to the OS.
+    pin_mmap_threshold()
     listener = bind_listener(args.port)
     # One machine-readable line tells the launcher which port this process holds.
     print(f"{PORT_ANNOUNCEMENT}{listener.getsockname()[1]}", flush=True)

tests/test_browser_malloc_tuning.py

@@ -0,0 +1,92 @@
+"""#516: the browser test server pins glibc's mmap threshold.
+
+Large frees then go back to the OS instead of fragmenting per-thread arenas.
+"""
+
+import ctypes
+import sys
+import unittest
+from unittest import mock
+
+from tests.browser import malloc_tuning
+
+
+class FakeLibc:
+    def __init__(self, result=1):
+        self.calls = []
+        self.result = result
+
+    def mallopt(self, option, value):
+        self.calls.append((option, value))
+        return self.result
+
+
+class PinMmapThresholdTests(unittest.TestCase):
+    def test_sets_a_fixed_threshold_through_mallopt(self):
+        libc = FakeLibc()
+        with mock.patch.object(sys, "platform", "linux"):
+            self.assertTrue(malloc_tuning.pin_mmap_threshold(libc))
+        # M_MMAP_THRESHOLD is -3; setting it also turns off glibc's dynamic
+        # raising of the threshold.
+        self.assertEqual(libc.calls, [(-3, 128 * 1024)])
+
+    def test_reports_a_refused_setting(self):
+        with mock.patch.object(sys, "platform", "linux"):
+            self.assertFalse(malloc_tuning.pin_mmap_threshold(FakeLibc(result=0)))
+
+    def test_skips_a_libc_without_mallopt(self):
+        with mock.patch.object(sys, "platform", "linux"):
+            self.assertFalse(malloc_tuning.pin_mmap_threshold(object()))
+
+    def test_skips_other_platforms_without_loading_a_libc(self):
+        for name in ("win32", "darwin"):
+            with (
+                self.subTest(name),
+                mock.patch.object(sys, "platform", name),
+                mock.patch.object(
+                    ctypes, "CDLL", side_effect=AssertionError("no libc load")
+                ),
+            ):
+                self.assertFalse(malloc_tuning.pin_mmap_threshold())
+
+    def test_skips_when_the_process_libc_cannot_be_loaded(self):
+        with (
+            mock.patch.object(sys, "platform", "linux"),
+            mock.patch.object(ctypes, "CDLL", side_effect=OSError("no libc")),
+        ):
+            self.assertFalse(malloc_tuning.pin_mmap_threshold())
+
+    @unittest.skipUnless(
+        sys.platform.startswith("linux")
+        and hasattr(ctypes.CDLL(None), "gnu_get_libc_version"),
+        "glibc only",
+    )
+    def test_glibc_accepts_the_setting(self):
+        # This pins the threshold for the rest of this test process too; the
+        # test server does the same at start, so nothing here depends on it.
+        self.assertTrue(malloc_tuning.pin_mmap_threshold())
+
+
+class ServerMainTests(unittest.TestCase):
+    def test_main_pins_the_threshold_before_it_starts_serving(self):
+        from tests.browser import server
+
+        order = []
+
+        def bind(port):
+            order.append("bind")
+            raise RuntimeError("stop before serving")
+
+        with (
+            mock.patch.object(
+                server, "pin_mmap_threshold", side_effect=lambda: order.append("pin")
+            ),
+            mock.patch.object(server, "bind_listener", side_effect=bind),
+        ):
+            with self.assertRaises(RuntimeError):
+                server.main(["--port", "0"])
+        self.assertEqual(order, ["pin", "bind"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_image_delete_routes.py

@@ -320,86 +320,19 @@ class TestFileCheck(ImageDeleteTestCase):
             self.assertIn(code, image_delete.REASONS)
 
 
-class TestMacOSOutcomes(ImageDeleteTestCase):
-    """The codes and the unconfirmed outcome the macOS backend adds (#499)."""
-
-    async def test_permission_and_cloud_refusals_show_their_sentences(self):
-        self.write("a.png", b"aaa")
-        self.write("b.png", b"bbb")
-        self.fake.refuse[str(self.output_dir / "a.png")] = "no-permission"
-        self.fake.refuse[str(self.output_dir / "b.png")] = "cloud-sync"
-        plan = await self.review(["a.png", "b.png"])
-        self.assertEqual(plan["trash"], {"count": 0})
-        self.assertEqual(
-            plan["trash_refused"]["counts"], {"no-permission": 1, "cloud-sync": 1}
-        )
-        details = {row["detail"] for row in plan["trash_refused"]["rows"]}
-        self.assertEqual(
-            details,
-            {image_delete.REASONS["no-permission"], image_delete.REASONS["cloud-sync"]},
-        )
-
-    async def test_an_unconfirmed_move_keeps_records_and_stops_the_job(self):
-        from utils import trash
-
-        self.write("a.png", b"aaa", mtime=1_000_000)
-        self.write("b.png", b"bbb", mtime=2_000_000)
-        self.link("a.png")
-        original = self.fake.move
-
-        def move(path, roots):
-            if path.endswith("a.png"):
-                original(path, roots)  # the file did leave its folder
-                raise trash.TrashUnconfirmed("resultingItemURL=None")
-            return original(path, roots)
-
-        self.fake.move = move
-        _plan, job = await self.delete(["a.png", "b.png"], "trash")
-        self.assertEqual(job.state, "failed")
-        self.assertEqual(job.scope.get("failure"), "records-failed")
-        self.assertEqual(
-            self.details(job, "errors"), [image_delete.REASONS["trash-unconfirmed"]]
-        )
-        self.assertEqual(self.count("SELECT COUNT(*) FROM generated_images"), 1)
-        self.assertEqual(
-            self.details(job, "not_changed"), [image_delete.REASONS["not-attempted"]]
-        )
-        self.assertTrue((self.output_dir / "b.png").exists())
-
-    async def test_a_refusal_detail_is_logged(self):
-        from utils import trash
-
-        self.write("a.png", b"aaa")
-
-        def move(path, roots):
-            raise trash.TrashRefused("no-permission", "NSCocoaErrorDomain 513")
-
-        self.fake.move = move
-        with self.assertLogs("prompt_manager.api", level="INFO") as logs:
-            _plan, job = await self.delete(["a.png"], "trash")
-        self.assertEqual(
-            self.details(job, "not_changed"), [image_delete.REASONS["no-permission"]]
-        )
-        self.assertTrue(any("NSCocoaErrorDomain 513" in line for line in logs.output))
-        self.assertTrue((self.output_dir / "a.png").exists())
-
-
 class TestNativeBackend(ImageDeleteTestCase):
-    """The real backend of this OS: Windows has no trash yet (#500).
-
-    Only library mode is applied, so even on macOS and Linux no file reaches
-    a real trash.
-    """
+    """The real backend of this OS: Windows and macOS have no trash yet (#480)."""
 
     async def test_library_only_works_and_the_trash_is_unavailable_here(self):
         fake_trash.uninstall()
         self.write("a.png", b"aaa")
         self.link("a.png")
         plan = await self.review(["a.png"])
-        native_trash = sys.platform != "win32"
+        native_trash = sys.platform not in ("win32", "darwin")
         if not native_trash:
             self.assertEqual(plan["trash"], {"count": 0})
-            self.assertEqual(plan["trash_refused"]["counts"], {"windows-recycle": 1})
+            code = "windows-recycle" if sys.platform == "win32" else "no-trash"
+            self.assertEqual(plan["trash_refused"]["counts"], {code: 1})
         body = await self.post(APPLY, {"token": plan["token"], "mode": "library"}, 202)
         job = self.api._jobs.get(body["job"]["id"])
         await asyncio.wait_for(asyncio.shield(job.task), DISK_BOUND_SECONDS)

tests/test_trash_freedesktop.py

@@ -277,13 +277,12 @@ class UnavailableBackendTests(unittest.TestCase):
             self.assertEqual(caught.exception.code, trash.NO_TRASH)
             self.assertTrue(os.path.exists(path))
 
-    def test_each_platform_gets_its_backend(self):
-        from utils import trash_macos
-
-        self.assertIsInstance(trash.default_backend("win32"), trash.UnavailableBackend)
-        self.assertIsInstance(
-            trash.default_backend("darwin"), trash_macos.MacTrashBackend
-        )
+    def test_windows_and_macos_have_no_trash_yet(self):
+        for platform in ("win32", "darwin"):
+            with self.subTest(platform=platform):
+                self.assertIsInstance(
+                    trash.default_backend(platform), trash.UnavailableBackend
+                )
         self.assertIsInstance(trash.default_backend("linux"), trash.FreedesktopBackend)
 
     def test_windows_says_why_the_recycle_bin_is_not_used(self):
@@ -296,6 +295,9 @@ class UnavailableBackendTests(unittest.TestCase):
             backend.move("C:/x/a.png", ["C:/x"])
         self.assertEqual(caught.exception.code, trash.WINDOWS_RECYCLE)
         self.assertIn(trash.WINDOWS_RECYCLE, trash.CODES)
+        self.assertEqual(
+            trash.default_backend("darwin").available("/x", ["/"]), trash.NO_TRASH
+        )
 
     def test_module_functions_use_the_replaceable_backend(self):
         fake = mock.Mock()

tests/test_trash_macos.py

@@ -1,360 +0,0 @@
-"""The macOS trash backend's decisions (#499), with a fake Foundation bridge.
-
-No Objective-C is involved: the bridge is faked, ``uid``/``home``/``lstat``/
-``access`` are injected, and the only real files are temp files used for the
-"source still exists" and output-root identity checks. Runs on every OS.
-"""
-
-import os
-import shutil
-import stat
-import sys
-import tempfile
-import unittest
-from types import SimpleNamespace
-from unittest import mock
-
-sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
-
-from utils import trash, trash_macos  # noqa: E402
-from utils.trash_macos import Facts, MacTrashBackend, NSErr  # noqa: E402
-
-LOCAL = Facts(local=True, read_only=False, root_fs=True, volume="/", ubiquitous=False)
-COCOA = "NSCocoaErrorDomain"
-NO_TRASH_DIR = NSErr(COCOA, 3328, "NSPOSIXErrorDomain", 2)
-
-
-class FakeBridge:
-    """The §6.3 bridge with canned answers; ``trash`` moves nothing by default."""
-
-    def __init__(self, facts=LOCAL, trash_dir=("/Users/me/.Trash", None), result=None):
-        self.facts_value = facts
-        self.trash_dir_value = trash_dir
-        self.result = result or (True, None, None)
-        self.calls = []
-
-    def facts(self, path):
-        self.calls.append(("facts", path))
-        return self.facts_value
-
-    def trash_dir(self, path):
-        self.calls.append(("trash_dir", path))
-        return self.trash_dir_value
-
-    def trash(self, path):
-        self.calls.append(("trash", path))
-        result = self.result
-        return result(path) if callable(result) else result
-
-
-def _stat(mode=stat.S_IFREG | 0o644, uid=501, flags=0):
-    return SimpleNamespace(st_mode=mode, st_uid=uid, st_flags=flags)
-
-
-class MacBackendTestCase(unittest.TestCase):
-    def setUp(self):
-        # realpath: the backend resolves paths (/var -> /private/var on macOS, 8.3
-        # short names on Windows), so the lstat fakes must use the same spelling.
-        self.tmp = os.path.realpath(tempfile.mkdtemp(prefix="pm-mac-trash-"))
-        self.addCleanup(shutil.rmtree, self.tmp, True)
-        self.home = os.path.join(self.tmp, "home")
-        self.root = os.path.join(self.tmp, "output")
-        os.makedirs(os.path.join(self.home, ".Trash"))
-        os.makedirs(self.root)
-        self.file = os.path.join(self.root, "a.png")
-        with open(self.file, "wb") as handle:
-            handle.write(b"png")
-        self.home_trash = os.path.join(self.home, ".Trash")
-        self.stats = {}
-        self.access_ok = True
-
-    def _lstat(self, path):
-        if path in self.stats:
-            value = self.stats[path]
-            if isinstance(value, Exception):
-                raise value
-            return value
-        return os.lstat(path)
-
-    def backend(self, **bridge_args):
-        bridge_args.setdefault("trash_dir", (self.home_trash, None))
-        self.bridge = FakeBridge(**bridge_args)
-        return MacTrashBackend(
-            bridge=self.bridge,
-            uid=501,
-            home=self.home,
-            lstat=self._lstat,
-            access=lambda path, mode: self.access_ok,
-        )
-
-    def moved_to(self, destination):
-        """A ``trash`` result that really renames the file to *destination*."""
-
-        def run(path):
-            os.makedirs(os.path.dirname(destination), exist_ok=True)
-            os.rename(path, destination)
-            return True, destination, None
-
-        return run
-
-    def assert_refused(self, backend, code):
-        self.assertEqual(backend.available(self.file, [self.root]), code)
-        with self.assertRaises(trash.TrashRefused) as caught:
-            backend.move(self.file, [self.root])
-        self.assertEqual(caught.exception.code, code)
-        self.assertNotIn("trash", [name for name, _ in self.bridge.calls])
-        self.assertTrue(os.path.exists(self.file))
-        return caught.exception
-
-
-class AvailableOrderTests(MacBackendTestCase):
-    def test_a_writable_local_file_is_available(self):
-        self.assertIsNone(self.backend().available(self.file, [self.root]))
-
-    def test_network_volume_is_refused_without_trying(self):
-        facts = Facts(False, False, False, "/Volumes/nas", False)
-        self.assert_refused(self.backend(facts=facts), trash.NETWORK)
-
-    def test_unknown_locality_counts_as_network(self):
-        facts = Facts(None, False, True, "/", False)
-        self.assert_refused(self.backend(facts=facts), trash.NETWORK)
-
-    def test_read_only_volume_or_unknown_is_no_trash(self):
-        for read_only in (True, None):
-            with self.subTest(read_only=read_only):
-                facts = Facts(True, read_only, False, "/Volumes/ro", False)
-                self.assert_refused(self.backend(facts=facts), trash.NO_TRASH)
-
-    def test_ubiquitous_item_is_cloud_sync(self):
-        facts = Facts(True, False, True, "/", True)
-        self.assert_refused(self.backend(facts=facts), trash.CLOUD_SYNC)
-
-    def test_icloud_and_file_provider_folders_are_cloud_sync(self):
-        for parts in (("Library", "Mobile Documents"), ("Library", "CloudStorage")):
-            with self.subTest(parts=parts):
-                folder = os.path.join(self.home, *parts, "Box")
-                os.makedirs(folder)
-                path = os.path.join(folder, "b.png")
-                with open(path, "wb") as handle:
-                    handle.write(b"x")
-                backend = self.backend()
-                self.assertEqual(backend.available(path, [folder]), trash.CLOUD_SYNC)
-                with self.assertRaises(trash.TrashRefused):
-                    backend.move(path, [folder])
-                self.assertTrue(os.path.exists(path))
-
-    def test_found_trash_folder_inside_an_output_root_is_trash_in_root(self):
-        self.assertEqual(
-            self.backend().available(self.file, [self.home]), trash.TRASH_IN_ROOT
-        )
-
-    def test_trash_dir_error_other_than_missing_is_no_trash(self):
-        backend = self.backend(
-            trash_dir=(None, NSErr(COCOA, 3328, "NSPOSIXErrorDomain", 30))
-        )
-        self.assert_refused(backend, trash.NO_TRASH)
-
-    def test_missing_trash_on_the_startup_volume_is_no_trash(self):
-        self.assert_refused(
-            self.backend(trash_dir=(None, NO_TRASH_DIR)), trash.NO_TRASH
-        )
-
-
-class FreshVolumeTests(MacBackendTestCase):
-    def setUp(self):
-        super().setUp()
-        self.volume = os.path.join(self.tmp, "vol")
-        self.root = os.path.join(self.volume, "output")
-        os.makedirs(self.root)
-        self.file = os.path.join(self.root, "a.png")
-        with open(self.file, "wb") as handle:
-            handle.write(b"png")
-        self.facts = Facts(True, False, False, self.volume, False)
-
-    def test_missing_volume_trash_is_predicted(self):
-        backend = self.backend(facts=self.facts, trash_dir=(None, NO_TRASH_DIR))
-        self.assertIsNone(backend.available(self.file, [self.root]))
-
-    def test_prediction_refused_when_trashes_is_a_link(self):
-        os.symlink(self.tmp, os.path.join(self.volume, ".Trashes"))
-        backend = self.backend(facts=self.facts, trash_dir=(None, NO_TRASH_DIR))
-        self.assert_refused(backend, trash.NO_TRASH)
-
-    def test_predicted_trash_inside_the_volume_root_is_trash_in_root(self):
-        backend = self.backend(facts=self.facts, trash_dir=(None, NO_TRASH_DIR))
-        self.assert_refused_root = backend.available(self.file, [self.volume])
-        self.assertEqual(self.assert_refused_root, trash.TRASH_IN_ROOT)
-
-    def test_move_to_a_predicted_trash_returns_the_volume_label(self):
-        destination = os.path.join(self.volume, ".Trashes", "501", "a.png")
-        backend = self.backend(
-            facts=self.facts,
-            trash_dir=(None, NO_TRASH_DIR),
-            result=self.moved_to(destination),
-        )
-        self.assertEqual(
-            backend.move(self.file, [self.root]), ".Trashes/501 on its drive"
-        )
-        self.assertFalse(os.path.exists(self.file))
-
-
-class PermissionTests(MacBackendTestCase):
-    def test_parent_not_writable_is_no_permission(self):
-        self.access_ok = False
-        self.assert_refused(self.backend(), trash.NO_PERMISSION)
-
-    def test_immutable_or_append_flags_on_file_or_parent(self):
-        for flag in ("UF_IMMUTABLE", "SF_IMMUTABLE", "UF_APPEND", "SF_APPEND"):
-            for target in (self.file, self.root):
-                with self.subTest(flag=flag, target=target):
-                    mode = (
-                        stat.S_IFDIR | 0o755
-                        if target == self.root
-                        else stat.S_IFREG | 0o644
-                    )
-                    self.stats = {target: _stat(mode=mode, flags=getattr(stat, flag))}
-                    self.assert_refused(self.backend(), trash.NO_PERMISSION)
-
-    def test_sticky_parent_and_foreign_owner(self):
-        self.stats = {
-            self.root: _stat(mode=stat.S_IFDIR | 0o1777),
-            self.file: _stat(uid=502),
-        }
-        self.assert_refused(self.backend(), trash.NO_PERMISSION)
-
-    def test_sticky_parent_with_own_file_is_allowed(self):
-        self.stats = {self.root: _stat(mode=stat.S_IFDIR | 0o1777), self.file: _stat()}
-        self.assertIsNone(self.backend().available(self.file, [self.root]))
-
-
-class MoveOutcomeTests(MacBackendTestCase):
-    def test_success_into_the_home_trash(self):
-        destination = os.path.join(self.home_trash, "a.png")
-        backend = self.backend(result=self.moved_to(destination))
-        self.assertEqual(backend.move(self.file, [self.root]), trash.HOME_TRASH_LABEL)
-        self.assertTrue(os.path.exists(destination))
-
-    def test_cocoa_errors_map_to_codes_with_detail(self):
-        cases = {
-            3328: trash.NO_TRASH,
-            642: trash.NO_TRASH,
-            513: trash.NO_PERMISSION,
-            257: trash.NO_PERMISSION,
-            4: trash.FAILED,
-            640: trash.FAILED,
-        }
-        for code, expected in cases.items():
-            with self.subTest(code=code):
-                error = NSErr(COCOA, code, "NSOSStatusErrorDomain", -5000)
-                backend = self.backend(result=(False, None, error))
-                with self.assertRaises(trash.TrashRefused) as caught:
-                    backend.move(self.file, [self.root])
-                self.assertEqual(caught.exception.code, expected)
-                self.assertEqual(
-                    caught.exception.detail,
-                    f"NSCocoaErrorDomain {code} (NSOSStatusErrorDomain -5000)",
-                )
-                self.assertTrue(os.path.exists(self.file))
-
-    def test_other_domain_or_missing_error_is_failed(self):
-        for error in (NSErr("NSPOSIXErrorDomain", 13), None):
-            with self.subTest(error=error):
-                backend = self.backend(result=(False, None, error))
-                with self.assertRaises(trash.TrashRefused) as caught:
-                    backend.move(self.file, [self.root])
-                self.assertEqual(caught.exception.code, trash.FAILED)
-
-    def test_yes_with_the_source_still_present_is_failed(self):
-        backend = self.backend(
-            result=(True, os.path.join(self.home_trash, "a.png"), None)
-        )
-        with self.assertRaises(trash.TrashRefused) as caught:
-            backend.move(self.file, [self.root])
-        self.assertEqual(caught.exception.code, trash.FAILED)
-        self.assertTrue(os.path.exists(self.file))
-
-    def test_yes_with_no_result_path_is_unconfirmed(self):
-        def vanished(path):
-            os.rename(path, os.path.join(self.tmp, "elsewhere.png"))
-            return True, None, None
-
-        backend = self.backend(result=vanished)
-        with self.assertRaises(trash.TrashUnconfirmed):
-            backend.move(self.file, [self.root])
-
-    def test_yes_into_an_output_root_is_unconfirmed(self):
-        backend = self.backend(
-            result=self.moved_to(os.path.join(self.root, "x", "a.png"))
-        )
-        with self.assertRaises(trash.TrashUnconfirmed):
-            backend.move(self.file, [self.root])
-
-    def test_yes_into_an_unplanned_folder_is_trashed_and_logged(self):
-        other = os.path.join(self.tmp, "other-trash", "a.png")
-        backend = self.backend(result=self.moved_to(other))
-        with self.assertLogs("prompt_manager.trash", level="WARNING") as logs:
-            label = backend.move(self.file, [self.root])
-        self.assertEqual(label, os.path.realpath(os.path.dirname(other)))
-        self.assertIn(other, "\n".join(logs.output))
-
-
-class IdentityTests(MacBackendTestCase):
-    @unittest.skipUnless(hasattr(os, "symlink") and os.name == "posix", "symlinks")
-    def test_root_given_through_a_symlink_still_contains_its_trash(self):
-        alias = os.path.join(self.tmp, "home-alias")
-        os.symlink(self.home, alias)
-        self.assertEqual(
-            self.backend().available(self.file, [alias]), trash.TRASH_IN_ROOT
-        )
-
-    def test_missing_roots_are_ignored(self):
-        missing = os.path.join(self.tmp, "nope")
-        self.assertIsNone(self.backend().available(self.file, [missing, self.root]))
-
-
-class LoadingTests(unittest.TestCase):
-    def test_default_backend_for_darwin_does_not_load_the_bridge(self):
-        with mock.patch.object(trash_macos, "load_bridge") as loader:
-            backend = trash.default_backend("darwin")
-        self.assertIsInstance(backend, MacTrashBackend)
-        loader.assert_not_called()
-
-    def test_failed_load_answers_no_trash_and_is_cached_and_logged_once(self):
-        backend = MacTrashBackend(uid=501)
-        with (
-            mock.patch.object(
-                trash_macos, "load_bridge", side_effect=OSError("no Foundation")
-            ) as loader,
-            self.assertLogs("prompt_manager.trash", level="WARNING") as logs,
-        ):
-            self.assertEqual(backend.available("/x/a.png", ["/x"]), trash.NO_TRASH)
-            with self.assertRaises(trash.TrashRefused) as caught:
-                backend.move("/x/a.png", ["/x"])
-        self.assertEqual(caught.exception.code, trash.NO_TRASH)
-        self.assertEqual(loader.call_count, 1)
-        self.assertEqual(len(logs.output), 1)
-
-    def test_load_bridge_off_macos_raises(self):
-        if sys.platform == "darwin":
-            self.skipTest("the real bridge loads on macOS (native test covers it)")
-        with self.assertRaises(OSError):
-            trash_macos.load_bridge()
-
-
-class ErrorTextTests(unittest.TestCase):
-    def test_detail_without_underlying(self):
-        self.assertEqual(
-            trash_macos.error_detail(NSErr(COCOA, 4)), "NSCocoaErrorDomain 4"
-        )
-        self.assertEqual(trash_macos.error_detail(None), "no NSError")
-
-    def test_refusal_keeps_detail(self):
-        refused = trash.TrashRefused(trash.FAILED, "why")
-        self.assertEqual(
-            (refused.code, refused.detail, str(refused)), ("failed", "why", "why")
-        )
-        self.assertEqual(trash.TrashRefused(trash.FAILED).detail, "")
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/test_trash_macos_native.py

@@ -1,238 +0,0 @@
-"""Native proof of the macOS trash backend (#499 design §10.2).
-
-Runs only on macOS with ``PM_NATIVE_TRASH=macos`` (the ``native-trash`` job in
-portability.yml), because it puts files into the real Trash of the user
-running it. With the variable set, a bridge that cannot load fails the test
-rather than skipping it. Files moved into a trash folder are left there: the
-runner is disposable and this test never deletes from a trash folder.
-"""
-
-import os
-import subprocess
-import sys
-import tempfile
-import time
-import unittest
-
-sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
-
-from utils import trash  # noqa: E402
-
-NATIVE = sys.platform == "darwin" and os.environ.get("PM_NATIVE_TRASH") == "macos"
-HDIUTIL_SECONDS = 60
-# Measured: macos-15-intel kept a fresh image busy after a trash (run 40954618922).
-DETACH_WAIT_SECONDS = 30
-
-
-class SpyBridge:
-    """The real bridge, recording every ``trash`` result."""
-
-    def __init__(self, real):
-        self.real = real
-        self.results = []
-
-    def facts(self, path):
-        return self.real.facts(path)
-
-    def trash_dir(self, path):
-        return self.real.trash_dir(path)
-
-    def trash(self, path):
-        result = self.real.trash(path)
-        self.results.append(result)
-        return result
-
-
-def _hdiutil(*args):
-    return subprocess.run(
-        ["hdiutil", *args], capture_output=True, text=True, timeout=HDIUTIL_SECONDS
-    )
-
-
-def _our_open_files(mount):
-    """Files this process holds open under *mount* (``lsof -p``): the handle proof."""
-    listed = subprocess.run(
-        ["lsof", "-Fn", "-p", str(os.getpid())],
-        capture_output=True,
-        text=True,
-        timeout=HDIUTIL_SECONDS,
-    )
-    prefix = os.path.realpath(mount) + os.sep
-    return [
-        line[1:]
-        for line in listed.stdout.splitlines()
-        if line.startswith("n") and os.path.realpath(line[1:]).startswith(prefix)
-    ]
-
-
-def _detach(mount):
-    """Detach *mount*; None on success, else why (after a forced detach).
-
-    This process must hold nothing under it. A volume macOS has just written a
-    trash folder to can stay busy for a moment (Spotlight, fseventsd), so the
-    plain detach is retried until the system releases it, within a bound.
-    """
-    ours = _our_open_files(mount)
-    if ours:
-        _hdiutil("detach", mount, "-force")
-        return f"this test still held files open under {mount}: {ours}"
-    deadline = time.monotonic() + DETACH_WAIT_SECONDS
-    while True:
-        if not os.path.ismount(mount):
-            return None  # a "busy" detach can still finish after it returns
-        done = _hdiutil("detach", mount)
-        if done.returncode == 0:
-            return None
-        if time.monotonic() > deadline:
-            holders = subprocess.run(
-                ["lsof", "+f", "--", mount],
-                capture_output=True,
-                text=True,
-                timeout=HDIUTIL_SECONDS,
-            ).stdout
-            _hdiutil("detach", mount, "-force")
-            return f"detach {mount}: {done.stderr.strip()}; held by: {holders!r}"
-        time.sleep(1)  # poll the volume's release (a state), never a fixed wait
-
-
-@unittest.skipUnless(NATIVE, "macOS native trash proof (PM_NATIVE_TRASH=macos)")
-class MacTrashNativeTests(unittest.TestCase):
-    def setUp(self):
-        from utils import trash_macos
-
-        self.trash_macos = trash_macos
-        self.spy = SpyBridge(trash_macos.load_bridge())  # fails, never skips
-        self.backend = trash_macos.MacTrashBackend(bridge=self.spy)
-        self._tmp = tempfile.TemporaryDirectory(prefix="pm-native-trash-")
-        self.tmp = os.path.realpath(self._tmp.name)
-        self._mounts = []
-        self._modes = []
-
-    def tearDown(self):
-        failures = []
-        try:
-            for path, mode in reversed(self._modes):
-                os.chmod(path, mode)
-            for mount in reversed(self._mounts):
-                problem = _detach(mount)
-                if problem:
-                    failures.append(problem)
-        finally:
-            self._tmp.cleanup()
-        self.assertEqual(failures, [])
-
-    def _file(self, folder, name="a.png", content=None):
-        os.makedirs(folder, exist_ok=True)
-        path = os.path.join(folder, name)
-        with open(path, "wb") as handle:
-            handle.write(content or os.urandom(64))
-        return path
-
-    def _image(self, name, fs="APFS"):
-        image = os.path.join(self.tmp, name + ".dmg")
-        mount = os.path.join(self.tmp, name + "-mnt")
-        os.makedirs(mount)
-        made = _hdiutil("create", "-size", "40m", "-fs", fs, "-volname", name, image)
-        self.assertEqual(made.returncode, 0, made.stderr)
-        attached = _hdiutil("attach", image, "-nobrowse", "-mountpoint", mount)
-        self.assertEqual(attached.returncode, 0, attached.stderr)
-        self._mounts.append(mount)
-        return image, mount
-
-    def _assert_trashed(self, source, content, info, folder):
-        self.assertFalse(os.path.lexists(source))
-        ok, resulting, error = self.spy.results[-1]
-        self.assertTrue(ok, error)
-        self.assertEqual(os.path.realpath(os.path.dirname(resulting)), folder)
-        moved = os.lstat(resulting)
-        self.assertEqual((moved.st_dev, moved.st_ino), (info.st_dev, info.st_ino))
-        with open(resulting, "rb") as handle:
-            self.assertEqual(handle.read(), content)
-        return resulting
-
-    def _assert_kept(self, path, content):
-        with open(path, "rb") as handle:
-            self.assertEqual(handle.read(), content)
-
-    def _assert_refused(self, path, roots, code):
-        with open(path, "rb") as handle:
-            content = handle.read()
-        self.assertEqual(self.backend.available(path, roots), code)
-        with self.assertRaises(trash.TrashRefused) as caught:
-            self.backend.move(path, roots)
-        self.assertEqual(caught.exception.code, code)
-        self._assert_kept(path, content)
-
-    def test_n1_startup_volume_goes_to_the_home_trash(self):
-        root = os.path.join(self.tmp, "output")
-        source = self._file(root)
-        with open(source, "rb") as handle:
-            content = handle.read()
-        info = os.lstat(source)
-        self.assertIsNone(self.backend.available(source, [root]))
-        self.assertEqual(self.backend.move(source, [root]), trash.HOME_TRASH_LABEL)
-        home_trash = os.path.realpath(os.path.expanduser("~/.Trash"))
-        self._assert_trashed(source, content, info, home_trash)
-
-    def test_n2_a_name_clash_never_replaces(self):
-        first_root = os.path.join(self.tmp, "one")
-        second_root = os.path.join(self.tmp, "two")
-        first = self._file(first_root, "same.png", b"first")
-        second = self._file(second_root, "same.png", b"second")
-        self.backend.move(first, [first_root])
-        self.backend.move(second, [second_root])
-        (_, one, _), (_, two, _) = self.spy.results
-        self.assertNotEqual(one, two)
-        self._assert_kept(one, b"first")
-        self._assert_kept(two, b"second")
-
-    def test_n3_fresh_external_volume_uses_its_own_trash(self):
-        for fs in ("APFS", "MS-DOS FAT32"):
-            with self.subTest(fs=fs):
-                _image, mount = self._image("fresh" + fs.split()[0], fs)
-                root = os.path.join(mount, "output")
-                source = self._file(root)
-                with open(source, "rb") as handle:
-                    content = handle.read()
-                info = os.lstat(source)
-                self.assertFalse(os.path.lexists(os.path.join(mount, ".Trashes")))
-                self.assertIsNone(self.backend.available(source, [root]))
-                label = self.backend.move(source, [root])
-                uid = str(os.getuid())
-                self.assertEqual(label, f".Trashes/{uid} on its drive")
-                folder = os.path.realpath(os.path.join(mount, ".Trashes", uid))
-                self._assert_trashed(source, content, info, folder)
-
-    def test_n4_volume_root_as_output_root_is_trash_in_root(self):
-        _image, mount = self._image("rooted")
-        source = self._file(os.path.join(mount, "output"))
-        self._assert_refused(source, [mount], trash.TRASH_IN_ROOT)
-        self.assertEqual(self.spy.results, [])
-
-    def test_n5_read_only_volume_is_no_trash(self):
-        image, mount = self._image("readonly")
-        source = self._file(mount, "ro.png", b"read-only bytes")
-        self.assertIsNone(_detach(mount))
-        self._mounts.remove(mount)
-        attached = _hdiutil(
-            "attach", image, "-nobrowse", "-readonly", "-mountpoint", mount
-        )
-        self.assertEqual(attached.returncode, 0, attached.stderr)
-        self._mounts.append(mount)
-        self._assert_refused(source, [os.path.join(mount, "output")], trash.NO_TRASH)
-
-    def test_n6_unwritable_folder_is_no_permission_and_macos_agrees(self):
-        root = os.path.join(self.tmp, "locked")
-        source = self._file(root, "p.png", b"kept")
-        self._modes.append((root, 0o755))
-        os.chmod(root, 0o555)
-        self._assert_refused(source, [self.tmp], trash.NO_PERMISSION)
-        self.assertEqual(self.spy.results, [])
-        ok, _resulting, error = self.spy.real.trash(source)
-        self.assertFalse(ok)
-        self.assertEqual((error.domain, error.code), ("NSCocoaErrorDomain", 513))
-        self._assert_kept(source, b"kept")
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/vendor/ci-workflow.test.mjs

@@ -353,20 +353,6 @@ test('Windows and macOS run 3.12-3.14 on main, nightly and on demand, never on a
     assert.doesNotMatch(text, /cancel-in-progress: true|cancel-in-progress: \$/, 'no portability run is cancelled');
 });
 
-// #499: the native trash proof runs on both macOS CPUs, apart from the suite so a red suite never skips it.
-test('portability runs the native trash proof on arm64 and Intel macOS', () => {
-    const body = jobs(readWorkflow('portability.yml')).get('native-trash');
-    assert.ok(body, 'portability.yml has a native-trash job');
-    assert.deepEqual(matrixList(body, 'os'), ['macos-latest', 'macos-15-intel']);
-    assert.match(body, /^ {6}fail-fast: false$/m);
-    assert.doesNotMatch(body, /^ {4}needs:/m, 'it does not wait for the suite job');
-    assert.match(body, /^ {10}PM_NATIVE_TRASH: macos$/m);
-    assert.match(body, /^ {8}run: python -m unittest tests\.test_trash_macos_native tests\.test_trash_macos -v$/m);
-    const jobMinutes = Number(/^ {4}timeout-minutes: (\d+)$/m.exec(body)[1]);
-    const stepMinutes = Number(/^ {8}timeout-minutes: (\d+)$/m.exec(body)?.[1]);
-    assert.ok(stepMinutes > 0 && jobMinutes - stepMinutes >= 2, 'the proof step stops before the job limit');
-});
-
 // #493: the portability limit follows the rule in its comment: ceil(1.5 x the slowest of three
 // green main runs) + 3 minutes, never under 5, with those runs and times written above jobs:.
 // The times are whole-job wall times. That the runs were green and on main is the author's word.

utils/trash.py

@@ -14,11 +14,9 @@ same mount, else the file's mount's ``.Trash/$uid`` (only under a real, sticky
 ``.Trash``) or ``.Trash-$uid``. Names are reserved by creating the
 ``.trashinfo`` with ``O_EXCL`` and the file is moved with
 ``renameat2(RENAME_NOREPLACE)``; without that call the move is refused.
-macOS uses :class:`utils.trash_macos.MacTrashBackend` (#499): NSFileManager's
-``trashItemAtURL`` through ctypes, after refusing what it cannot show to be
-recoverable. Windows never uses the Recycle Bin (#500: files over its capacity
-are deleted permanently, undetectably), so it answers ``windows-recycle``
-through :class:`UnavailableBackend`.
+Windows never uses the Recycle Bin (#500: files over its capacity are deleted
+permanently, undetectably), so it answers ``windows-recycle``; macOS has no
+backend yet (#499). Both use :class:`UnavailableBackend`.
 
 ``BACKEND`` is the one replaceable seam; tests install a fake.
 """
@@ -39,22 +37,10 @@ TRASH_IN_ROOT = "trash-in-root"
 PATH_TOO_LONG = "path-too-long"
 LOCKED = "locked"
 FAILED = "failed"
-NO_PERMISSION = "no-permission"
-CLOUD_SYNC = "cloud-sync"
 # Windows never uses the Recycle Bin (#500): it can delete permanently.
 WINDOWS_RECYCLE = "windows-recycle"
 CODES = frozenset(
-    {
-        NO_TRASH,
-        NETWORK,
-        TRASH_IN_ROOT,
-        PATH_TOO_LONG,
-        LOCKED,
-        FAILED,
-        NO_PERMISSION,
-        CLOUD_SYNC,
-        WINDOWS_RECYCLE,
-    }
+    {NO_TRASH, NETWORK, TRASH_IN_ROOT, PATH_TOO_LONG, LOCKED, FAILED, WINDOWS_RECYCLE}
 )
 
 HOME_TRASH_LABEL = "home trash"
@@ -80,19 +66,6 @@ class TrashRefused(Exception):
     def __init__(self, code: str, detail: str = ""):
         super().__init__(detail or code)
         self.code = code
-        self.detail = detail
-
-
-class TrashUnconfirmed(Exception):
-    """The system reported the file moved, but not to a place shown to be a trash.
-
-    The file is no longer where it was, so this is not a refusal: the caller
-    keeps the file's records and stops (#499 design §7.4).
-    """
-
-    def __init__(self, detail: str):
-        super().__init__(detail)
-        self.detail = detail
 
 
 def parse_mountinfo(text: str) -> list[str]:
@@ -380,21 +353,14 @@ class FreedesktopBackend:
 
 
 def default_backend(platform: str = sys.platform):
-    """The backend for *platform* (#480, #499, #500).
-
-    macOS gets :class:`utils.trash_macos.MacTrashBackend`, which loads its
-    Foundation bridge only on first use (#499). Windows never uses the Recycle
-    Bin (#500).
-    """
-    if platform == "darwin":
-        from . import trash_macos
-
-        return trash_macos.MacTrashBackend()
+    """The backend for *platform*; Windows and macOS have none yet (#480)."""
     if platform == "win32" or platform.startswith("cygwin"):
         # #500, ruling (a): Windows deletes files larger than the Recycle Bin's
         # capacity permanently, and IFileOperation's PreDeleteItem guard cannot
         # tell (native probe runs 40951037575, 40953347904).
         return UnavailableBackend(WINDOWS_RECYCLE)
+    if platform == "darwin":
+        return UnavailableBackend()
     return FreedesktopBackend()
 
 

utils/trash_macos.py

@@ -1,489 +0,0 @@
-"""The macOS Trash backend (#499): NSFileManager through ctypes, or refuse.
-
-``-[NSFileManager trashItemAtURL:resultingItemURL:error:]`` moves a file into
-its volume's trash folder (``~/.Trash`` on the startup volume,
-``<volume>/.Trashes/<uid>`` elsewhere) or fails with an ``NSError``; it renames
-name clashes itself and never replaces a file. Before calling it,
-:class:`MacTrashBackend` refuses what it cannot show to be recoverable:
-network volumes, read-only volumes, iCloud Drive and File Provider sync
-folders, a trash folder inside an output root, and files macOS would refuse
-for permission. Nothing here unlinks, copies or overwrites a file.
-
-The :class:`Bridge` is the only code that touches the Objective-C runtime. It
-sends ten fixed messages, each through ``objc_msgSend`` cast to its exact
-prototype (required on arm64), checks every object for nil before a send (a
-nil can raise an Objective-C exception, which would abort the process), and
-runs each call inside its own autorelease pool. It is loaded lazily, on the
-first ``available``/``move``; when it cannot load, every file answers
-``no-trash`` as before. The facts behind these rules were measured on macOS
-14, 15 (x86_64) and 26 (docs/egm/499-macos-trash/design.md §4).
-"""
-
-from __future__ import annotations
-
-import ctypes
-import logging
-import os
-import stat
-import sys
-import threading
-from dataclasses import dataclass
-
-from . import trash
-
-logger = logging.getLogger("prompt_manager.trash")
-
-COCOA = "NSCocoaErrorDomain"
-POSIX = "NSPOSIXErrorDomain"
-FEATURE_UNSUPPORTED = 3328
-_ENOENT = 2
-# Top-level Cocoa codes of a refused move (design §7.5); anything else is "failed".
-_MOVE_CODES = {
-    FEATURE_UNSUPPORTED: trash.NO_TRASH,
-    642: trash.NO_TRASH,  # NSFileWriteVolumeReadOnlyError
-    513: trash.NO_PERMISSION,  # NSFileWriteNoPermissionError
-    257: trash.NO_PERMISSION,  # NSFileReadNoPermissionError
-}
-_LOCK_FLAGS = stat.UF_IMMUTABLE | stat.SF_IMMUTABLE | stat.UF_APPEND | stat.SF_APPEND
-_CLOUD_FOLDERS = (("Library", "Mobile Documents"), ("Library", "CloudStorage"))
-
-
-@dataclass(frozen=True)
-class NSErr:
-    domain: str
-    code: int
-    under_domain: str | None = None
-    under_code: int | None = None
-
-
-@dataclass(frozen=True)
-class Facts:
-    """Resource values of a file's URL; ``None`` means the lookup failed or was nil."""
-
-    local: bool | None
-    read_only: bool | None
-    root_fs: bool | None
-    volume: str | None
-    ubiquitous: bool | None
-
-
-def error_detail(error: NSErr | None) -> str:
-    """``"<domain> <code> (<under_domain> <under_code>)"`` for the log."""
-    if error is None:
-        return "no NSError"
-    text = f"{error.domain} {error.code}"
-    if error.under_domain is not None:
-        text += f" ({error.under_domain} {error.under_code})"
-    return text
-
-
-def refusal_code(error: NSErr | None) -> str:
-    if error is None or error.domain != COCOA:
-        return trash.FAILED
-    return _MOVE_CODES.get(error.code, trash.FAILED)
-
-
-def _within(path: str, folder: str) -> bool:
-    path = os.path.normcase(path)
-    folder = os.path.normcase(folder)
-    trimmed = folder.rstrip("\\/") or folder
-    return path == trimmed or path.startswith(trimmed.rstrip(os.sep) + os.sep)
-
-
-def _identity(path: str):
-    info = os.stat(path)
-    return info.st_dev, info.st_ino
-
-
-def inside_any_root(path: str, roots) -> bool:
-    """Whether *path* (existing or not) lies in an output root, by file identity.
-
-    Every existing folder from *path* up to the top is compared with the
-    roots' ``(st_dev, st_ino)``, so case, NFC/NFD and symlinked spellings of a
-    root all match.
-    """
-    identities = set()
-    for root in roots:
-        try:
-            identities.add(_identity(str(root)))
-        except OSError:
-            continue
-    current = os.path.abspath(path)
-    while True:
-        try:
-            if _identity(current) in identities:
-                return True
-        except OSError:
-            pass
-        parent = os.path.dirname(current)
-        if parent == current:
-            return False
-        current = parent
-
-
-def _same_folder(left: str, right: str) -> bool:
-    try:
-        return _identity(left) == _identity(right)
-    except OSError:
-        return os.path.realpath(left) == os.path.realpath(right)
-
-
-class MacTrashBackend:
-    """``available``/``move`` for macOS; the bridge and file checks are seams."""
-
-    def __init__(
-        self, bridge=None, uid=None, home=None, lstat=os.lstat, access=os.access
-    ):
-        self._bridge = bridge
-        self._load_failed = False
-        self._lock = threading.Lock()
-        if uid is None:
-            uid = os.getuid() if hasattr(os, "getuid") else -1
-        self._uid = uid
-        self._home = home
-        self._lstat = lstat
-        self._access = access
-
-    def _get_bridge(self):
-        if self._bridge is None and not self._load_failed:
-            with self._lock:
-                if self._bridge is None and not self._load_failed:
-                    try:
-                        self._bridge = load_bridge()
-                    except (OSError, AttributeError, ValueError) as error:
-                        self._load_failed = True
-                        logger.warning("The macOS trash is unavailable: %s", error)
-        if self._bridge is None:
-            raise trash.TrashRefused(trash.NO_TRASH, "Foundation bridge not loaded")
-        return self._bridge
-
-    def _home_dir(self) -> str:
-        return os.path.realpath(self._home or os.path.expanduser("~"))
-
-    def _plan(self, path: str, roots):
-        """``(trash folder, facts)`` for *path*; raises TrashRefused (design §7.2)."""
-        bridge = self._get_bridge()
-        real = os.path.realpath(path)
-        facts = bridge.facts(real)
-        if facts.local is not True:
-            raise trash.TrashRefused(trash.NETWORK)
-        if facts.read_only is not False:
-            raise trash.TrashRefused(trash.NO_TRASH, "read-only volume")
-        home = self._home_dir()
-        if facts.ubiquitous is True or any(
-            _within(real, os.path.join(home, *parts)) for parts in _CLOUD_FOLDERS
-        ):
-            raise trash.TrashRefused(trash.CLOUD_SYNC)
-        folder, error = bridge.trash_dir(real)
-        if folder is None:
-            folder = self._predict(facts, error)
-        if inside_any_root(folder, roots):
-            raise trash.TrashRefused(trash.TRASH_IN_ROOT)
-        self._check_permission(real)
-        return folder, facts
-
-    def _predict(self, facts: Facts, error: NSErr | None) -> str:
-        """``<volume>/.Trashes/<uid>`` for a volume whose trash is not made yet."""
-        missing = (
-            error is not None
-            and error.domain == COCOA
-            and error.code == FEATURE_UNSUPPORTED
-            and error.under_domain == POSIX
-            and error.under_code == _ENOENT
-        )
-        if not missing or facts.root_fs is not False or not facts.volume:
-            raise trash.TrashRefused(trash.NO_TRASH, error_detail(error))
-        trashes = os.path.join(facts.volume, ".Trashes")
-        try:
-            info = self._lstat(trashes)
-        except FileNotFoundError:
-            info = None
-        except OSError as failure:
-            raise trash.TrashRefused(trash.NO_TRASH, str(failure)) from failure
-        if info is not None and not stat.S_ISDIR(info.st_mode):
-            raise trash.TrashRefused(trash.NO_TRASH, ".Trashes is not a folder")
-        return os.path.join(trashes, str(self._uid))
-
-    def _check_permission(self, real: str) -> None:
-        parent = os.path.dirname(real)
-        if not self._access(parent, os.W_OK | os.X_OK):
-            raise trash.TrashRefused(trash.NO_PERMISSION, "folder not writable")
-        try:
-            file_info = self._lstat(real)
-            parent_info = self._lstat(parent)
-        except OSError as failure:
-            raise trash.TrashRefused(trash.FAILED, str(failure)) from failure
-        for info in (file_info, parent_info):
-            if getattr(info, "st_flags", 0) & _LOCK_FLAGS:
-                raise trash.TrashRefused(trash.NO_PERMISSION, "immutable flag")
-        if parent_info.st_mode & stat.S_ISVTX and file_info.st_uid != self._uid:
-            raise trash.TrashRefused(trash.NO_PERMISSION, "sticky folder, foreign file")
-
-    def _label(self, folder: str, facts: Facts) -> str:
-        if _same_folder(folder, os.path.join(self._home_dir(), ".Trash")):
-            return trash.HOME_TRASH_LABEL
-        real = os.path.realpath(folder)
-        volume = os.path.realpath(facts.volume) if facts.volume else None
-        if volume and volume != os.path.realpath(os.sep) and _within(real, volume):
-            relative = os.path.relpath(real, volume).replace(os.sep, "/")
-            return f"{relative} on its drive"
-        return real
-
-    def available(self, path: str, roots) -> str | None:
-        """None when *path* can be trashed, else the refusal code (no writes)."""
-        try:
-            self._plan(path, roots)
-        except trash.TrashRefused as refused:
-            return refused.code
-        except OSError:
-            return trash.NO_TRASH
-        return None
-
-    def move(self, path: str, roots) -> str:
-        """Move *path* into its trash folder and return the label (design §7.4)."""
-        try:
-            planned, facts = self._plan(path, roots)
-        except OSError as failure:
-            raise trash.TrashRefused(trash.NO_TRASH, str(failure)) from failure
-        real = os.path.realpath(path)
-        ok, resulting, error = self._get_bridge().trash(real)
-        if not ok:
-            raise trash.TrashRefused(refusal_code(error), error_detail(error))
-        if os.path.lexists(real):
-            raise trash.TrashRefused(trash.FAILED, "reported moved, still in place")
-        if not resulting or inside_any_root(os.path.dirname(resulting), roots):
-            raise trash.TrashUnconfirmed(f"{real} -> {resulting!r}")
-        folder = os.path.dirname(resulting)
-        if not _same_folder(folder, planned):
-            logger.warning(
-                "macOS trashed %s to %s, not the planned %s", real, resulting, planned
-            )
-        return self._label(folder, facts)
-
-
-# ── the Foundation bridge ────────────────────────────────────────────────
-
-_NS_TRASH_DIRECTORY = 102
-_NS_USER_DOMAIN_MASK = 1
-_LIBOBJC = "/usr/lib/libobjc.A.dylib"
-_FOUNDATION = "/System/Library/Frameworks/Foundation.framework/Foundation"
-_KEYS = (
-    "NSURLVolumeIsLocalKey",
-    "NSURLVolumeIsReadOnlyKey",
-    "NSURLVolumeIsRootFileSystemKey",
-    "NSURLVolumeURLKey",
-    "NSURLIsUbiquitousItemKey",
-    "NSUnderlyingErrorKey",
-)
-_REQUIRED_SELECTORS = (
-    ("NSURL", "fileURLWithFileSystemRepresentation:isDirectory:relativeToURL:"),
-    ("NSURL", "getResourceValue:forKey:error:"),
-    ("NSFileManager", "URLForDirectory:inDomain:appropriateForURL:create:error:"),
-    ("NSFileManager", "trashItemAtURL:resultingItemURL:error:"),
-)
-
-
-class _Nil(Exception):
-    """An Objective-C object was nil where a send needs one."""
-
-
-class Bridge:  # pragma: no cover - macOS only; the native-trash job runs it
-    """Foundation through ctypes: ``facts``, ``trash_dir`` and ``trash`` (design §6)."""
-
-    def __init__(self):
-        if sys.platform != "darwin":
-            raise OSError("the Foundation bridge exists only on macOS")
-        objc = ctypes.CDLL(_LIBOBJC)
-        foundation = ctypes.CDLL(_FOUNDATION)
-        void_p, char_p, byte = ctypes.c_void_p, ctypes.c_char_p, ctypes.c_byte
-        objc.objc_getClass.argtypes = [char_p]
-        objc.objc_getClass.restype = void_p
-        objc.sel_registerName.argtypes = [char_p]
-        objc.sel_registerName.restype = void_p
-        objc.objc_autoreleasePoolPush.argtypes = []
-        objc.objc_autoreleasePoolPush.restype = void_p
-        objc.objc_autoreleasePoolPop.argtypes = [void_p]
-        objc.objc_autoreleasePoolPop.restype = None
-        self._objc = objc
-        address = ctypes.cast(objc.objc_msgSend, void_p).value
-
-        def send(restype, *argtypes):
-            return ctypes.CFUNCTYPE(restype, void_p, void_p, *argtypes)(address)
-
-        out = ctypes.POINTER(void_p)
-        self._send_id = send(void_p)
-        self._send_responds = send(byte, void_p)
-        self._send_url = send(void_p, char_p, byte, void_p)
-        self._send_resource = send(byte, out, void_p, out)
-        self._send_bool = send(byte)
-        self._send_dir = send(void_p, ctypes.c_ulong, ctypes.c_ulong, void_p, byte, out)
-        self._send_trash = send(byte, void_p, out, out)
-        self._send_cstr = send(char_p)
-        self._send_long = send(ctypes.c_long)
-        self._send_object = send(void_p, void_p)
-        self._keys = {}
-        for name in _KEYS:
-            value = void_p.in_dll(foundation, name).value
-            if not value:
-                raise ValueError(f"Foundation constant {name} is nil")
-            self._keys[name] = value
-        for class_name, selector in _REQUIRED_SELECTORS:
-            if not self._send_responds(
-                self._class(class_name),
-                self._sel("respondsToSelector:"),
-                self._sel(selector),
-            ) and not self._instances_respond(class_name, selector):
-                raise ValueError(f"{class_name} does not respond to {selector}")
-
-    # helpers (call inside a pool) ------------------------------------------------
-
-    def _class(self, name: str):
-        value = self._objc.objc_getClass(name.encode())
-        if not value:
-            raise ValueError(f"no Objective-C class {name}")
-        return value
-
-    def _sel(self, name: str):
-        return self._objc.sel_registerName(name.encode())
-
-    def _instances_respond(self, class_name: str, selector: str) -> bool:
-        return (
-            self._send_responds(
-                self._class(class_name),
-                self._sel("instancesRespondToSelector:"),
-                self._sel(selector),
-            )
-            != 0
-        )
-
-    @staticmethod
-    def _need(value):
-        if not value:
-            raise _Nil()
-        return value
-
-    def _manager(self):
-        return self._need(
-            self._send_id(self._class("NSFileManager"), self._sel("defaultManager"))
-        )
-
-    def _url(self, path: str):
-        return self._need(
-            self._send_url(
-                self._class("NSURL"),
-                self._sel(
-                    "fileURLWithFileSystemRepresentation:isDirectory:relativeToURL:"
-                ),
-                os.fsencode(path),
-                0,
-                None,
-            )
-        )
-
-    def _path(self, url) -> str | None:
-        if not url:
-            return None
-        raw = self._send_cstr(url, self._sel("fileSystemRepresentation"))
-        return os.fsdecode(raw) if raw else None
-
-    def _text(self, obj) -> str | None:
-        if not obj:
-            return None
-        raw = self._send_cstr(obj, self._sel("UTF8String"))
-        return raw.decode("utf-8", "replace") if raw else None
-
-    def _error(self, error) -> NSErr | None:
-        if not error:
-            return None
-        domain = self._text(self._send_id(error, self._sel("domain"))) or "?"
-        code = self._send_long(error, self._sel("code"))
-        under_domain = under_code = None
-        info = self._send_id(error, self._sel("userInfo"))
-        if info:
-            under = self._send_object(
-                info, self._sel("objectForKey:"), self._keys["NSUnderlyingErrorKey"]
-            )
-            if under:
-                under_domain = self._text(self._send_id(under, self._sel("domain")))
-                under_code = self._send_long(under, self._sel("code"))
-        return NSErr(domain, code, under_domain, under_code)
-
-    def _resource(self, url, key_name: str):
-        value, error = ctypes.c_void_p(), ctypes.c_void_p()
-        ok = self._send_resource(
-            url,
-            self._sel("getResourceValue:forKey:error:"),
-            ctypes.byref(value),
-            self._keys[key_name],
-            ctypes.byref(error),
-        )
-        return value.value if ok != 0 and value.value else None
-
-    def _flag(self, url, key_name: str) -> bool | None:
-        value = self._resource(url, key_name)
-        if value is None:
-            return None
-        return self._send_bool(value, self._sel("boolValue")) != 0
-
-    def _pooled(self, work):
-        pool = self._objc.objc_autoreleasePoolPush()
-        try:
-            return work()
-        except _Nil as nil:
-            raise trash.TrashRefused(trash.NO_TRASH, "nil Objective-C object") from nil
-        finally:
-            self._objc.objc_autoreleasePoolPop(pool)
-
-    # the bridge API (design §6.3) ----------------------------------------------
-
-    def facts(self, path: str) -> Facts:
-        def work():
-            url = self._url(path)
-            return Facts(
-                local=self._flag(url, "NSURLVolumeIsLocalKey"),
-                read_only=self._flag(url, "NSURLVolumeIsReadOnlyKey"),
-                root_fs=self._flag(url, "NSURLVolumeIsRootFileSystemKey"),
-                volume=self._path(self._resource(url, "NSURLVolumeURLKey")),
-                ubiquitous=self._flag(url, "NSURLIsUbiquitousItemKey"),
-            )
-
-        return self._pooled(work)
-
-    def trash_dir(self, path: str):
-        def work():
-            error = ctypes.c_void_p()
-            folder = self._send_dir(
-                self._manager(),
-                self._sel("URLForDirectory:inDomain:appropriateForURL:create:error:"),
-                _NS_TRASH_DIRECTORY,
-                _NS_USER_DOMAIN_MASK,
-                self._url(path),
-                0,
-                ctypes.byref(error),
-            )
-            if folder:
-                return self._path(folder), None
-            return None, self._error(error.value)
-
-        return self._pooled(work)
-
-    def trash(self, path: str):
-        def work():
-            resulting, error = ctypes.c_void_p(), ctypes.c_void_p()
-            ok = self._send_trash(
-                self._manager(),
-                self._sel("trashItemAtURL:resultingItemURL:error:"),
-                self._url(path),
-                ctypes.byref(resulting),
-                ctypes.byref(error),
-            )
-            if ok != 0:
-                return True, self._path(resulting.value), None
-            return False, None, self._error(error.value)
-
-        return self._pooled(work)
-
-
-def load_bridge() -> Bridge:
-    """The real Foundation bridge; raises OSError/ValueError where it cannot load."""
-    return Bridge()