_X11ClipboardMonitor

def __init__(self, display_name: Optional[str] = None) -> None:
    self._d = display.Display(display_name, blocking_timeout=INPUT_X_REPLY_TIMEOUT_S)
    self._cmd_r = self._cmd_w = -1
    try:
        self._build()
    except BaseException:
        self._release_resources()
        raise

def _release_resources(self) -> None:
    for fd in (self._cmd_r, self._cmd_w):
        if fd >= 0:
            try:
                os.close(fd)
            except OSError:
                pass
    self._cmd_r = self._cmd_w = -1
    try:
        self._d.close()
    except Exception:
        pass

def _build(self) -> None:
    """Create the transfer window, intern atoms, arm XFixes, start the event thread.

    Every atom is interned here, before the event thread starts, so read()
    only compares atom ids and no Display call happens off-thread.
    """
    if not self._d.has_extension('XFIXES'):
        raise RuntimeError("XFixes not available")
    self._d.xfixes_query_version()
    screen = self._d.screen()
    self._win = screen.root.create_window(
        0, 0, 1, 1, 0, screen.root_depth, window_class=X.InputOutput,
        event_mask=X.PropertyChangeMask)
    self._clipboard = self._d.get_atom('CLIPBOARD')
    self._primary = self._d.get_atom('PRIMARY')
    # One property per conversion, cycled: a reply the reader has stopped
    # waiting for lands on a retired property, where neither it nor the
    # INCR stream behind it can be mistaken for the current transfer.
    self._props = [self._d.get_atom(f'SELKIES_CLIP_{i}') for i in range(8)]
    self._prop_next = 0
    self._prop = self._props[0]
    self._incr = self._d.get_atom('INCR')
    self._targets = self._d.get_atom('TARGETS')
    self._image_targets = [(self._d.get_atom(m), m) for m in (
        'image/png', 'image/jpeg', 'image/bmp', 'image/webp', 'image/svg+xml',
        'image/svg')]
    self._html_atom = self._d.get_atom('text/html')
    self._text_targets = [(self._d.get_atom(t), t) for t in (
        'UTF8_STRING', 'text/plain;charset=utf-8', 'STRING')]
    self._uri_list_atom = self._d.get_atom('text/uri-list')
    self._d.xfixes_select_selection_input(
        self._win, self._clipboard,
        xfixes.XFixesSetSelectionOwnerNotifyMask
        | xfixes.XFixesSelectionWindowDestroyNotifyMask
        | xfixes.XFixesSelectionClientCloseNotifyMask)
    self._d.flush()
    self._atom_atom = self._d.get_atom('ATOM')
    self._multiple = self._d.get_atom('MULTIPLE')
    self._text_alias_atoms = [a for a, _n in self._text_targets] + [
        self._d.get_atom('TEXT'), self._d.get_atom('text/plain')]
    self._changed = threading.Event()
    self._pending_target = None
    self._gen = 0
    self._serving = None
    self._progress = 0
    self._reply = None
    self._reply_done = threading.Event()
    self._reply_lock = threading.Lock()
    self._abort_read = threading.Event()
    self._read_lock = threading.Lock()
    self._own_offers: list = []
    self._pending_own = None
    self._own_done = threading.Event()
    self._own_ok = False
    self._own_clipboard = False
    self._cmd_r, self._cmd_w = os.pipe()
    self._stop = False
    self._thread = threading.Thread(target=self._event_loop, daemon=True,
                                    name="X11ClipboardMonitor")
    self._thread.start()

