test(collections): wait for the restore job, not 5 s; no hard-coded saved-view id (#520)

The restore test waited 5 s for a whole restore. Under load (webkit,
merge queue of #514) the restore outlasted it and kept running into the
next tests on the worker, replacing their database. It now waits for
the restore job's event stream to end, as the backups spec does. The
route-change test used saved view 5, which a worker can have made.

.github/workflows/portability.yml

@@ -26,6 +26,9 @@ 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 }}
@@ -76,3 +79,28 @@ 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

@@ -53,6 +53,14 @@ stdin open and the server (`--exit-on-stdin-eof`) shuts down cleanly, removing i
 temp root, when that pipe closes, so a worker killed without teardown leaves no
 server behind. A test that fails gets the stderr its server wrote during the test
 attached to the report as `server-stderr` (an aiohttp traceback behind a 500).
+Each worker also records its browser's own log (#513, `tools/vendor/browser-logs.mjs`):
+Playwright's `pw:browser` channel, kept as a 300-line tail without printing it
+unless `DEBUG` asks for it. A browser that exits abnormally writes one
+`pm-browser-exit| browser pid=<n> exited: exitCode=<c>, signal=<s>` line to the job
+log, and a test that fails on a closed or crashed target ("Target page, context or
+browser has been closed") gets the tail attached as `browser-stderr.txt`, which the
+uploaded test-results carry. `tests/browser/browser-logs.spec.mjs` kills a browser of
+its own on purpose, so each browser job logs that line once for it.
 A `/__test__/.../reset` still running after 2 s writes a report to the job log,
 passing test or not: every thread's stack taken while it waits, the default
 executor's threads and queue, the open connections and WAL size, then the total
@@ -2487,14 +2495,30 @@ 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`). Windows
+  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
   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."). 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.
+  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).
 
 ### Prompt library cards and recorded deviations (#249 PR E)
 

py/api/image_delete.py

@@ -65,6 +65,9 @@ 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 = {
@@ -83,6 +86,17 @@ 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."
@@ -327,7 +341,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, False
+        changed, failed = 0, None
         for done, target in enumerate(targets, start=1):
             if failed:
                 self._record_delete(job, "not_changed", target, "not-attempted")
@@ -336,7 +350,8 @@ class ImageDeleteMixin:
                 self._delete_one_image, db, target, mode, roots
             )
             if group == "errors":
-                failed = True
+                failed = code or RECORDS_FAILED_CODE
+                text = REASONS[code] if code else RECORDS_FAILED
                 job.record(
                     "errors",
                     {
@@ -344,9 +359,9 @@ class ImageDeleteMixin:
                         "root": target["root_label"],
                         "root_index": target["root_index"],
                         "root_scope": target["root_scope"],
-                        "detail": RECORDS_FAILED,
+                        "detail": text,
                     },
-                    f"! {target['root_label']}/{target['rel']} · {RECORDS_FAILED}",
+                    f"! {target['root_label']}/{target['rel']} · {text}",
                 )
             elif group == done_group:
                 changed += 1
@@ -365,8 +380,10 @@ class ImageDeleteMixin:
         job.log_line(line)
         job.update(progress=100, status_line=f"{line} · {backup_note(backup)}")
         if failed:
-            job.scope["failure"] = "records-failed"
-            job.publish({"type": "error", "code": "records-failed", "message": line})
+            # "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.finish("failed")
 
     @staticmethod
@@ -405,7 +422,19 @@ 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/browser-logs.spec.mjs

@@ -0,0 +1,26 @@
+// #513: a worker records its browser's own log, so a browser that dies between two tests leaves its exit and stderr
+// behind instead of only "Target page, context or browser has been closed" on the next test.
+import { test, expect } from './fixtures.mjs';
+import { isTargetGone } from '../../tools/vendor/browser-logs.mjs';
+
+test('the worker records the launch of the browser its tests use', async ({ browser, browserLogs }) => {
+    expect(browser.isConnected()).toBe(true);
+    expect(browserLogs.launches().length).toBeGreaterThan(0);
+});
+
+test('a killed browser leaves its signal and stderr in the worker log', async ({ playwright, browserName, browserLogs }) => {
+    const before = browserLogs.launches();
+    const victim = await playwright[browserName].launch();
+    const pid = browserLogs.launches().find(item => !before.includes(item));
+    expect(pid, 'the new browser\'s launch was recorded').toBeGreaterThan(0);
+    const disconnected = new Promise(resolve => victim.once('disconnected', resolve));
+    // Playwright starts a browser in a process group of its own; WebKit's launched pid is a wrapper script.
+    process.kill(-pid, 'SIGKILL');
+    await disconnected;
+    const error = await victim.newContext().then(() => null, item => item);
+    expect(error, 'a dead browser refuses a new context').not.toBeNull();
+    expect(isTargetGone(error.message)).toBe(true);
+    // The pipe can close before the launcher has logged the exit.
+    await expect.poll(() => browserLogs.exits()).toContainEqual({ pid, exitCode: null, signal: 'SIGKILL' });
+    expect(browserLogs.tail()).toContain(`[pid=${pid}] <process did exit: exitCode=null, signal=SIGKILL>`);
+});

tests/browser/fixtures.mjs

@@ -4,13 +4,29 @@
 // points at that worker's server. A server that dies fails the running test
 // with its stderr tail, and the next test in the worker gets a fresh server.
 // A test that fails gets the stderr its server wrote during it attached as
-// `server-stderr`.
+// `server-stderr`. The worker also records its browser's own log (#513): a
+// browser that exits abnormally prints one `pm-browser-exit|` line, and a test
+// that fails on a closed or crashed target gets that log attached as
+// `browser-stderr`.
 import { test as base, expect } from '../../tools/vendor/node_modules/@playwright/test/index.mjs';
 import { attachServerStderr, createServerSupervisor, startBrowserServer } from '../../tools/vendor/start-browser-server.mjs';
+import { attachBrowserLogs, createBrowserLogRecorder } from '../../tools/vendor/browser-logs.mjs';
 
 const SERVER_FIXTURE_TIMEOUT_MS = 90_000;
 
 export const test = base.extend({
+    browserLogs: [async ({}, use) => {
+        const recorder = createBrowserLogRecorder();
+        try {
+            await use(recorder);
+        } finally {
+            recorder.uninstall();
+        }
+    }, { scope: 'worker', auto: true }],
+    browserLogsOnFailure: [async ({ browserLogs }, use, testInfo) => {
+        await use();
+        await attachBrowserLogs(testInfo, browserLogs);
+    }, { auto: true }],
     testServer: [async ({}, use) => {
         const supervisor = createServerSupervisor({ start: () => startBrowserServer() });
         try {

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/browser/workspace-collections.spec.mjs

@@ -5,6 +5,7 @@ import { test, expect } from './fixtures.mjs';
 // page `#/collection/<id>` with its Edit (D137, concept render a1186–1189) and saved views on a collection (D140).
 const workspace = '/prompt_manager/workspace';
 const API = '/prompt_manager/collections';
+const RESTORE_APPLY_URL = '/prompt_manager/restore/apply';
 const main = page => page.locator('#workspace-main');
 const heading = page => main(page).getByRole('heading', { level: 1 });
 const breadcrumb = page => page.locator('#workspace-breadcrumb');
@@ -305,7 +306,12 @@ test('a finished restore reads the collections again', async ({ page, request })
         const chooser = page.waitForEvent('filechooser');
         await page.getByRole('button', { name: 'Review restore…', exact: true }).click();
         await (await chooser).setFiles({ name: 'prompts_backup_20261027_093000.db', mimeType: 'application/octet-stream', buffer: bytes });
+        // Wait for the restore job itself, as workspace-system-backups.spec.mjs does: under load a restore outlasts the
+        // 5 s expect default, and a restore still running after this test replaces the next tests' database (#520).
+        const applied = page.waitForResponse(response => response.url().endsWith(RESTORE_APPLY_URL));
         await page.locator('dialog#modal[open]').getByRole('button', { name: 'Restore', exact: true }).click();
+        const { job } = await (await applied).json();
+        await expect(await request.get(`/prompt_manager/jobs/${job.id}/events`)).toBeOK();
         await expect(toast(page)).toHaveText(/^Database restore finished · /);
         // The list was read again without a reload.
         await expect(rows(page)).toHaveCount(2);
@@ -480,7 +486,7 @@ test('a rename that settles after its dialog closed leaves the next dialog open
 });
 
 test('a page that says a collection or view could not be opened takes focus after a route change (review L3)', async ({ page, request }) => {
-    await seed(request);
+    const manifest = await seed(request);
     await page.goto(`${workspace}#/images`);
     await expect(heading(page)).toHaveText('Images');
     await page.evaluate(() => { window.location.hash = '#/collection/999999'; });
@@ -499,7 +505,8 @@ test('a page that says a collection or view could not be opened takes focus afte
     await page.route('**/prompt_manager/saved-views', route => route.request().method() === 'GET'
         ? route.fulfill({ status: 500, contentType: 'application/json', body: '{"success":false,"error":"x"}' })
         : route.fallback());
-    await page.evaluate(() => { window.location.hash = '#/saved/5'; });
+    // An id the seed never made (its one view is manifest.saved_view), whatever ids earlier tests used (#520).
+    await page.evaluate(id => { window.location.hash = `#/saved/${id}`; }, manifest.saved_view.id + 1000);
     await expect(description(page)).toHaveText('Saved views could not be loaded. Your saved data has not changed.');
     await expect(heading(page)).toBeFocused();
 });

tests/browser/workspace-review.spec.mjs

@@ -221,6 +221,33 @@ test('tag image from review saves tags and the item remains pending', async ({ p
     await expect(dialog).toHaveCount(0);
 });
 
+test('#515: tags typed before the image observation answers are saved', async ({ page, request }) => {
+    await seed(request, 1);
+    await open(page);
+    // Hold the dialog's observation so the typing lands while it is pending (a slow host or WebKit in CI).
+    let release;
+    const held = new Promise(resolve => { release = resolve; });
+    const matcher = '**/prompt_manager/image-annotations/observe';
+    let reachedObserve;
+    const reached = new Promise(resolve => { reachedObserve = resolve; });
+    await page.route(matcher, async route => { reachedObserve(); await held; await route.continue(); });
+    await panel(page).getByRole('button', { name: 'Tag image', exact: true }).click();
+    const dialog = page.getByRole('dialog', { name: 'Tag image', exact: true });
+    await reached;
+    const field = dialog.getByRole('textbox', { name: 'Image tags · edit manually', exact: true });
+    await field.fill('dunes');
+    const save = dialog.getByRole('button', { name: 'Save manual tags', exact: true });
+    await expect(save).toHaveAttribute('aria-disabled', 'true');
+    const answered = page.waitForResponse(item => item.url().endsWith('/image-annotations/observe'));
+    release();
+    await answered;
+    await expect(save).not.toHaveAttribute('aria-disabled', 'true');
+    await expect(field).toHaveValue('dunes');
+    await save.click();
+    await expect(toast(page)).toHaveText('1 tags saved for review_0000.png. Prompt association still awaits review.');
+    await page.unroute(matcher);
+});
+
 test('an empty queue shows the empty notice only', async ({ page, request }) => {
     await seed(request, 0);
     await page.goto(review);

tests/js/workspace-image-tagging-state.test.js

@@ -876,3 +876,63 @@ test('retry after a changed storage, file or expired capture observes again and
     await settle();
     assert.equal(c.state.manual, 'Minimal, warm_light');
 });
+
+test('#515: text typed while the first observation is pending is kept, after the saved tags', async () => {
+    const { createImageTaggingController } = await load();
+    const answers = [deferred(), deferred()];
+    let n = 0;
+    const h = harness(defaultRoutes({ [`POST ${OBSERVE}`]: () => answers[n++].promise }));
+    // An untagged file: the typed text is the whole draft and Save sends it.
+    const bare = createImageTaggingController({ api: h.api, target: TARGET, source: 'review' });
+    const first = bare.open();
+    bare.setManual('dunes');
+    answers[0].resolve(response(200, observed({ tags: [], tag_revision: 0 })));
+    await first;
+    await settle();
+    assert.equal(bare.state.phase, 'ready');
+    assert.equal(bare.state.manual, 'dunes');
+    assert.equal(bare.state.statusText, '', 'nothing changed elsewhere');
+    await bare.save();
+    await settle();
+    assert.deepEqual(h.of('PUT', TAGS).at(-1).body.tags, ['dunes']);
+    // A tagged file: its saved tags lead, so a Save never drops tags the user had not seen yet.
+    const tagged = createImageTaggingController({ api: h.api, target: TARGET });
+    const second = tagged.open();
+    tagged.setManual('dunes, minimal');
+    answers[1].resolve(response(200, observed()));
+    await second;
+    await settle();
+    assert.equal(tagged.state.manual, 'Minimal, warm_light, dunes');
+    assert.equal(tagged.state.statusText, '');
+});
+
+test('#515: a kept draft takes the text typed while the reopen observes', async () => {
+    const { c } = await opened({ [`PUT ${TAGS}`]: () => fail(409, 'file-changed', 'The file changed.') });
+    c.setManual('kept draft');
+    await c.save();
+    await settle();
+    const reopened = c.retry();
+    c.setManual('kept draft, later');
+    await reopened;
+    await settle();
+    assert.equal(c.state.phase, 'ready');
+    assert.equal(c.state.manual, 'kept draft, later');
+});
+
+test('#515: text typed before Retry after a failed first observation follows the saved tags', async () => {
+    const { createImageTaggingController } = await load();
+    let n = 0;
+    const h = harness(defaultRoutes({
+        [`POST ${OBSERVE}`]: () => (n++ === 0 ? fail(500, 'internal', 'boom') : response(200, observed())),
+    }));
+    const c = createImageTaggingController({ api: h.api, target: TARGET });
+    await c.open();
+    await settle();
+    assert.equal(c.state.phase, 'observe-error');
+    c.setManual('dunes');
+    await c.retry();
+    await settle();
+    assert.equal(c.state.phase, 'ready');
+    assert.equal(c.state.manual, 'Minimal, warm_light, dunes');
+    assert.equal(c.state.statusText, '');
+});

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,19 +320,86 @@ 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 and macOS have no trash yet (#480)."""
+    """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.
+    """
 
     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 not in ("win32", "darwin")
+        native_trash = sys.platform != "win32"
         if not native_trash:
             self.assertEqual(plan["trash"], {"count": 0})
-            code = "windows-recycle" if sys.platform == "win32" else "no-trash"
-            self.assertEqual(plan["trash_refused"]["counts"], {code: 1})
+            self.assertEqual(plan["trash_refused"]["counts"], {"windows-recycle": 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_thumbnail_jobs.py

@@ -490,16 +490,25 @@ class TestThumbnailJob(ThumbnailJobCase):
         real_write = images_module._write_image_thumbnail
         job_ids = []
         written = []
+        # #504: the worker can write its first preview before the response
+        # carrying the job id is read. The first write waits until the test
+        # has stored the id, so the cancel always lands after exactly one write.
+        first_written, id_stored = threading.Event(), threading.Event()
+        self.addCleanup(id_stored.set)
 
         def counting(src, dst, *args, **kwargs):
             real_write(src, dst, *args, **kwargs)
             written.append(dst)
             if len(written) == 1:
+                first_written.set()
+                id_stored.wait(10)
                 self.api._jobs.cancel(job_ids[0])
 
         with patch.object(images_module, "_write_image_thumbnail", counting):
             _, body = await self._start()
+            await self._wait(first_written)
             job_ids.append(body["job"]["id"])
+            id_stored.set()
             job = await self._finished(body["job"]["id"])
         self.assertEqual(job.state, "cancelled")
         self.assertEqual(job.status_line, "Cancelled · 1 preview written")

tests/test_trash_freedesktop.py

@@ -277,12 +277,13 @@ class UnavailableBackendTests(unittest.TestCase):
             self.assertEqual(caught.exception.code, trash.NO_TRASH)
             self.assertTrue(os.path.exists(path))
 
-    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
-                )
+    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
+        )
         self.assertIsInstance(trash.default_backend("linux"), trash.FreedesktopBackend)
 
     def test_windows_says_why_the_recycle_bin_is_not_used(self):
@@ -295,9 +296,6 @@ 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

@@ -0,0 +1,360 @@
+"""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

@@ -0,0 +1,238 @@
+"""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/browser-logs.test.mjs

@@ -0,0 +1,127 @@
+import test from 'node:test';
+import assert from 'node:assert/strict';
+import { mkdtempSync, readFileSync, rmSync } from 'node:fs';
+import { tmpdir } from 'node:os';
+import { join } from 'node:path';
+import { attachBrowserLogs, BROWSER_EXIT_PREFIX, createBrowserLogRecorder, isAbnormalExit, isTargetGone, parseExitLine,
+    playwrightDebug } from '../../tools/vendor/browser-logs.mjs';
+
+const debug = playwrightDebug();
+
+/** Runs `body` with DEBUG set to `namespaces` and restores the caller's setting afterwards. */
+function withDebug(namespaces, body) {
+    const saved = debug.disable();
+    const savedLog = debug.log;
+    if (namespaces) debug.enable(namespaces);
+    try {
+        return body();
+    } finally {
+        debug.log = savedLog;
+        debug.disable();
+        if (saved) debug.enable(saved);
+    }
+}
+
+const outputDir = mkdtempSync(join(tmpdir(), 'pm-browser-logs-'));
+test.after(() => rmSync(outputDir, { recursive: true, force: true }));
+
+function testInfo({ status = 'failed', errors = [] } = {}) {
+    const attachments = [];
+    const dir = mkdtempSync(join(outputDir, 'test-'));
+    return { status, expectedStatus: 'passed', errors, attachments, outputPath: name => join(dir, name),
+        attach: async (name, options) => { attachments.push({ name, ...options }); } };
+}
+
+test('parses the exit line Playwright logs for a browser process', () => {
+    assert.deepEqual(parseExitLine('[pid=7124] <process did exit: exitCode=null, signal=SIGKILL> +21ms'),
+        { pid: 7124, exitCode: null, signal: 'SIGKILL' });
+    assert.deepEqual(parseExitLine('[pid=9] <process did exit: exitCode=0, signal=null>'), { pid: 9, exitCode: 0, signal: null });
+    assert.deepEqual(parseExitLine('[pid=9] <process did exit: exitCode=11, signal=null>'), { pid: 9, exitCode: 11, signal: null });
+    assert.equal(parseExitLine('[pid=9][err] JavaScript warning'), null);
+    assert.equal(isAbnormalExit({ pid: 9, exitCode: 0, signal: null }), false);
+    assert.equal(isAbnormalExit({ pid: 9, exitCode: null, signal: 'SIGKILL' }), true);
+    assert.equal(isAbnormalExit({ pid: 9, exitCode: 139, signal: null }), true);
+});
+
+test('recognises the errors of a page, context or browser that went away', () => {
+    for (const message of ['Error: browser.newContext: Target page, context or browser has been closed',
+        'Protocol error (Page.navigate): Target closed', 'page.goto: Page crashed', 'Target crashed']) {
+        assert.equal(isTargetGone(message), true, message);
+    }
+    for (const message of ['expect(locator).toBeVisible() failed', undefined, 'Test timeout of 30000ms exceeded.']) {
+        assert.equal(isTargetGone(message), false, String(message));
+    }
+});
+
+test('records the browser channel into a bounded tail without printing it', () => withDebug('', () => {
+    const printed = [];
+    debug.log = (...args) => printed.push(args.join(' '));
+    const written = [];
+    const recorder = createBrowserLogRecorder({ debug, limit: 3, write: text => written.push(text) });
+    const browser = debug('pw:browser');
+    browser('<launched> pid=5');
+    for (const n of [1, 2, 3, 4]) browser(`[pid=5][err] line ${n}`);
+    assert.deepEqual(printed, [], 'the browser channel stays quiet when DEBUG did not ask for it');
+    assert.match(recorder.tail(), /line 2[\s\S]*line 3[\s\S]*line 4/);
+    assert.doesNotMatch(recorder.tail(), /line 1/);
+    assert.deepEqual(written, []);
+    assert.deepEqual(recorder.launches(), [5], 'launches outlive the bounded tail');
+    browser('[pid=5] <process did exit: exitCode=null, signal=SIGKILL>');
+    assert.deepEqual(recorder.exits(), [{ pid: 5, exitCode: null, signal: 'SIGKILL' }]);
+    assert.deepEqual(written, [`${BROWSER_EXIT_PREFIX}browser pid=5 exited: exitCode=null, signal=SIGKILL\n`]);
+    browser('[pid=6] <process did exit: exitCode=0, signal=null>');
+    assert.equal(written.length, 1, 'a clean exit prints nothing');
+    recorder.uninstall();
+    assert.equal(debug.enabled('pw:browser'), false, 'uninstall restores the caller\'s DEBUG');
+}));
+
+test('keeps printing the channels the user enabled, the browser one included', () => withDebug('pw:api,pw:browser', () => {
+    const printed = [];
+    debug.log = function (...args) { printed.push(`${this.namespace} ${args.join(' ')}`); };
+    const recorder = createBrowserLogRecorder({ debug, write: () => {} });
+    debug('pw:api')('call');
+    debug('pw:browser')('[pid=1][out] hello');
+    debug('pw:protocol')('not enabled');
+    assert.equal(printed.length, 2);
+    assert.match(printed[0], /^pw:api /);
+    assert.match(printed[1], /^pw:browser /);
+    assert.doesNotMatch(recorder.tail(), /call/);
+    recorder.uninstall();
+    assert.equal(debug.enabled('pw:api'), true);
+    assert.equal(debug.enabled('pw:browser'), true);
+}));
+
+test('attaches the tail and the exit only to a test whose target went away', async () => withDebug('', async () => {
+    debug.log = () => {};
+    const recorder = createBrowserLogRecorder({ debug, write: () => {} });
+    const browser = debug('pw:browser');
+    browser('\u001b[30m[pid=8][err] Crash Annotation GraphicsCriticalError\u001b[0m');
+    browser('[pid=8] <process did exit: exitCode=null, signal=SIGKILL>');
+
+    const passed = testInfo({ status: 'passed' });
+    assert.equal(await attachBrowserLogs(passed, recorder), false);
+    const assertion = testInfo({ errors: [{ message: 'expect(locator).toBeVisible() failed' }] });
+    assert.equal(await attachBrowserLogs(assertion, recorder), false);
+    assert.deepEqual(assertion.attachments, []);
+
+    const closed = testInfo({ errors: [{ message: 'Error: browser.newContext: Target page, context or browser has been closed' }] });
+    assert.equal(await attachBrowserLogs(closed, recorder), true);
+    assert.equal(closed.attachments.length, 1);
+    const [{ name, path, contentType }] = closed.attachments;
+    assert.equal(name, 'browser-stderr');
+    assert.equal(contentType, 'text/plain');
+    const body = readFileSync(path, 'utf8');
+    assert.match(body, /^browser pid=8 exited: exitCode=null, signal=SIGKILL\n/);
+    assert.match(body, /\[pid=8\]\[err\] Crash Annotation GraphicsCriticalError/);
+    assert.doesNotMatch(body, /\u001b/, 'the attached log has no terminal colours');
+    recorder.uninstall();
+}));
+
+test('says so when the worker logged no abnormal exit', async () => withDebug('', async () => {
+    debug.log = () => {};
+    const recorder = createBrowserLogRecorder({ debug, write: () => {} });
+    const closed = testInfo({ errors: [{ message: 'x', stack: 'Error: Target closed\n    at y' }] });
+    assert.equal(await attachBrowserLogs(closed, recorder), true);
+    assert.match(readFileSync(closed.attachments[0].path, 'utf8'), /^no abnormal browser exit was logged in this worker\n/);
+    recorder.uninstall();
+}));

tests/vendor/ci-workflow.test.mjs

@@ -353,6 +353,20 @@ 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.

tools/vendor/browser-logs.mjs

@@ -0,0 +1,91 @@
+// What a browser printed before it died (#513). Playwright writes a browser's stderr, its launch and its exit
+// (`<process did exit: exitCode=…, signal=…>`) only to its `pw:browser` debug channel, which is off unless DEBUG is
+// set, so a browser that dies between two tests leaves the next test only "Target page, context or browser has been
+// closed". A worker records that channel into a bounded tail, prints one line when a browser exits abnormally and
+// attaches the tail to a test that fails on a closed or crashed target. Nothing is retried or hidden.
+import { createRequire } from 'node:module';
+import { writeFile } from 'node:fs/promises';
+import { format, stripVTControlCharacters } from 'node:util';
+
+const BROWSER_NAMESPACE = 'pw:browser';
+export const BROWSER_LOG_LIMIT = 300;
+export const BROWSER_EXIT_PREFIX = 'pm-browser-exit| ';
+const LAUNCH_LINE = /<launched> pid=(\d+)/;
+const EXIT_LINE = /\[pid=(\d+)\] <process did exit: exitCode=([^,]+), signal=([^>]+)>/;
+// The client's TargetClosedError, a protocol error of a closed or crashed target, and a crashed page.
+const TARGET_GONE = /Target page, context or browser has been closed|Target (?:closed|crashed)|Browser closed|browser has disconnected|Page crashed/i;
+
+/** The `debug` module Playwright logs through (the instance this worker's tests launch browsers with). */
+export function playwrightDebug() {
+    return createRequire(import.meta.url)('playwright-core/lib/utilsBundle').debug;
+}
+
+/** The pid, exit code and signal of a `process did exit` line, or null. */
+export function parseExitLine(line) {
+    const match = EXIT_LINE.exec(line);
+    if (!match) return null;
+    const value = text => (text === 'null' ? null : text);
+    const code = value(match[2]);
+    return { pid: Number(match[1]), exitCode: code === null ? null : Number(code), signal: value(match[3]) };
+}
+
+export const isAbnormalExit = exit => exit !== null && (exit.exitCode !== 0 || exit.signal !== null);
+
+/** True for an error message that says the page, context or browser under the test went away. */
+export const isTargetGone = message => TARGET_GONE.test(String(message ?? ''));
+
+/**
+ * Records `pw:browser` into a tail of at most `limit` lines. Other debug channels keep printing as before; the
+ * browser channel prints only when the user's own DEBUG asked for it. `write` gets one line per abnormal exit.
+ */
+export function createBrowserLogRecorder({ debug = playwrightDebug(), limit = BROWSER_LOG_LIMIT,
+    write = text => process.stderr.write(text) } = {}) {
+    const lines = [];
+    const launches = [];
+    const exits = [];
+    const userWantsBrowser = debug.enabled(BROWSER_NAMESPACE);
+    const previous = debug.disable();
+    const original = debug.log;
+    debug.log = function record(...args) {
+        if (this?.namespace !== BROWSER_NAMESPACE || userWantsBrowser) {
+            (original ?? (() => {})).apply(this, args);
+        }
+        if (this?.namespace !== BROWSER_NAMESPACE) return;
+        const line = stripVTControlCharacters(format(...args));
+        lines.push(line);
+        if (lines.length > limit) lines.splice(0, lines.length - limit);
+        const launched = LAUNCH_LINE.exec(line);
+        if (launched) launches.push(Number(launched[1]));
+        const exit = parseExitLine(line);
+        if (!exit) return;
+        exits.push(exit);
+        if (isAbnormalExit(exit)) {
+            write(`${BROWSER_EXIT_PREFIX}browser pid=${exit.pid} exited: exitCode=${exit.exitCode}, signal=${exit.signal}\n`);
+        }
+    };
+    debug.enable([previous, BROWSER_NAMESPACE].filter(Boolean).join(','));
+    return {
+        tail: () => lines.join('\n'),
+        launches: () => [...launches],
+        exits: () => [...exits],
+        uninstall() {
+            debug.log = original;
+            debug.disable();
+            if (previous) debug.enable(previous);
+        },
+    };
+}
+
+/** Attaches the browser log tail to a test that failed because its page, context or browser went away. */
+export async function attachBrowserLogs(testInfo, recorder) {
+    if (testInfo.status === testInfo.expectedStatus) return false;
+    if (!testInfo.errors.some(error => isTargetGone(error.message) || isTargetGone(error.stack))) return false;
+    const exits = recorder.exits().filter(isAbnormalExit)
+        .map(exit => `browser pid=${exit.pid} exited: exitCode=${exit.exitCode}, signal=${exit.signal}`);
+    const summary = exits.length ? exits.join('\n') : 'no abnormal browser exit was logged in this worker';
+    // A file, so the log travels with the uploaded test-results (the list reporter shortens a long body).
+    const path = testInfo.outputPath('browser-stderr.txt');
+    await writeFile(path, `${summary}\n--- last ${BROWSER_LOG_LIMIT} pw:browser lines ---\n${recorder.tail()}\n`);
+    await testInfo.attach('browser-stderr', { path, contentType: 'text/plain' });
+    return true;
+}

utils/trash.py

@@ -14,9 +14,11 @@ 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.
-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`.
+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`.
 
 ``BACKEND`` is the one replaceable seam; tests install a fake.
 """
@@ -37,10 +39,22 @@ 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, WINDOWS_RECYCLE}
+    {
+        NO_TRASH,
+        NETWORK,
+        TRASH_IN_ROOT,
+        PATH_TOO_LONG,
+        LOCKED,
+        FAILED,
+        NO_PERMISSION,
+        CLOUD_SYNC,
+        WINDOWS_RECYCLE,
+    }
 )
 
 HOME_TRASH_LABEL = "home trash"
@@ -66,6 +80,19 @@ 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]:
@@ -353,14 +380,21 @@ class FreedesktopBackend:
 
 
 def default_backend(platform: str = sys.platform):
-    """The backend for *platform*; Windows and macOS have none yet (#480)."""
+    """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()
     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

@@ -0,0 +1,489 @@
+"""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()

web/js/workspace/image-tagging-state.mjs

@@ -317,6 +317,9 @@ export function createImageTaggingController({api, target, source = 'images', on
     // A save in flight: closing hands the capture release to its completion.
     let saving = false;
     let observedOnce = false;
+    // #515: the user typed since the last observation started (or before the first one answered), so the answer
+    // keeps that text.
+    let typed = false;
 
     function set(patch) {
         if (closed) return;
@@ -345,7 +348,8 @@ export function createImageTaggingController({api, target, source = 'images', on
     async function open({keepDraft = false} = {}) {
         if (closed) return;
         const serial = ++openSerial;
-        const draft = keepDraft ? state.manual : null;
+        // A reopen after an observation starts from its answer; text typed before any observation is kept (#515).
+        if (observedOnce) typed = false;
         const seenRevision = state.tagRevision;
         stopGenerating();
         releaseCapture();
@@ -385,15 +389,19 @@ export function createImageTaggingController({api, target, source = 'images', on
         const savedModel = !chosen && !modelPicked && read.model ? {model: read.model} : {};
         observedOnce = true;
         const {width, height, media_type: mediaType, preview_url: previewUrl, generate} = value.capture;
-        // A kept draft stays as typed unless another writer changed the set meanwhile: then the set leads it (I2).
-        const changed = draft !== null && value.tag_revision !== seenRevision;
+        // The draft is read now, so text typed while observing is kept (#515). A kept draft stays as typed unless
+        // another writer changed the set meanwhile: then the set leads it (I2). Text typed before the user saw any
+        // saved tags follows them, so a Save never drops tags it never showed.
+        const draft = keepDraft || typed ? state.manual : null;
+        const changed = keepDraft && value.tag_revision !== seenRevision;
+        const unseen = draft !== null && !keepDraft;
         const reason = generate.supported ? '' : (generate.reason || '');
         set({
             phase: 'ready',
             image: {filename: value.filename, title: value.title, width, height, mediaType, previewUrl,
                 dimensions: width && height ? `${width} × ${height}` : null},
             savedTags: value.tags.slice(), tagRevision: value.tag_revision,
-            manual: draft === null ? value.tags.join(', ') : changed ? mergeCurrent(value.tags, draft) : draft,
+            manual: draft === null ? value.tags.join(', ') : changed || unseen ? mergeCurrent(value.tags, draft) : draft,
             draftRevision: state.draftRevision + 1,
             ...savedModel, options: {...defaults, ...chosen?.options}, optionsRevision: state.optionsRevision + 1,
             generateSupported: generate.supported, generateReason: generate.reason,
@@ -402,6 +410,7 @@ export function createImageTaggingController({api, target, source = 'images', on
     }
 
     function setManual(text) {
+        typed = true;
         set({manual: typeof text === 'string' ? text : ''});
     }