_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')
    self._prop = self._d.get_atom('SELKIES_CLIP')
    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._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._reply = None
    self._reply_done = threading.Event()
    self._read_lock = threading.Lock()
    self._own_data = None
    self._own_mime_atom = None
    self._own_is_text = False
    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)
            target = self._pending_target
            if target is not None:
                self._pending_target = None
                try:
                    self._win.convert_selection(self._clipboard, target,
                                                self._prop, X.CurrentTime)
                    self._d.flush()
                except Exception:
                    self._reply = None
                    self._reply_done.set()
            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):
        self._changed.set()
    elif ev.type == X.SelectionNotify:
        self._collect_selection(ev)
    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_data = None
        self._own_mime_atom = None

def _take_ownership(self, payload: tuple) -> None:
    """On the event thread: stage the payload and claim CLIPBOARD + PRIMARY."""
    data, mime_atom, is_text = payload
    try:
        self._own_data = data
        self._own_mime_atom = mime_atom
        self._own_is_text = is_text
        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
        data = self._own_data
        if data is not None and ev.target == self._targets:
            offered = [self._targets]
            if self._own_is_text:
                offered += self._text_alias_atoms
            elif self._own_mime_atom is not None:
                offered.append(self._own_mime_atom)
            requestor.change_property(prop, self._atom_atom, 32, offered)
            granted = prop
        elif data is not None and ev.target != self._multiple and (
                (self._own_is_text and ev.target in self._text_alias_atoms)
                or ev.target == self._own_mime_atom):
            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 _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. Events are awaited with the
    remaining deadline so a stalled owner times the read out instead of
    wedging the event thread inside a blocking next_event(), and the total
    is capped like the Wayland read so a hostile owner cannot balloon
    memory; paste requests keep being served mid-transfer.
    """
    try:
        if ev.property == X.NONE:
            self._reply = None
            self._reply_done.set()
            return
        prop = self._win.get_full_property(self._prop, X.AnyPropertyType)
        self._win.delete_property(self._prop)
        self._d.flush()
        if prop is None:
            self._reply = None
        elif prop.property_type == self._incr:
            chunks = []
            total = 0
            deadline = time.monotonic() + self._READ_TIMEOUT_S
            while time.monotonic() < deadline and total <= self._READ_MAX_BYTES:
                if not self._d.pending_events():
                    remaining = deadline - time.monotonic()
                    if remaining <= 0:
                        break
                    r, _, _ = select.select([self._d.fileno()], [], [], remaining)
                    if not r or not self._d.pending_events():
                        continue
                e = self._d.next_event()
                if (e.type == X.PropertyNotify and e.atom == self._prop
                        and e.state == X.PropertyNewValue):
                    part = self._win.get_full_property(self._prop, X.AnyPropertyType)
                    self._win.delete_property(self._prop)
                    self._d.flush()
                    if part is None or len(part.value) == 0:
                        break
                    piece = self._prop_bytes(part)
                    chunks.append(piece)
                    total += len(piece)
                elif e.type in (X.SelectionRequest, X.SelectionClear) \
                        or isinstance(e, xfixes.SelectionNotify):
                    self._dispatch_event(e)
            self._reply = (b"".join(chunks), 8)
        elif prop.format == 32:
            self._reply = (list(prop.value), 32)
        else:
            self._reply = (self._prop_bytes(prop), prop.format)
        self._reply_done.set()
    except Exception:
        self._reply = None
        self._reply_done.set()

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

    Returns:
        (value, format) — bytes for format 8, an atom list for format 32 —
        or None on timeout/failure.
    """
    with self._read_lock:
        self._reply = None
        self._reply_done.clear()
        self._pending_target = target_atom
        os.write(self._cmd_w, b"x")
        if not self._reply_done.wait(self._READ_TIMEOUT_S):
            return None
        return self._reply

def read(self, use_binary: bool) -> tuple:
    """Blocking read (call via executor): (data, mime) like read_clipboard —
    text as str with mime 'text/plain', images as bytes with their mime."""
    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
    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, data: Union[str, bytes], mime_type: str) -> bool:
    """Blocking (call via executor): take CLIPBOARD ownership and serve `data`
    until another app copies. Returns True when ownership was acquired."""
    if not data:
        return False
    is_text = mime_type == "text/plain"
    data_bytes = data if isinstance(data, bytes) else data.encode('utf-8')
    mime_atom = None
    if not is_text:
        known = dict((m, a) for a, m in self._image_targets)
        mime_atom = known.get(mime_type)
        if mime_atom is None:
            return False
    with self._read_lock:
        self._own_done.clear()
        self._own_ok = False
        self._pending_own = (data_bytes, mime_atom, is_text)
        os.write(self._cmd_w, b"o")
        if not self._own_done.wait(self._READ_TIMEOUT_S):
            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()