def _event_loop(self) -> None:
    """Event-thread main loop: drain X events and run self-pipe commands."""
    xfd = self._d.fileno()
    while not self._stop:
        try:
            r, _, _ = select.select([xfd, self._cmd_r], [], [], 1.0)
        except (OSError, ValueError):
            break
        if self._cmd_r in r:
            os.read(self._cmd_r, 64)
            pending = self._pending_target
            if pending is not None:
                self._pending_target = None
                gen, target = pending
                self._prop_next = (self._prop_next + 1) % len(self._props)
                self._prop = self._props[self._prop_next]
                self._serving = (gen, self._prop)
                try:
                    self._win.convert_selection(self._clipboard, target,
                                                self._prop, X.CurrentTime)
                    self._d.flush()
                except Exception:
                    self._finish_reply(None)
            own = self._pending_own
            if own is not None:
                self._pending_own = None
                self._take_ownership(own)
        if xfd in r or self._d.pending_events():
            try:
                while self._d.pending_events():
                    ev = self._d.next_event()
                    self._dispatch_event(ev)
            except Exception:
                if not self._stop:
                    time.sleep(0.1)

def _dispatch_event(self, ev: Any) -> None:
    """Route one X event on the event thread.

    One payload is served on CLIPBOARD and PRIMARY, so a SelectionClear
    drops it only once a foreign owner has taken BOTH: a text-selection
    steal of PRIMARY must not orphan the browser-written CLIPBOARD
    payload, and vice versa.
    """
    if isinstance(ev, xfixes.SelectionNotify):
        if ev.selection == self._clipboard:
            # The event names the new owner, so ownership is known before
            # the SelectionClear that follows it is dispatched.
            owner = getattr(ev.owner, 'id', ev.owner)
            self._own_clipboard = owner == self._win.id
        self._changed.set()
    elif ev.type == X.SelectionNotify:
        self._collect_selection(ev)
    elif ev.type == X.PropertyNotify and ev.state == X.PropertyNewValue \
            and ev.atom in self._props:
        # The transfer in flight is read by the collector, which the
        # SelectionNotify still to come drives; anything else is a retired
        # transfer, drained here so the owner's INCR stream ends.
        serving = self._serving
        if serving is None or ev.atom != serving[1]:
            self._retire_property(ev.atom)
    elif ev.type == X.SelectionRequest:
        self._serve_selection(ev)
    elif ev.type == X.SelectionClear:
        if ev.atom == self._clipboard:
            self._own_clipboard = False
        try:
            owners = (self._d.get_selection_owner(self._clipboard),
                      self._d.get_selection_owner(self._primary))
            if any(getattr(o, 'id', o) == self._win.id for o in owners):
                return
        except Exception:
            pass
        self._own_offers = []

def _take_ownership(self, offers: list) -> None:
    """On the event thread: stage the offers and claim CLIPBOARD + PRIMARY."""
    try:
        self._own_offers = offers
        self._win.set_selection_owner(self._clipboard, X.CurrentTime)
        self._win.set_selection_owner(self._primary, X.CurrentTime)
        self._d.flush()
        owner = self._d.get_selection_owner(self._clipboard)
        self._own_ok = (getattr(owner, 'id', owner) == self._win.id)
    except Exception:
        self._own_ok = False
    self._own_clipboard = self._own_ok
    self._own_done.set()

def _serve_selection(self, ev: Any) -> None:
    """Answer a SelectionRequest for the payload offer() staged (ICCCM).

    Whatever selection the request names, the answer comes from the one
    staged payload, so CLIPBOARD and PRIMARY are backed alike.
    """
    prop = ev.property if ev.property != X.NONE else ev.target
    granted = X.NONE
    try:
        requestor = ev.requestor
        offers = self._own_offers
        data = None
        if offers and ev.target != self._multiple:
            data = next((p for atom, p in offers if atom == ev.target), None)
        if offers and ev.target == self._targets:
            requestor.change_property(prop, self._atom_atom, 32,
                                      [self._targets] + [a for a, _p in offers])
            granted = prop
        elif data is not None:
            requestor.change_property(prop, ev.target, 8,
                                      data[:self._WRITE_CHUNK])
            offset = self._WRITE_CHUNK
            while offset < len(data):
                requestor.change_property(prop, ev.target, 8,
                                          data[offset:offset + self._WRITE_CHUNK],
                                          mode=X.PropModeAppend)
                offset += self._WRITE_CHUNK
            granted = prop
    except Exception:
        granted = X.NONE
    try:
        notify = xevent.SelectionNotify(
            time=ev.time, requestor=ev.requestor, selection=ev.selection,
            target=ev.target, property=granted)
        ev.requestor.send_event(notify)
        self._d.flush()
    except Exception:
        pass

def _prop_bytes(self, prop: Any) -> bytes:
    v = prop.value
    if isinstance(v, str):
        return v.encode('latin-1')
    if isinstance(v, (bytes, bytearray)):
        return bytes(v)
    return bytes(bytearray(v))

def _finish_reply(self, reply: Optional[tuple]) -> None:
    """Hand a conversion's result to the caller waiting on that generation.

    A reply that arrives after its caller gave up is dropped: handed on, it
    would answer the next conversion with the previous target's bytes.
    """
    with self._reply_lock:
        serving, self._serving = self._serving, None
        if serving is None or serving[0] != self._gen:
            return
        self._reply = reply
        self._reply_done.set()

def _collect_selection(self, ev: Any) -> None:
    """On the event thread: fetch the converted property (INCR-aware).

    Under INCR each property delete requests the next chunk and a
    zero-length chunk ends the transfer. Only that ends it successfully: a
    transfer cut short by the idle bound, the overall bound or the size cap
    is discarded, since half an image handed on as content is worse than a
    read that failed. Events are awaited with the remaining deadline so a
    stalled owner cannot wedge the event thread inside a blocking
    next_event(); paste requests keep being served mid-transfer.
    """
    serving = self._serving
    if serving is None or ev.property not in (X.NONE, serving[1]):
        # An answer to a conversion nobody is waiting on any more. Its
        # property is retired, so the owner is released without letting
        # the bytes near the transfer in flight.
        self._retire_property(ev.property)
        return
    prop_atom = serving[1]
    try:
        if ev.property == X.NONE:
            self._finish_reply(None)
            return
        prop = self._win.get_full_property(prop_atom, X.AnyPropertyType)
        self._win.delete_property(prop_atom)
        self._d.flush()
        self._progress += 1
        if prop is None:
            reply = None
        elif prop.property_type == self._incr:
            chunks = []
            total = 0
            complete = False
            hard_deadline = time.monotonic() + self._TRANSFER_TIMEOUT_S
            deadline = time.monotonic() + self._READ_TIMEOUT_S
            while time.monotonic() < min(deadline, hard_deadline):
                if self._abort_read.is_set():
                    break
                if not self._d.pending_events():
                    remaining = min(deadline, hard_deadline) - time.monotonic()
                    if remaining <= 0:
                        break
                    r, _, _ = select.select(
                        [self._d.fileno(), self._cmd_r], [], [], remaining)
                    if self._cmd_r in r:
                        # A command is a paste or a shutdown, and neither
                        # may wait behind a streaming owner: the transfer
                        # is the one to lose. The byte stays unread for
                        # the main loop to serve.
                        break
                    if not r or not self._d.pending_events():
                        continue
                e = self._d.next_event()
                if (e.type == X.PropertyNotify and e.state == X.PropertyNewValue
                        and e.atom in self._props):
                    if e.atom != prop_atom:
                        # A retired transfer still streaming; drained so it
                        # ends instead of waiting on a delete forever.
                        self._retire_property(e.atom)
                        continue
                    part = self._win.get_full_property(prop_atom, X.AnyPropertyType)
                    self._win.delete_property(prop_atom)
                    self._d.flush()
                    self._progress += 1
                    deadline = time.monotonic() + self._READ_TIMEOUT_S
                    if part is None or len(part.value) == 0:
                        complete = True
                        break
                    piece = self._prop_bytes(part)
                    chunks.append(piece)
                    total += len(piece)
                    if total > self._READ_MAX_BYTES:
                        break
                elif e.type in (X.SelectionRequest, X.SelectionClear) \
                        or isinstance(e, xfixes.SelectionNotify):
                    self._dispatch_event(e)
            reply = (b"".join(chunks), 8) if complete else None
        elif prop.format == 32:
            reply = (list(prop.value), 32)
        else:
            reply = (self._prop_bytes(prop), prop.format)
        self._finish_reply(reply)
    except Exception:
        self._finish_reply(None)

def _retire_property(self, atom: int) -> None:
    """Delete a transfer property whose conversion no longer has a reader."""
    if atom == X.NONE:
        return
    try:
        self._win.delete_property(atom)
        self._d.flush()
    except Exception:
        pass

def _convert_and_wait(self, target_atom: int) -> Optional[tuple]:
    """Request a selection conversion and wait (bounded) for its reply.

    The wait follows the transfer rather than a fixed clock: it extends
    while the collector keeps fetching property chunks and expires only on
    a stall, so a multi-megabyte INCR image is not abandoned half-read.

    Returns:
        (value, format) — bytes for format 8, an atom list for format 32 —
        or None on timeout/failure.
    """
    with self._read_lock:
        with self._reply_lock:
            self._gen += 1
            self._reply = None
            self._reply_done.clear()
            self._pending_target = (self._gen, target_atom)
        os.write(self._cmd_w, b"x")
        seen = self._progress
        hard_deadline = time.monotonic() + self._TRANSFER_TIMEOUT_S
        deadline = time.monotonic() + self._FIRST_REPLY_TIMEOUT_S
        while True:
            remaining = min(deadline, hard_deadline) - time.monotonic()
            if remaining <= 0 or self._abort_read.is_set():
                return None
            if self._reply_done.wait(min(remaining, 0.25)):
                return self._reply
            if self._progress != seen:
                seen = self._progress
                deadline = time.monotonic() + self._READ_TIMEOUT_S

def read(self, use_binary: bool) -> tuple:
    """Blocking read (call via executor): (data, mime) like read_clipboard —
    text as str with mime 'text/plain', markup with the text beneath it as
    one envelope under CLIPBOARD_FLAVOURS_MIME, images as bytes with their
    mime.

    Images come first where the caller takes them, since a copied picture
    offers markup of its own (an `img` tag pointing back at a page) that is
    worth less than the picture; a text selection carries no image target,
    so its markup wins over the plain text beneath it.
    """
    reply = self._convert_and_wait(self._targets)
    if not reply or reply[1] != 32:
        # A fresh owner (xclip mid-fork) may not serve requests for a moment
        # after the owner-change event; one short retry covers it.
        time.sleep(0.1)
        reply = self._convert_and_wait(self._targets)
        if not reply or reply[1] != 32:
            return None, None
    offered = set(reply[0])
    if use_binary:
        for atom, mime in self._image_targets:
            if atom in offered:
                got = self._convert_and_wait(atom)
                if got and got[0]:
                    return bytes(got[0]), mime
        if self._uri_list_atom in offered:
            got = self._convert_and_wait(self._uri_list_atom)
            if got and got[0]:
                resolved = self._resolve_uri_list_image(bytes(got[0]))
                if resolved is not None:
                    return resolved
    if self._html_atom in offered:
        got = self._convert_and_wait(self._html_atom)
        if got is not None and got[0]:
            html = bytes(got[0])
            plain = b''
            for atom, _name in self._text_targets:
                if atom in offered:
                    beside = self._convert_and_wait(atom)
                    if beside is not None and beside[0]:
                        plain = bytes(beside[0])
                        break
            entries = [("text/html", html)] + ([("text/plain", plain)] if plain else [])
            return clipboard_envelope(entries), CLIPBOARD_FLAVOURS_MIME
    for atom, _name in self._text_targets:
        if atom in offered:
            got = self._convert_and_wait(atom)
            if got is not None and got[0] is not None:
                return bytes(got[0]).decode('utf-8', errors='replace'), 'text/plain'
    return None, None

def _resolve_uri_list_image(self, data_bytes: bytes) -> Optional[tuple]:
    """Resolve a text/uri-list (file-manager copy) to (image_bytes, mime): the
    first local file:// URI with a known image extension, read bounded. Returns
    None when nothing qualifies."""
    mime_map = {'.png': 'image/png', '.jpg': 'image/jpeg', '.jpeg': 'image/jpeg',
                '.bmp': 'image/bmp', '.webp': 'image/webp', '.svg': 'image/svg+xml'}
    try:
        text = data_bytes.decode('utf-8', errors='replace')
    except Exception:
        return None
    for line in text.splitlines():
        line = line.strip()
        if not line or line.startswith('#'):
            continue
        try:
            parsed = urllib.parse.urlparse(line)
            if parsed.scheme != 'file':
                continue
            path = urllib.request.url2pathname(parsed.path)
            if not os.path.isfile(path):
                continue
            mime = mime_map.get(os.path.splitext(path)[1].lower())
            if not mime:
                continue
            if 0 < os.path.getsize(path) <= self._URI_FILE_MAX_BYTES:
                with open(path, 'rb') as f:
                    return f.read(self._URI_FILE_MAX_BYTES), mime
        except (OSError, ValueError):
            # ValueError: urlparse rejects malformed bracketed authorities;
            # one bad line must not kill the whole clipboard read.
            continue
    return None

def offer(self, entries: List[Tuple[str, Union[str, bytes]]]) -> bool:
    """Blocking (call via executor): take CLIPBOARD ownership and serve one
    payload per `(mime, data)` entry until another app copies, so a paste
    into a rich editor takes the markup and one into a plain field takes the
    text. Returns True when ownership was acquired."""
    offerable = dict((m, a) for a, m in self._image_targets)
    offerable['text/html'] = self._html_atom
    offers: list = []
    for mime_type, data in entries:
        if not data:
            continue
        payload = data if isinstance(data, bytes) else data.encode('utf-8')
        if mime_type == "text/plain":
            offers += [(atom, payload) for atom in self._text_alias_atoms]
        elif mime_type in offerable:
            offers.append((offerable[mime_type], payload))
    if not offers:
        return False
    # A read of the old selection is worth nothing next to content a client
    # just pasted, and waiting behind a slow owner would hold that paste for
    # as long as the owner takes to answer.
    self._abort_read.set()
    with self._read_lock:
        self._abort_read.clear()
        self._own_done.clear()
        self._own_ok = False
        self._pending_own = offers
        os.write(self._cmd_w, b"o")
        if not self._own_done.wait(self._READ_TIMEOUT_S):
            # Withdrawn: taken late, the ownership would revert the
            # clipboard to this stale payload after a newer X copy.
            self._pending_own = None
            return False
        return self._own_ok

async def wait_change(self, timeout: float) -> bool:
    """Await a selection-owner change (True) or timeout (False), consuming
    it: the outbound monitor loop is the one consumer of the change edge."""
    loop = asyncio.get_running_loop()
    got = await loop.run_in_executor(None, self._changed.wait, timeout)
    if got:
        self._changed.clear()
    return got

async def peek_change(self, timeout: float) -> bool:
    """Await a selection-owner change without consuming it, for a reader
    that wants the fresh content but must leave the edge to the monitor
    loop (which broadcasts it to every client)."""
    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(None, self._changed.wait, timeout)

def alive(self) -> bool:
    """False once the event thread has exited (display disruption, XFixes
    error). A dead monitor never reports another change, so the outbound
    monitor loop rebuilds it instead of waiting on it forever."""
    return not self._stop and self._thread.is_alive()

def owns_selection(self) -> bool:
    """True while CLIPBOARD still belongs to the last offer(), i.e. no X
    application has copied since (a PRIMARY steal alone does not count,
    nor does PRIMARY being ours after an X copy took CLIPBOARD)."""
    return self._own_clipboard

def close(self) -> None:
    self._stop = True
    try:
        os.write(self._cmd_w, b"q")
    except OSError:
        pass
    # Join before close: the thread must leave select() before the display goes.
    self._thread.join(timeout=2.0)
    self._release_resources()