From b3dfb5ddd5207155aeb0a33d424f2de4910705b2 Mon Sep 17 00:00:00 2001
From: xulek <xulek@users.noreply.github.com>
Date: Mon, 7 Sep 2026 07:32:50 +0200
Subject: [PATCH 01/20] feat(v3.3): harvest native DXGI dirty and move metadata

---
 src/vnc_lib/capture_backends.py | 89 ++++++++++++++++++++++++++++++---
 1 file changed, 81 insertions(+), 8 deletions(-)

diff --git a/src/vnc_lib/capture_backends.py b/src/vnc_lib/capture_backends.py
index 59b6f41..1507bdd 100644
--- a/src/vnc_lib/capture_backends.py
+++ b/src/vnc_lib/capture_backends.py
@@ -1,9 +1,9 @@
 """
 Capture backend abstractions and metadata hints.
 
-This module does not implement full DXGI dirty/move rect harvesting yet.
-It provides a stable interface so the server runtime can consume backend-
-supplied metadata as soon as a Windows backend starts exposing it.
+The DXCam adapter can harvest native DXGI Desktop Duplication dirty/move
+rectangles through the optional metadata hook while MSS/Pillow retain the
+portable software-diff fallback.
 """
 
 from __future__ import annotations
@@ -11,6 +11,8 @@
 from dataclasses import dataclass, field
 from typing import Any
 
+from .dxgi_metadata import DXGIMetadataHook
+
 
 Rectangle = tuple[int, int, int, int]
 
@@ -179,22 +181,92 @@ class DXCamCaptureBackend(BaseCaptureBackend):
         supports_bgra=True,
         supports_rgb=True,
         supports_pil_image=False,
-        # The runtime is now ready for these metadata hints, but dxcam does not
-        # provide them in this implementation yet.
-        supports_dirty_regions=False,
-        supports_move_rects=False,
+        supports_dirty_regions=True,
+        supports_move_rects=True,
     )
 
+    def __init__(self, owner: Any):
+        super().__init__(owner)
+        self._metadata_hook = DXGIMetadataHook(logger=getattr(owner, "logger", None))
+
     def is_available(self) -> bool:
         return bool(getattr(self.owner, "_dxcam_available", False))
 
+    def _get_camera(self) -> Any:
+        camera = self.owner._get_dxcam_session()
+        if camera is not None:
+            self._metadata_hook.ensure_attached(camera)
+        return camera
+
+    def _metadata_is_usable(self, camera: Any) -> bool:
+        if not bool(getattr(self.owner, "enable_dxgi_metadata", True)):
+            return False
+        if camera is None or not self._metadata_hook.is_available:
+            return False
+        if float(getattr(self.owner, "scale_factor", 1.0)) != 1.0:
+            return False
+        if int(getattr(camera, "rotation_angle", 0) or 0) != 0:
+            return False
+
+        region = getattr(camera, "region", None)
+        camera_width = int(getattr(camera, "width", 0) or 0)
+        camera_height = int(getattr(camera, "height", 0) or 0)
+        if region is not None and camera_width > 0 and camera_height > 0:
+            if tuple(region) != (0, 0, camera_width, camera_height):
+                return False
+        return True
+
     def healthcheck(self) -> bool:
         try:
-            return self.is_available() and self.owner._get_dxcam_session() is not None
+            return self.is_available() and self._get_camera() is not None
         except Exception:
             return False
 
+    def build_metadata(self, width: int, height: int) -> CaptureMetadata:
+        try:
+            camera = self._get_camera()
+        except Exception:
+            camera = None
+
+        supported = self._metadata_is_usable(camera)
+        if not supported:
+            return CaptureMetadata(
+                backend_name=self.name,
+                dirty_regions=None,
+                move_rects=[],
+                supports_dirty_regions=False,
+                supports_move_rects=False,
+            )
+
+        if width <= 0 or height <= 0:
+            return CaptureMetadata(
+                backend_name="dxcam+dxgi-metadata",
+                dirty_regions=None,
+                move_rects=[],
+                supports_dirty_regions=True,
+                supports_move_rects=True,
+            )
+
+        hints = self._metadata_hook.consume(width, height)
+        if hints is None:
+            return CaptureMetadata(
+                backend_name="dxcam+dxgi-fallback",
+                dirty_regions=None,
+                move_rects=[],
+                supports_dirty_regions=True,
+                supports_move_rects=True,
+            )
+
+        return CaptureMetadata(
+            backend_name="dxcam+dxgi-metadata",
+            dirty_regions=list(hints.dirty_regions),
+            move_rects=list(hints.move_rects),
+            supports_dirty_regions=True,
+            supports_move_rects=True,
+        )
+
     def grab_bgra(self) -> tuple[bytes | None, int, int]:
+        self._get_camera()
         frame_bytes, width, height, channels, color_hint = self.owner._grab_dxcam_frame()
         if frame_bytes is None:
             return None, 0, 0
@@ -206,6 +278,7 @@ def grab_bgra(self) -> tuple[bytes | None, int, int]:
         return bgra, width, height
 
     def grab_rgb(self) -> tuple[bytes | None, int, int]:
+        self._get_camera()
         frame_bytes, width, height, channels, color_hint = self.owner._grab_dxcam_frame()
         if frame_bytes is None:
             return None, 0, 0

From b388af6231c0d49dd019ca5a2bb1550766a9b45b Mon Sep 17 00:00:00 2001
From: xulek <xulek@users.noreply.github.com>
Date: Mon, 7 Sep 2026 07:33:23 +0200
Subject: [PATCH 02/20] feat(v3.3): add DXGI metadata hook

---
 src/vnc_lib/dxgi_metadata.py | 322 +++++++++++++++++++++++++++++++++++
 1 file changed, 322 insertions(+)
 create mode 100644 src/vnc_lib/dxgi_metadata.py

diff --git a/src/vnc_lib/dxgi_metadata.py b/src/vnc_lib/dxgi_metadata.py
new file mode 100644
index 0000000..6ecdd17
--- /dev/null
+++ b/src/vnc_lib/dxgi_metadata.py
@@ -0,0 +1,322 @@
+"""DXGI Desktop Duplication dirty/move rectangle metadata support.
+
+DXCam already owns the IDXGIOutputDuplication object used for capture. The
+public DXCam API exposes pixels rather than frame metadata, so PyVNCServer
+attaches a small optional hook around DXCam's duplicator to read
+GetFrameDirtyRects/GetFrameMoveRects while the acquired frame is still held.
+
+The hook is deliberately fail-safe for correctness: when metadata cannot be
+read safely, callers receive ``None`` and fall back to the software change
+detector instead of assuming that the framebuffer did not change.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+import ctypes
+import logging
+import os
+from typing import Any, Protocol, TYPE_CHECKING
+
+Rectangle = tuple[int, int, int, int]
+
+if TYPE_CHECKING:
+    from .capture_backends import CaptureMoveRect
+
+
+_GET_FRAME_DIRTY_RECTS_INDEX = 9
+_GET_FRAME_MOVE_RECTS_INDEX = 10
+_MAX_METADATA_BYTES = 8 * 1024 * 1024
+_MAX_METADATA_RECTS = 65535
+
+_UINT = ctypes.c_uint32
+_HRESULT = ctypes.c_int32
+_WINFUNCTYPE = getattr(ctypes, "WINFUNCTYPE", ctypes.CFUNCTYPE)
+
+
+class _Point(ctypes.Structure):
+    _fields_ = [("x", ctypes.c_int32), ("y", ctypes.c_int32)]
+
+
+class _Rect(ctypes.Structure):
+    _fields_ = [
+        ("left", ctypes.c_int32),
+        ("top", ctypes.c_int32),
+        ("right", ctypes.c_int32),
+        ("bottom", ctypes.c_int32),
+    ]
+
+
+class _MoveRect(ctypes.Structure):
+    _fields_ = [("source_point", _Point), ("destination_rect", _Rect)]
+
+
+@dataclass(frozen=True, slots=True)
+class DXGIFrameMetadata:
+    """Validated Desktop Duplication metadata for one acquired frame."""
+
+    dirty_regions: tuple[Rectangle, ...]
+    move_rects: tuple[CaptureMoveRect, ...]
+
+
+class DXGIMetadataReaderProtocol(Protocol):
+    def is_supported(self) -> bool: ...
+    def read(self, duplicator_pointer: Any) -> DXGIFrameMetadata | None: ...
+
+
+class DXGIMetadataReader:
+    """Read dirty and move rectangles from an IDXGIOutputDuplication pointer."""
+
+    def __init__(self, logger: logging.Logger | None = None) -> None:
+        self.logger = logger or logging.getLogger(__name__)
+
+    def is_supported(self) -> bool:
+        return os.name == "nt"
+
+    def read(self, duplicator_pointer: Any) -> DXGIFrameMetadata | None:
+        if not self.is_supported() or not duplicator_pointer:
+            return None
+        try:
+            move_buffer = self._read_buffer(
+                duplicator_pointer,
+                _GET_FRAME_MOVE_RECTS_INDEX,
+                ctypes.sizeof(_MoveRect),
+            )
+            dirty_buffer = self._read_buffer(
+                duplicator_pointer,
+                _GET_FRAME_DIRTY_RECTS_INDEX,
+                ctypes.sizeof(_Rect),
+            )
+            if move_buffer is None or dirty_buffer is None:
+                return None
+            move_rects = self._decode_moves(move_buffer)
+            dirty_regions = self._decode_dirty(dirty_buffer)
+            if move_rects is None or dirty_regions is None:
+                return None
+            return DXGIFrameMetadata(tuple(dirty_regions), tuple(move_rects))
+        except Exception as exc:
+            self.logger.debug("Unable to read DXGI frame metadata: %s", exc)
+            return None
+
+    def _read_buffer(self, duplicator_pointer: Any, method_index: int,
+                     item_size: int) -> bytes | None:
+        method, this_pointer = self._bind_metadata_method(
+            duplicator_pointer, method_index
+        )
+        required = _UINT(0)
+        try:
+            probe_hr = int(method(this_pointer, 0, None, ctypes.byref(required)))
+        except Exception:
+            return None
+
+        size = int(required.value)
+        if size == 0:
+            return b"" if probe_hr >= 0 else None
+        if size < item_size or size > _MAX_METADATA_BYTES:
+            return None
+        if size // item_size > _MAX_METADATA_RECTS:
+            return None
+
+        buffer = ctypes.create_string_buffer(size)
+        returned = _UINT(size)
+        hr = int(method(
+            this_pointer,
+            size,
+            ctypes.cast(buffer, ctypes.c_void_p),
+            ctypes.byref(returned),
+        ))
+        if hr < 0:
+            return None
+
+        used = int(returned.value)
+        if used < 0 or used > size or used % item_size != 0:
+            return None
+        return bytes(buffer.raw[:used])
+
+    @staticmethod
+    def _bind_metadata_method(duplicator_pointer: Any, method_index: int):
+        this_pointer = ctypes.cast(duplicator_pointer, ctypes.c_void_p)
+        if not this_pointer.value:
+            raise ValueError("null IDXGIOutputDuplication pointer")
+        vtable_ptr = ctypes.cast(
+            duplicator_pointer,
+            ctypes.POINTER(ctypes.POINTER(ctypes.c_void_p)),
+        ).contents
+        function_address = vtable_ptr[method_index]
+        if isinstance(function_address, ctypes.c_void_p):
+            function_address = function_address.value
+        if not function_address:
+            raise ValueError("missing IDXGIOutputDuplication vtable entry")
+        prototype = _WINFUNCTYPE(
+            _HRESULT,
+            ctypes.c_void_p,
+            _UINT,
+            ctypes.c_void_p,
+            ctypes.POINTER(_UINT),
+        )
+        return prototype(function_address), this_pointer
+
+    @staticmethod
+    def _decode_dirty(buffer: bytes) -> list[Rectangle] | None:
+        if not buffer:
+            return []
+        item_size = ctypes.sizeof(_Rect)
+        if len(buffer) % item_size:
+            return None
+        values = (_Rect * (len(buffer) // item_size)).from_buffer_copy(buffer)
+        regions: list[Rectangle] = []
+        for rect in values:
+            width = int(rect.right - rect.left)
+            height = int(rect.bottom - rect.top)
+            if width > 0 and height > 0:
+                regions.append((int(rect.left), int(rect.top), width, height))
+        return regions
+
+    @staticmethod
+    def _decode_moves(buffer: bytes) -> list[CaptureMoveRect] | None:
+        if not buffer:
+            return []
+        item_size = ctypes.sizeof(_MoveRect)
+        if len(buffer) % item_size:
+            return None
+        from .capture_backends import CaptureMoveRect
+        values = (_MoveRect * (len(buffer) // item_size)).from_buffer_copy(buffer)
+        moves: list[CaptureMoveRect] = []
+        for item in values:
+            dst = item.destination_rect
+            width = int(dst.right - dst.left)
+            height = int(dst.bottom - dst.top)
+            if width > 0 and height > 0:
+                moves.append(CaptureMoveRect(
+                    src_x=int(item.source_point.x),
+                    src_y=int(item.source_point.y),
+                    dst_x=int(dst.left),
+                    dst_y=int(dst.top),
+                    width=width,
+                    height=height,
+                ))
+        return moves
+
+
+class DXGIMetadataHook:
+    """Attach metadata collection to the private DXCam DXGI duplicator."""
+
+    def __init__(self, reader: DXGIMetadataReaderProtocol | None = None,
+                 logger: logging.Logger | None = None) -> None:
+        self.logger = logger or logging.getLogger(__name__)
+        self.reader = reader or DXGIMetadataReader(self.logger)
+        self._attached_duplicator: Any = None
+        self._pending: DXGIFrameMetadata | None = None
+        self._pending_valid = False
+
+    @property
+    def is_available(self) -> bool:
+        return bool(self._attached_duplicator is not None and self.reader.is_supported())
+
+    def ensure_attached(self, camera: Any) -> bool:
+        if not self.reader.is_supported():
+            return False
+        if getattr(camera, "backend", "dxgi") != "dxgi":
+            return False
+        duplicator = getattr(camera, "_duplicator", None)
+        if duplicator is None or getattr(duplicator, "duplicator", None) is None:
+            return False
+        if duplicator is self._attached_duplicator:
+            return True
+        original_update = getattr(duplicator, "update_frame", None)
+        if not callable(original_update):
+            return False
+        hook = self
+
+        def update_frame_with_metadata(*args: Any, **kwargs: Any) -> bool:
+            ok = bool(original_update(*args, **kwargs))
+            hook._pending = None
+            hook._pending_valid = False
+            if not ok:
+                return False
+            if not bool(getattr(duplicator, "updated", False)):
+                hook._pending = DXGIFrameMetadata((), ())
+                hook._pending_valid = True
+                return True
+            metadata = hook.reader.read(getattr(duplicator, "duplicator", None))
+            if metadata is not None:
+                hook._pending = metadata
+                hook._pending_valid = True
+            return True
+
+        duplicator.update_frame = update_frame_with_metadata
+        self._attached_duplicator = duplicator
+        self._pending = None
+        self._pending_valid = False
+        return True
+
+    def consume(self, width: int, height: int) -> DXGIFrameMetadata | None:
+        if not self._pending_valid:
+            return None
+        pending = self._pending
+        self._pending = None
+        self._pending_valid = False
+        if pending is None:
+            return None
+
+        dirty = tuple(
+            region for region in (
+                _clip_rectangle(rect, width, height)
+                for rect in pending.dirty_regions
+            ) if region is not None
+        )
+        clipped_moves: list[CaptureMoveRect] = []
+        for move in pending.move_rects:
+            clipped = _clip_move_rect(move, width, height)
+            if clipped is not None:
+                clipped_moves.append(clipped)
+                continue
+            destination = _clip_rectangle(
+                (move.dst_x, move.dst_y, move.width, move.height), width, height
+            )
+            if destination is not None:
+                return None
+        return DXGIFrameMetadata(dirty, tuple(clipped_moves))
+
+
+def _clip_rectangle(rect: Rectangle, width: int, height: int) -> Rectangle | None:
+    x, y, rect_width, rect_height = map(int, rect)
+    x1 = max(0, x)
+    y1 = max(0, y)
+    x2 = min(int(width), x + rect_width)
+    y2 = min(int(height), y + rect_height)
+    if x1 >= x2 or y1 >= y2:
+        return None
+    return x1, y1, x2 - x1, y2 - y1
+
+
+def _clip_move_rect(move: "CaptureMoveRect", width: int,
+                    height: int) -> "CaptureMoveRect | None":
+    from .capture_backends import CaptureMoveRect
+    dst_x1 = max(0, int(move.dst_x))
+    dst_y1 = max(0, int(move.dst_y))
+    dst_x2 = min(int(width), int(move.dst_x + move.width))
+    dst_y2 = min(int(height), int(move.dst_y + move.height))
+    if dst_x1 >= dst_x2 or dst_y1 >= dst_y2:
+        return None
+
+    offset_x = dst_x1 - int(move.dst_x)
+    offset_y = dst_y1 - int(move.dst_y)
+    src_x = int(move.src_x) + offset_x
+    src_y = int(move.src_y) + offset_y
+    clipped_width = dst_x2 - dst_x1
+    clipped_height = dst_y2 - dst_y1
+    if (
+        src_x < 0 or src_y < 0
+        or src_x + clipped_width > width
+        or src_y + clipped_height > height
+    ):
+        return None
+    return CaptureMoveRect(
+        src_x=src_x,
+        src_y=src_y,
+        dst_x=dst_x1,
+        dst_y=dst_y1,
+        width=clipped_width,
+        height=clipped_height,
+    )

From 5e8bfb29310b15f44721fac2214710add1ea3c28 Mon Sep 17 00:00:00 2001
From: xulek <xulek@users.noreply.github.com>
Date: Mon, 7 Sep 2026 07:35:45 +0200
Subject: [PATCH 03/20] fix(v3.3): preserve move destinations for non-CopyRect
 clients

---
 src/vnc_lib/capture_backends.py | 15 ++++++++++++++-
 1 file changed, 14 insertions(+), 1 deletion(-)

diff --git a/src/vnc_lib/capture_backends.py b/src/vnc_lib/capture_backends.py
index 1507bdd..9ab22a1 100644
--- a/src/vnc_lib/capture_backends.py
+++ b/src/vnc_lib/capture_backends.py
@@ -257,9 +257,22 @@ def build_metadata(self, width: int, height: int) -> CaptureMetadata:
                 supports_move_rects=True,
             )
 
+        # Conservative v3.3 safety rule: move destinations are included in the
+        # pixel-dirty list as well as exposed as CopyRect hints. This guarantees
+        # correctness for clients without CopyRect and for clients that skip
+        # producer generations. A later optimization may suppress the redundant
+        # pixel rectangle for clients that are exactly one generation behind
+        # and advertise CopyRect.
+        dirty_regions = list(hints.dirty_regions)
+        dirty_regions.extend(
+            (move.dst_x, move.dst_y, move.width, move.height)
+            for move in hints.move_rects
+        )
+        dirty_regions = list(dict.fromkeys(dirty_regions))
+
         return CaptureMetadata(
             backend_name="dxcam+dxgi-metadata",
-            dirty_regions=list(hints.dirty_regions),
+            dirty_regions=dirty_regions,
             move_rects=list(hints.move_rects),
             supports_dirty_regions=True,
             supports_move_rects=True,

From 0bf99c82bd9c033feb2eec7a54766529d98afb91 Mon Sep 17 00:00:00 2001
From: xulek <xulek@users.noreply.github.com>
Date: Mon, 7 Sep 2026 07:36:15 +0200
Subject: [PATCH 04/20] test(v3.3): cover DXGI metadata ABI and fallback safety

---
 tests/test_dxgi_metadata.py | 213 ++++++++++++++++++++++++++++++++++++
 1 file changed, 213 insertions(+)
 create mode 100644 tests/test_dxgi_metadata.py

diff --git a/tests/test_dxgi_metadata.py b/tests/test_dxgi_metadata.py
new file mode 100644
index 0000000..567f1f6
--- /dev/null
+++ b/tests/test_dxgi_metadata.py
@@ -0,0 +1,213 @@
+"""DXGI Desktop Duplication metadata hook tests."""
+
+from __future__ import annotations
+
+import ctypes
+from dataclasses import dataclass
+import logging
+
+import vnc_lib.dxgi_metadata as dxgi_metadata
+from vnc_lib.capture_backends import CaptureMoveRect, DXCamCaptureBackend
+from vnc_lib.dxgi_metadata import DXGIFrameMetadata, DXGIMetadataHook, DXGIMetadataReader
+
+
+class _FakeReader:
+    def __init__(self, values):
+        self.values = list(values)
+        self.calls = 0
+
+    def is_supported(self) -> bool:
+        return True
+
+    def read(self, _duplicator_pointer):
+        self.calls += 1
+        return self.values.pop(0) if self.values else None
+
+
+class _FakeDuplicator:
+    def __init__(self):
+        self.duplicator = object()
+        self.updated = True
+
+    def update_frame(self, *_args, **_kwargs):
+        return True
+
+
+@dataclass
+class _FakeCamera:
+    _duplicator: _FakeDuplicator
+    backend: str = "dxgi"
+    width: int = 100
+    height: int = 100
+    region: tuple[int, int, int, int] = (0, 0, 100, 100)
+    rotation_angle: int = 0
+
+
+def test_decode_dirty_rectangles():
+    values = (dxgi_metadata._Rect * 2)(
+        dxgi_metadata._Rect(10, 20, 40, 60),
+        dxgi_metadata._Rect(100, 200, 150, 220),
+    )
+    assert DXGIMetadataReader._decode_dirty(bytes(values)) == [
+        (10, 20, 30, 40),
+        (100, 200, 50, 20),
+    ]
+
+
+def test_decode_move_rectangles():
+    values = (dxgi_metadata._MoveRect * 1)(
+        dxgi_metadata._MoveRect(
+            dxgi_metadata._Point(5, 7),
+            dxgi_metadata._Rect(20, 30, 60, 80),
+        )
+    )
+    assert DXGIMetadataReader._decode_moves(bytes(values)) == [
+        CaptureMoveRect(src_x=5, src_y=7, dst_x=20, dst_y=30, width=40, height=50)
+    ]
+
+
+def test_hook_captures_metadata_once_per_frame():
+    reader = _FakeReader([
+        DXGIFrameMetadata(
+            dirty_regions=((1, 2, 3, 4),),
+            move_rects=(CaptureMoveRect(0, 0, 10, 10, 5, 5),),
+        )
+    ])
+    duplicator = _FakeDuplicator()
+    hook = DXGIMetadataHook(reader=reader)
+    assert hook.ensure_attached(_FakeCamera(duplicator))
+    assert duplicator.update_frame()
+
+    metadata = hook.consume(100, 100)
+    assert metadata == DXGIFrameMetadata(
+        dirty_regions=((1, 2, 3, 4),),
+        move_rects=(CaptureMoveRect(0, 0, 10, 10, 5, 5),),
+    )
+    assert hook.consume(100, 100) is None
+    assert reader.calls == 1
+
+
+def test_hook_timeout_is_authoritative_empty_metadata():
+    reader = _FakeReader([])
+    duplicator = _FakeDuplicator()
+    duplicator.updated = False
+    hook = DXGIMetadataHook(reader=reader)
+    assert hook.ensure_attached(_FakeCamera(duplicator))
+    assert duplicator.update_frame()
+    assert hook.consume(100, 100) == DXGIFrameMetadata((), ())
+    assert reader.calls == 0
+
+
+def test_failed_metadata_read_requests_software_fallback():
+    reader = _FakeReader([None])
+    duplicator = _FakeDuplicator()
+    hook = DXGIMetadataHook(reader=reader)
+    assert hook.ensure_attached(_FakeCamera(duplicator))
+    duplicator.update_frame()
+    assert hook.consume(100, 100) is None
+
+
+def test_hook_clips_metadata_to_framebuffer():
+    reader = _FakeReader([
+        DXGIFrameMetadata(
+            dirty_regions=((-10, -5, 30, 20), (200, 200, 10, 10)),
+            move_rects=(CaptureMoveRect(5, 5, -2, -3, 10, 10),),
+        )
+    ])
+    duplicator = _FakeDuplicator()
+    hook = DXGIMetadataHook(reader=reader)
+    assert hook.ensure_attached(_FakeCamera(duplicator))
+    duplicator.update_frame()
+
+    metadata = hook.consume(100, 100)
+    assert metadata is not None
+    assert metadata.dirty_regions == ((0, 0, 20, 15),)
+    assert metadata.move_rects == (
+        CaptureMoveRect(src_x=7, src_y=8, dst_x=0, dst_y=0, width=8, height=7),
+    )
+
+
+def test_invalid_move_source_invalidates_native_metadata():
+    reader = _FakeReader([
+        DXGIFrameMetadata(
+            dirty_regions=(),
+            move_rects=(CaptureMoveRect(-5, 0, 20, 20, 10, 10),),
+        )
+    ])
+    duplicator = _FakeDuplicator()
+    hook = DXGIMetadataHook(reader=reader)
+    assert hook.ensure_attached(_FakeCamera(duplicator))
+    duplicator.update_frame()
+    assert hook.consume(100, 100) is None
+
+
+def test_raw_vtable_metadata_buffer_reader():
+    item = dxgi_metadata._Rect(1, 2, 11, 22)
+    payload = bytes(item)
+    callback_type = dxgi_metadata._WINFUNCTYPE(
+        dxgi_metadata._HRESULT,
+        ctypes.c_void_p,
+        dxgi_metadata._UINT,
+        ctypes.c_void_p,
+        ctypes.POINTER(dxgi_metadata._UINT),
+    )
+
+    def callback(_this, buffer_size, buffer, required):
+        required[0] = len(payload)
+        if int(buffer_size) < len(payload) or not buffer:
+            return -1
+        ctypes.memmove(buffer, payload, len(payload))
+        return 0
+
+    callback_fn = callback_type(callback)
+    vtable = (ctypes.c_void_p * 15)()
+    vtable[dxgi_metadata._GET_FRAME_DIRTY_RECTS_INDEX] = ctypes.cast(
+        callback_fn, ctypes.c_void_p
+    ).value
+
+    class _FakeComObject(ctypes.Structure):
+        _fields_ = [("vtable", ctypes.POINTER(ctypes.c_void_p))]
+
+    obj = _FakeComObject(ctypes.cast(vtable, ctypes.POINTER(ctypes.c_void_p)))
+    reader = DXGIMetadataReader()
+    raw = reader._read_buffer(
+        ctypes.pointer(obj),
+        dxgi_metadata._GET_FRAME_DIRTY_RECTS_INDEX,
+        ctypes.sizeof(dxgi_metadata._Rect),
+    )
+    assert raw == payload
+    assert reader._decode_dirty(raw) == [(1, 2, 10, 20)]
+
+
+def test_dxcam_backend_keeps_move_destination_dirty_for_safe_fallback():
+    move = CaptureMoveRect(1, 2, 20, 30, 10, 12)
+
+    class FakeMetadataHook:
+        is_available = True
+
+        def ensure_attached(self, _camera):
+            return True
+
+        def consume(self, _width, _height):
+            return DXGIFrameMetadata(((4, 5, 6, 7),), (move,))
+
+    class FakeOwner:
+        logger = logging.getLogger("test.dxgi.backend")
+        scale_factor = 1.0
+        enable_dxgi_metadata = True
+        _dxcam_available = True
+
+        def __init__(self):
+            self.camera = _FakeCamera(_FakeDuplicator())
+
+        def _get_dxcam_session(self):
+            return self.camera
+
+    backend = DXCamCaptureBackend(FakeOwner())
+    backend._metadata_hook = FakeMetadataHook()
+    metadata = backend.build_metadata(100, 100)
+
+    assert metadata.dirty_regions == [(4, 5, 6, 7), (20, 30, 10, 12)]
+    assert metadata.move_rects == [move]
+    assert metadata.supports_dirty_regions is True
+    assert metadata.supports_move_rects is True

From e988a70db568841778d39774f0557c96566acc5a Mon Sep 17 00:00:00 2001
From: xulek <xulek@users.noreply.github.com>
Date: Mon, 7 Sep 2026 07:36:28 +0200
Subject: [PATCH 05/20] build(v3.3): require DXCam metadata-capable release

---
 pyproject.toml | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/pyproject.toml b/pyproject.toml
index 3c8a399..57fbc76 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -18,7 +18,7 @@ dependencies = [
 [project.optional-dependencies]
 performance = [
     "numpy>=1.26",
-    "dxcam>=0.0.5; platform_system == 'Windows'",
+    "dxcam>=0.3.0; platform_system == 'Windows'",
 ]
 h264 = ["av>=12.0"]
 dev = ["pytest>=8.0"]

From 7a6f9dd5a7525ea70bb202bda10dd8ad372d6919 Mon Sep 17 00:00:00 2001
From: xulek <xulek@users.noreply.github.com>
Date: Mon, 7 Sep 2026 07:36:38 +0200
Subject: [PATCH 06/20] chore(v3.3): bump version to 3.3.0

---
 src/pyvncserver/_version.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/pyvncserver/_version.py b/src/pyvncserver/_version.py
index d5a083c..df50b22 100644
--- a/src/pyvncserver/_version.py
+++ b/src/pyvncserver/_version.py
@@ -1,4 +1,4 @@
 """Single source of truth for the project version."""
 
-__version__ = "3.2.1"
+__version__ = "3.3.0"
 SERVER_NAME = f"PyVNCServer {__version__}"

From 5f137704f12fbd533578a9f81df75a1ae6c9e22e Mon Sep 17 00:00:00 2001
From: xulek <xulek@users.noreply.github.com>
Date: Mon, 7 Sep 2026 07:36:54 +0200
Subject: [PATCH 07/20] bench(v3.3): add DXGI metadata diagnostic benchmark

---
 benchmarks/benchmark_dxgi_metadata.py | 107 ++++++++++++++++++++++++++
 1 file changed, 107 insertions(+)
 create mode 100644 benchmarks/benchmark_dxgi_metadata.py

diff --git a/benchmarks/benchmark_dxgi_metadata.py b/benchmarks/benchmark_dxgi_metadata.py
new file mode 100644
index 0000000..4533d68
--- /dev/null
+++ b/benchmarks/benchmark_dxgi_metadata.py
@@ -0,0 +1,107 @@
+"""Inspect DXGI dirty/move metadata quality on a real Windows desktop.
+
+Run on Windows with the performance extra installed:
+
+    python -m pip install -e ".[performance]"
+    python benchmarks/benchmark_dxgi_metadata.py --frames 300 --fps 60
+
+The benchmark is diagnostic rather than a synthetic score. It reports how
+often native Desktop Duplication metadata was available, rectangle counts and
+the approximate fraction of the framebuffer affected by each frame.
+"""
+
+from __future__ import annotations
+
+import argparse
+import statistics
+import time
+
+from vnc_lib.screen_capture import ScreenCapture
+
+
+NATIVE_BGR0 = {
+    "bits_per_pixel": 32,
+    "depth": 24,
+    "big_endian_flag": 0,
+    "true_colour_flag": 1,
+    "red_max": 255,
+    "green_max": 255,
+    "blue_max": 255,
+    "red_shift": 16,
+    "green_shift": 8,
+    "blue_shift": 0,
+}
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description="Benchmark DXGI dirty/move metadata")
+    parser.add_argument("--frames", type=int, default=300)
+    parser.add_argument("--fps", type=float, default=60.0)
+    args = parser.parse_args()
+
+    frames = max(1, args.frames)
+    fps = max(1.0, args.fps)
+    interval = 1.0 / fps
+    capture = ScreenCapture(backend_preference="dxcam")
+    if capture.get_backend_name() != "dxcam":
+        raise SystemExit(
+            "DXCam backend is unavailable; install the performance extra on Windows"
+        )
+
+    capture.set_cache_frame_rate(fps)
+    native_frames = 0
+    fallback_frames = 0
+    dirty_counts: list[int] = []
+    move_counts: list[int] = []
+    changed_ratios: list[float] = []
+    capture_ms: list[float] = []
+
+    next_tick = time.perf_counter()
+    try:
+        for _ in range(frames):
+            frame = capture.capture_frame(NATIVE_BGR0)
+            result = frame.result
+            metadata = frame.metadata
+            if result.pixel_data is None:
+                continue
+
+            capture_ms.append(result.capture_time * 1000.0)
+            if metadata.dirty_regions is None:
+                fallback_frames += 1
+            else:
+                native_frames += 1
+                dirty_counts.append(len(metadata.dirty_regions))
+                move_counts.append(len(metadata.move_rects))
+                changed_area = sum(w * h for _, _, w, h in metadata.dirty_regions)
+                changed_ratios.append(
+                    min(1.0, changed_area / max(1, result.width * result.height))
+                )
+
+            next_tick += interval
+            delay = next_tick - time.perf_counter()
+            if delay > 0:
+                time.sleep(delay)
+            else:
+                next_tick = time.perf_counter()
+    finally:
+        capture.close_current_thread_sessions()
+
+    total = native_frames + fallback_frames
+    print(f"backend:             {capture.get_backend_name()}")
+    print(f"frames sampled:      {total}")
+    print(f"native metadata:     {native_frames} ({native_frames / max(1, total):.1%})")
+    print(f"metadata fallbacks:  {fallback_frames}")
+    if capture_ms:
+        ordered = sorted(capture_ms)
+        p95 = ordered[int((len(ordered) - 1) * 0.95)]
+        print(f"capture avg:         {statistics.mean(capture_ms):.2f} ms")
+        print(f"capture p95:         {p95:.2f} ms")
+    if dirty_counts:
+        print(f"dirty rects avg:     {statistics.mean(dirty_counts):.2f}")
+        print(f"move rects avg:      {statistics.mean(move_counts):.2f}")
+        print(f"changed area avg:    {statistics.mean(changed_ratios):.2%}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

From 32f0ee15c5cfa4596bda94b4048c53a959fe2393 Mon Sep 17 00:00:00 2001
From: xulek <xulek@users.noreply.github.com>
Date: Mon, 7 Sep 2026 07:39:14 +0200
Subject: [PATCH 08/20] test(v3.3): add RRE to Tight live-session
 interoperability regression

---
 tests/test_e2e_encoding_switch.py | 163 ++++++++++++++++++++++++++++++
 1 file changed, 163 insertions(+)
 create mode 100644 tests/test_e2e_encoding_switch.py

diff --git a/tests/test_e2e_encoding_switch.py b/tests/test_e2e_encoding_switch.py
new file mode 100644
index 0000000..0c7aa22
--- /dev/null
+++ b/tests/test_e2e_encoding_switch.py
@@ -0,0 +1,163 @@
+"""Real socket regression tests for switching RFB encodings mid-session."""
+
+from __future__ import annotations
+
+import socket
+import struct
+import threading
+
+from pyvncserver.app import server as server_module
+from vnc_lib.capture_backends import CaptureBackendCapabilities, CaptureFrame, CaptureMetadata
+from vnc_lib.screen_capture import CaptureResult
+
+
+class _SolidCapture:
+    def __init__(self, scale_factor=1.0, monitor=0, backend_preference="auto"):
+        self.scale_factor = scale_factor
+        self.monitor = monitor
+        self.backend_preference = backend_preference
+        self._backend = self
+        self._pixels = bytes([10, 20, 30, 0] * 16)
+
+    def set_cache_frame_rate(self, _fps):
+        return None
+
+    def get_backend_name(self):
+        return "fake"
+
+    def get_backend_capabilities(self):
+        return CaptureBackendCapabilities(
+            name="fake",
+            supports_bgra=True,
+            supports_rgb=True,
+            supports_pil_image=False,
+            supports_dirty_regions=True,
+            supports_move_rects=False,
+        )
+
+    def healthcheck(self):
+        return True
+
+    def capture_frame(self, _pixel_format):
+        return CaptureFrame(
+            result=CaptureResult(self._pixels, None, 4, 4, 0.0001),
+            metadata=CaptureMetadata(
+                backend_name="fake",
+                dirty_regions=[(0, 0, 4, 4)],
+                supports_dirty_regions=True,
+            ),
+        )
+
+    def capture_fast(self, pixel_format):
+        return self.capture_frame(pixel_format).result
+
+    def convert_native_bgr0(self, pixel_data, _width, _height, _pixel_format):
+        return pixel_data
+
+    def close_current_thread_sessions(self):
+        return None
+
+
+class _FakeInputHandler:
+    def __init__(self, scale_factor=1.0):
+        self.scale_factor = scale_factor
+
+    def handle_pointer_event(self, *_args, **_kwargs):
+        return None
+
+    def handle_key_event(self, *_args, **_kwargs):
+        return None
+
+
+def _recv_exact(sock: socket.socket, size: int) -> bytes:
+    data = bytearray()
+    while len(data) < size:
+        chunk = sock.recv(size - len(data))
+        if not chunk:
+            raise AssertionError(f"connection closed after {len(data)}/{size} bytes")
+        data.extend(chunk)
+    return bytes(data)
+
+
+def _free_tcp_port() -> int:
+    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    sock.bind(("127.0.0.1", 0))
+    port = sock.getsockname()[1]
+    sock.close()
+    return port
+
+
+def _handshake(client: socket.socket):
+    version = _recv_exact(client, 12)
+    assert version == b"RFB 003.008\n"
+    client.sendall(version)
+    count = _recv_exact(client, 1)[0]
+    security_types = _recv_exact(client, count)
+    assert 1 in security_types
+    client.sendall(b"\x01")
+    assert _recv_exact(client, 4) == b"\x00\x00\x00\x00"
+    client.sendall(b"\x01")
+    width, height = struct.unpack(">HH", _recv_exact(client, 4))
+    _recv_exact(client, 16)
+    name_len = struct.unpack(">I", _recv_exact(client, 4))[0]
+    _recv_exact(client, name_len)
+    assert (width, height) == (4, 4)
+
+
+def _set_encodings(client: socket.socket, encodings: list[int]):
+    payload = struct.pack(">BBH", 2, 0, len(encodings))
+    payload += b"".join(struct.pack(">i", value) for value in encodings)
+    client.sendall(payload)
+
+
+def _request_full(client: socket.socket):
+    client.sendall(struct.pack(">BBHHHH", 3, 0, 0, 0, 4, 4))
+
+
+def _read_rect_header(client: socket.socket):
+    message_type, count = struct.unpack(">BxH", _recv_exact(client, 4))
+    assert message_type == 0
+    assert count == 1
+    return struct.unpack(">HHHHi", _recv_exact(client, 12))
+
+
+def test_switch_rre_to_tight_without_disconnect(tmp_path, monkeypatch):
+    monkeypatch.setattr(server_module, "ScreenCapture", _SolidCapture)
+    monkeypatch.setattr(server_module, "InputHandler", _FakeInputHandler)
+
+    port = _free_tcp_port()
+    config = tmp_path / "server.toml"
+    config.write_text(
+        f'''[server]\nhost = "127.0.0.1"\nport = {port}\nframe_rate = 30\nlan_frame_rate = 30\nnetwork_profile_override = "auto"\nmax_connections = 2\nmax_connections_per_ip = 2\nmax_unauthenticated_connections = 2\nhandshake_timeout = 2.0\nclient_socket_timeout = 2.0\n\n[features]\nenable_metrics = false\nenable_health_checks = false\nenable_capture_producer = false\nenable_parallel_encoding = false\nenable_region_detection = false\nenable_websocket = false\nenable_tight_extensions = false\nenable_tight_encoding = true\nenable_jpeg_encoding = false\nenable_zrle_encoding = false\nenable_copyrect_encoding = false\n''',
+        encoding="utf-8",
+    )
+
+    server = server_module.VNCServerV3(config)
+    thread = threading.Thread(target=server.start, name="test-vnc-encoding-switch")
+    thread.start()
+    client = socket.create_connection(("127.0.0.1", port), timeout=2.0)
+    client.settimeout(2.0)
+    try:
+        _handshake(client)
+
+        _set_encodings(client, [2, 0])
+        _request_full(client)
+        x, y, width, height, encoding = _read_rect_header(client)
+        assert (x, y, width, height, encoding) == (0, 0, 4, 4, 2)
+        subrect_count = struct.unpack(">I", _recv_exact(client, 4))[0]
+        background = _recv_exact(client, 4)
+        assert subrect_count == 0
+        assert background == bytes([10, 20, 30, 0])
+
+        _set_encodings(client, [7, 0])
+        _request_full(client)
+        x, y, width, height, encoding = _read_rect_header(client)
+        assert (x, y, width, height, encoding) == (0, 0, 4, 4, 7)
+        control = _recv_exact(client, 1)[0]
+        assert control >> 4 == 8  # Tight fill
+        assert _recv_exact(client, 3) == bytes([30, 20, 10])
+    finally:
+        client.close()
+        server.shutdown_handler.shutdown()
+        thread.join(timeout=5.0)
+        assert not thread.is_alive()

From 062317c5a10c225aec3829de9ecd87e74763a274 Mon Sep 17 00:00:00 2001
From: xulek <xulek@users.noreply.github.com>
Date: Mon, 7 Sep 2026 07:40:55 +0200
Subject: [PATCH 09/20] test(v3.3): validate full live-session encoding matrix

---
 tests/test_e2e_encoding_switch.py | 72 ++++++++++++++++++++++++++++---
 1 file changed, 65 insertions(+), 7 deletions(-)

diff --git a/tests/test_e2e_encoding_switch.py b/tests/test_e2e_encoding_switch.py
index 0c7aa22..b431ece 100644
--- a/tests/test_e2e_encoding_switch.py
+++ b/tests/test_e2e_encoding_switch.py
@@ -27,11 +27,8 @@ def get_backend_name(self):
 
     def get_backend_capabilities(self):
         return CaptureBackendCapabilities(
-            name="fake",
-            supports_bgra=True,
-            supports_rgb=True,
-            supports_pil_image=False,
-            supports_dirty_regions=True,
+            name="fake", supports_bgra=True, supports_rgb=True,
+            supports_pil_image=False, supports_dirty_regions=True,
             supports_move_rects=False,
         )
 
@@ -42,8 +39,7 @@ def capture_frame(self, _pixel_format):
         return CaptureFrame(
             result=CaptureResult(self._pixels, None, 4, 4, 0.0001),
             metadata=CaptureMetadata(
-                backend_name="fake",
-                dirty_regions=[(0, 0, 4, 4)],
+                backend_name="fake", dirty_regions=[(0, 0, 4, 4)],
                 supports_dirty_regions=True,
             ),
         )
@@ -161,3 +157,65 @@ def test_switch_rre_to_tight_without_disconnect(tmp_path, monkeypatch):
         server.shutdown_handler.shutdown()
         thread.join(timeout=5.0)
         assert not thread.is_alive()
+
+
+def _read_and_validate_payload(client: socket.socket, expected_encoding: int):
+    import zlib
+
+    x, y, width, height, encoding = _read_rect_header(client)
+    assert (x, y, width, height) == (0, 0, 4, 4)
+    assert encoding == expected_encoding
+
+    if encoding == 0:  # Raw
+        assert _recv_exact(client, 64) == bytes([10, 20, 30, 0] * 16)
+    elif encoding == 2:  # RRE solid background
+        assert struct.unpack(">I", _recv_exact(client, 4))[0] == 0
+        assert _recv_exact(client, 4) == bytes([10, 20, 30, 0])
+    elif encoding == 5:  # Hextile uses raw tile in current implementation
+        assert _recv_exact(client, 1) == b"\x01"
+        assert _recv_exact(client, 64) == bytes([10, 20, 30, 0] * 16)
+    elif encoding == 6:  # Zlib
+        length = struct.unpack(">I", _recv_exact(client, 4))[0]
+        compressed = _recv_exact(client, length)
+        decoder = zlib.decompressobj()
+        assert decoder.decompress(compressed) == bytes([10, 20, 30, 0] * 16)
+    elif encoding == 16:  # ZRLE solid tile: subencoding 1 + CPIXEL
+        length = struct.unpack(">I", _recv_exact(client, 4))[0]
+        compressed = _recv_exact(client, length)
+        decoder = zlib.decompressobj()
+        assert decoder.decompress(compressed) == bytes([1, 10, 20, 30])
+    elif encoding == 7:  # Tight fill
+        control = _recv_exact(client, 1)[0]
+        assert control >> 4 == 8
+        assert _recv_exact(client, 3) == bytes([30, 20, 10])
+    else:
+        raise AssertionError(f"unsupported test encoding {encoding}")
+
+
+def test_live_session_encoding_matrix(tmp_path, monkeypatch):
+    monkeypatch.setattr(server_module, "ScreenCapture", _SolidCapture)
+    monkeypatch.setattr(server_module, "InputHandler", _FakeInputHandler)
+
+    port = _free_tcp_port()
+    config = tmp_path / "server-matrix.toml"
+    config.write_text(
+        f'''[server]\nhost = "127.0.0.1"\nport = {port}\nframe_rate = 30\nlan_frame_rate = 30\nnetwork_profile_override = "auto"\nmax_connections = 2\nmax_connections_per_ip = 2\nmax_unauthenticated_connections = 2\nhandshake_timeout = 2.0\nclient_socket_timeout = 2.0\n\n[features]\nenable_metrics = false\nenable_health_checks = false\nenable_capture_producer = false\nenable_parallel_encoding = false\nenable_region_detection = false\nenable_websocket = false\nenable_tight_extensions = false\nenable_tight_encoding = true\nenable_jpeg_encoding = false\nenable_zrle_encoding = true\nenable_copyrect_encoding = false\n''',
+        encoding="utf-8",
+    )
+
+    server = server_module.VNCServerV3(config)
+    thread = threading.Thread(target=server.start, name="test-vnc-encoding-matrix")
+    thread.start()
+    client = socket.create_connection(("127.0.0.1", port), timeout=2.0)
+    client.settimeout(2.0)
+    try:
+        _handshake(client)
+        for encoding in (0, 2, 5, 6, 16, 7, 0):
+            _set_encodings(client, [encoding, 0] if encoding != 0 else [0])
+            _request_full(client)
+            _read_and_validate_payload(client, encoding)
+    finally:
+        client.close()
+        server.shutdown_handler.shutdown()
+        thread.join(timeout=5.0)
+        assert not thread.is_alive()

From 32cc812cfc848aca1938b3c5f31ccf7c2d2f644c Mon Sep 17 00:00:00 2001
From: xulek <xulek@users.noreply.github.com>
Date: Mon, 7 Sep 2026 07:41:23 +0200
Subject: [PATCH 10/20] test(v3.3): add Windows DXCam private-API contract

---
 tests/test_dxgi_dxcam_contract.py | 36 +++++++++++++++++++++++++++++++
 1 file changed, 36 insertions(+)
 create mode 100644 tests/test_dxgi_dxcam_contract.py

diff --git a/tests/test_dxgi_dxcam_contract.py b/tests/test_dxgi_dxcam_contract.py
new file mode 100644
index 0000000..27ce399
--- /dev/null
+++ b/tests/test_dxgi_dxcam_contract.py
@@ -0,0 +1,36 @@
+"""Contract tests for the optional DXCam integration used by v3.3.
+
+These tests do not create a Desktop Duplication session, so they can run on a
+headless GitHub-hosted Windows runner. Their purpose is to detect an upstream
+DXCam private-API change before it silently disables native metadata capture.
+"""
+
+from __future__ import annotations
+
+import os
+
+import pytest
+
+
+@pytest.mark.skipif(os.name != "nt", reason="DXCam contract is Windows-only")
+def test_dxcam_private_metadata_contract():
+    pytest.importorskip("dxcam")
+
+    from dxcam.core.dxgi_duplicator import DXGIDuplicator
+    from dxcam.dxcam import DXCamera
+    from dxcam._libs.dxgi import IDXGIOutputDuplication
+
+    assert hasattr(DXGIDuplicator, "update_frame")
+
+    # DXCamera stores the active duplicator on this private attribute. The v3.3
+    # hook intentionally checks it dynamically and falls back to software diff
+    # if it disappears, but CI should still make such an upstream change loud.
+    annotations = getattr(DXCamera, "__annotations__", {})
+    assert "_duplicator" in annotations or "_duplicator" in vars(DXCamera)
+
+    method_names = {
+        getattr(method, "name", None)
+        for method in getattr(IDXGIOutputDuplication, "_methods_", ())
+    }
+    assert "GetFrameDirtyRects" in method_names
+    assert "GetFrameMoveRects" in method_names

From 98b5089d6ebbc5b668d0cdf8a4b2f9702fc3cdc3 Mon Sep 17 00:00:00 2001
From: xulek <xulek@users.noreply.github.com>
Date: Mon, 7 Sep 2026 07:42:03 +0200
Subject: [PATCH 11/20] test(v3.3): harden DXCam private-API contract check

---
 tests/test_dxgi_dxcam_contract.py | 8 ++------
 1 file changed, 2 insertions(+), 6 deletions(-)

diff --git a/tests/test_dxgi_dxcam_contract.py b/tests/test_dxgi_dxcam_contract.py
index 27ce399..44b0da5 100644
--- a/tests/test_dxgi_dxcam_contract.py
+++ b/tests/test_dxgi_dxcam_contract.py
@@ -7,6 +7,7 @@
 
 from __future__ import annotations
 
+import inspect
 import os
 
 import pytest
@@ -21,12 +22,7 @@ def test_dxcam_private_metadata_contract():
     from dxcam._libs.dxgi import IDXGIOutputDuplication
 
     assert hasattr(DXGIDuplicator, "update_frame")
-
-    # DXCamera stores the active duplicator on this private attribute. The v3.3
-    # hook intentionally checks it dynamically and falls back to software diff
-    # if it disappears, but CI should still make such an upstream change loud.
-    annotations = getattr(DXCamera, "__annotations__", {})
-    assert "_duplicator" in annotations or "_duplicator" in vars(DXCamera)
+    assert "_duplicator" in inspect.getsource(DXCamera.__init__)
 
     method_names = {
         getattr(method, "name", None)

From ef965467d0812c7c5bd63038923f4ddbe7ca9c14 Mon Sep 17 00:00:00 2001
From: xulek <xulek@users.noreply.github.com>
Date: Mon, 7 Sep 2026 07:42:20 +0200
Subject: [PATCH 12/20] ci(v3.3): validate DXCam contract with performance
 extras on Windows

---
 .github/workflows/ci.yml | 27 ++++++++++++++++++++++++++-
 1 file changed, 26 insertions(+), 1 deletion(-)

diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 84d5e2a..72ea920 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -55,6 +55,31 @@ jobs:
       - name: Run test suite
         run: python -m pytest -q
 
+  dxgi-contract:
+    name: DXGI contract · Windows · Python 3.13
+    runs-on: windows-latest
+    timeout-minutes: 15
+
+    steps:
+      - name: Check out repository
+        uses: actions/checkout@v7
+
+      - name: Set up Python
+        uses: actions/setup-python@v7
+        with:
+          python-version: "3.13"
+          cache: pip
+          cache-dependency-path: pyproject.toml
+
+      - name: Upgrade packaging tools
+        run: python -m pip install --upgrade pip setuptools wheel
+
+      - name: Install performance and test dependencies
+        run: python -m pip install -e ".[dev,performance]"
+
+      - name: Verify DXCam private integration contract
+        run: python -m pytest -q tests/test_dxgi_dxcam_contract.py tests/test_dxgi_metadata.py
+
   coverage:
     name: Coverage
     runs-on: ubuntu-latest
@@ -91,7 +116,7 @@ jobs:
 
   build:
     name: Build package
-    needs: test
+    needs: [test, dxgi-contract]
     runs-on: ubuntu-latest
     timeout-minutes: 15
 

From e90d8a0e0982fa98685420ef12800ff198d693bd Mon Sep 17 00:00:00 2001
From: xulek <xulek@users.noreply.github.com>
Date: Mon, 7 Sep 2026 07:43:48 +0200
Subject: [PATCH 13/20] docs(v3.3): document native DXGI metadata and benchmark

---
 docs/performance.md | 41 ++++++++++++++++++++++++++++++++++++-----
 1 file changed, 36 insertions(+), 5 deletions(-)

diff --git a/docs/performance.md b/docs/performance.md
index 293b5b0..5f4931d 100644
--- a/docs/performance.md
+++ b/docs/performance.md
@@ -8,7 +8,7 @@ PyVNCServer optimizes the path from desktop capture to encoded rectangle rather
 
 | Backend | Platform | Role |
 | --- | --- | --- |
-| DXCam / DXGI | Windows | optional fast Desktop Duplication capture path |
+| DXCam / DXGI | Windows | optional fast Desktop Duplication capture path with native dirty/move metadata in v3.3 |
 | MSS | cross-platform | portable primary fallback |
 | Pillow ImageGrab | platform dependent | fallback capture path |
 
@@ -18,16 +18,29 @@ Install the performance extras:
 python -m pip install -e ".[performance]"
 ```
 
+PyVNCServer 3.3 requires DXCam `0.3.0+` for the native metadata integration.
+
 ## Shared capture producer
 
 A server-wide producer captures once and distributes framebuffer generations to sessions. This avoids N clients causing N independent desktop captures.
 
-## Changed regions
+## Native DXGI changed regions
+
+On an unscaled, unrotated full-output DXCam capture, v3.3 reads Desktop Duplication metadata directly from the active `IDXGIOutputDuplication` object:
+
+- `GetFrameDirtyRects` identifies framebuffer regions whose pixel contents changed;
+- `GetFrameMoveRects` identifies regions copied from another framebuffer location and exposes them as CopyRect hints;
+- a DXGI timeout/no-new-frame is treated as an authoritative empty update;
+- any metadata read/validation failure returns to the existing software change detector instead of assuming the screen is unchanged.
 
-Incremental requests benefit from limiting work to changed regions. When a backend cannot provide native dirty metadata, PyVNCServer performs change detection above the backend.
+Native metadata is deliberately disabled for scaled, rotated or cropped DXCam captures until coordinate translation is implemented. MSS and Pillow also continue to use software change detection.
 
-!!! note
-    Native DXGI dirty/move rectangle harvesting is not yet implemented in the current DXCam integration. CopyRect/dirty-region opportunities therefore depend on metadata available to the higher layers.
+!!! important
+    Native metadata is an optimization, never a correctness requirement. An ambiguous native result becomes `dirty_regions = None`, which explicitly activates the software differ.
+
+### Move-rectangle safety in v3.3
+
+Move destinations are currently included in the dirty pixel list even when a CopyRect hint is emitted. This is intentionally conservative: clients without CopyRect support and clients that skip capture generations still converge to the correct framebuffer. Once the real-client interoperability matrix has broader coverage, the redundant pixel update can be removed for clients that are exactly one generation behind and advertise CopyRect.
 
 ## Request coalescing
 
@@ -59,6 +72,8 @@ Relevant LAN settings include zlib/ZRLE compression levels, raw thresholds and J
 
 ## Benchmarks
 
+Encoder/capture benchmarks:
+
 ```bash
 PYTHONPATH=src python benchmarks/benchmark_encoders.py
 PYTHONPATH=src python benchmarks/benchmark_screen_capture.py
@@ -66,9 +81,25 @@ PYTHONPATH=src python benchmarks/benchmark_screen_capture_methods.py
 PYTHONPATH=src python benchmarks/benchmark_lan_latency.py
 ```
 
+### DXGI metadata benchmark
+
+On a real Windows desktop with the performance extra installed:
+
+```powershell
+python benchmarks/benchmark_dxgi_metadata.py --frames 300 --fps 60
+```
+
+The diagnostic reports:
+
+- native metadata hit rate vs software-diff fallback rate;
+- average and p95 capture time;
+- average dirty/move rectangle count;
+- approximate changed framebuffer area.
+
 For meaningful numbers:
 
 - benchmark on the target OS/GPU/display setup;
+- test idle desktop, text editing, window movement, scrolling and video separately;
 - separate capture time from encode time;
 - test full-screen and small-region updates;
 - include the actual viewer over the intended network path;

From 7a534faa4aab7eab8c83653da8f41f73bc76404e Mon Sep 17 00:00:00 2001
From: xulek <xulek@users.noreply.github.com>
Date: Mon, 7 Sep 2026 07:45:12 +0200
Subject: [PATCH 14/20] docs(v3.3): document live encoding interoperability
 regression

---
 docs/ultravnc.md | 18 ++++++++++++++++--
 1 file changed, 16 insertions(+), 2 deletions(-)

diff --git a/docs/ultravnc.md b/docs/ultravnc.md
index ae25897..6affa93 100644
--- a/docs/ultravnc.md
+++ b/docs/ultravnc.md
@@ -51,6 +51,20 @@ These are useful control encodings when isolating a compatibility problem:
 - **Zlib**: compressed full rectangles;
 - **ZRLE**: tiled zlib/RLE, usually a good general-purpose option.
 
+## v3.3 interoperability regression suite
+
+Version 3.3 adds a real TCP/RFB regression test that keeps one connection open while changing the advertised encoding set. The test parses the actual rectangle payload rather than only checking that the socket remained connected.
+
+The automated sequence is:
+
+```text
+Raw → RRE → Hextile → Zlib → ZRLE → Tight → Raw
+```
+
+For each step it verifies the rectangle header encoding ID and the corresponding wire format. This specifically guards against failures such as advertising RRE while sending Raw bytes, stale zlib state after an encoding switch, or a connection reset caused by stream desynchronization.
+
+This test is not a substitute for running the UltraVNC binary itself: client-specific decoder behavior still needs manual validation before a release. It does, however, make the protocol-level failure modes reproducible in Linux and Windows CI.
+
 ## What to log
 
 Run:
@@ -68,9 +82,9 @@ Useful lines include:
 - selected encoding per update/region;
 - socket reset/timeout messages.
 
-## Suggested compatibility matrix
+## Suggested manual compatibility matrix
 
-After a code change to an encoder, test a single connection while switching in this order:
+After a code change to an encoder, test a single UltraVNC connection while switching in this order:
 
 ```text
 Raw → Hextile → Zlib → ZRLE → Tight → RRE → Auto

From d51b0e6ee03b9344d65c2b21262561ee3e134ac2 Mon Sep 17 00:00:00 2001
From: xulek <xulek@users.noreply.github.com>
Date: Mon, 7 Sep 2026 07:52:21 +0200
Subject: [PATCH 15/20] test(v3.3): add real UltraVNC Viewer smoke script

---
 scripts/ultravnc_smoke.ps1 | 128 +++++++++++++++++++++++++++++++++++++
 1 file changed, 128 insertions(+)
 create mode 100644 scripts/ultravnc_smoke.ps1

diff --git a/scripts/ultravnc_smoke.ps1 b/scripts/ultravnc_smoke.ps1
new file mode 100644
index 0000000..f2caeb5
--- /dev/null
+++ b/scripts/ultravnc_smoke.ps1
@@ -0,0 +1,128 @@
+<#
+.SYNOPSIS
+    Smoke-test PyVNCServer with the real UltraVNC Viewer executable.
+
+.DESCRIPTION
+    Opens a fresh UltraVNC Viewer connection for each requested encoding and
+    keeps it alive for a short observation window. A viewer that exits before
+    the window ends is treated as a failed interoperability check.
+
+    Run PyVNCServer separately before starting this script. The default target
+    is the safe local listener at 127.0.0.1:5900 and the viewer is always
+    launched in view-only mode.
+
+.EXAMPLE
+    .\scripts\ultravnc_smoke.ps1 `
+      -ViewerPath 'C:\Program Files\uvnc bvba\UltraVNC\vncviewer.exe'
+
+.EXAMPLE
+    .\scripts\ultravnc_smoke.ps1 -Encodings raw,rre,hextile,zlib,tight -SecondsPerEncoding 5
+#>
+
+[CmdletBinding()]
+param(
+    [Parameter(Mandatory = $false)]
+    [string]$ViewerPath = "$env:ProgramFiles\uvnc bvba\UltraVNC\vncviewer.exe",
+
+    [Parameter(Mandatory = $false)]
+    [string]$ServerHost = "127.0.0.1",
+
+    [Parameter(Mandatory = $false)]
+    [ValidateRange(1, 65535)]
+    [int]$Port = 5900,
+
+    [Parameter(Mandatory = $false)]
+    [ValidateRange(1, 120)]
+    [int]$SecondsPerEncoding = 4,
+
+    [Parameter(Mandatory = $false)]
+    [ValidateSet("raw", "rre", "corre", "hextile", "zlib", "zlibhex", "tight", "ultra")]
+    [string[]]$Encodings = @("raw", "rre", "hextile", "zlib", "tight"),
+
+    [Parameter(Mandatory = $false)]
+    [string]$LogDirectory = (Join-Path $env:TEMP "pyvncserver-ultravnc")
+)
+
+$ErrorActionPreference = "Stop"
+
+if (-not (Test-Path -LiteralPath $ViewerPath -PathType Leaf)) {
+    throw "UltraVNC Viewer not found: $ViewerPath"
+}
+
+New-Item -ItemType Directory -Force -Path $LogDirectory | Out-Null
+
+# UltraVNC accepts host::port for an explicit TCP port. Keep the short host
+# form for the standard 5900 port to match the most common local setup.
+$Target = if ($Port -eq 5900) { $ServerHost } else { "${ServerHost}::${Port}" }
+$Failures = @()
+
+Write-Host "PyVNCServer / UltraVNC interoperability smoke test"
+Write-Host "Viewer: $ViewerPath"
+Write-Host "Target: $Target"
+Write-Host "Encodings: $($Encodings -join ', ')"
+Write-Host ""
+
+foreach ($Encoding in $Encodings) {
+    $Timestamp = Get-Date -Format "yyyyMMdd-HHmmssfff"
+    $ViewerLog = Join-Path $LogDirectory "ultravnc-$Encoding-$Timestamp.log"
+    $Arguments = @(
+        "-connect", $Target,
+        "-encoding", $Encoding,
+        "-viewonly",
+        "-nocursorshape",
+        "-noremotecursor",
+        "-loglevel", "10",
+        "-logfile", $ViewerLog
+    )
+
+    Write-Host "[$Encoding] connecting..." -ForegroundColor Cyan
+    $Process = Start-Process `
+        -FilePath $ViewerPath `
+        -ArgumentList $Arguments `
+        -PassThru
+
+    try {
+        $Deadline = (Get-Date).AddSeconds($SecondsPerEncoding)
+        $ExitedEarly = $false
+        while ((Get-Date) -lt $Deadline) {
+            Start-Sleep -Milliseconds 200
+            $Process.Refresh()
+            if ($Process.HasExited) {
+                $ExitedEarly = $true
+                break
+            }
+        }
+
+        if ($ExitedEarly) {
+            $ExitCode = $Process.ExitCode
+            $Failures += $Encoding
+            Write-Host "[$Encoding] FAILED: viewer exited early (code $ExitCode)" -ForegroundColor Red
+            if (Test-Path -LiteralPath $ViewerLog) {
+                Write-Host "[$Encoding] UltraVNC log: $ViewerLog"
+            }
+            continue
+        }
+
+        Write-Host "[$Encoding] OK: connection stayed alive for $SecondsPerEncoding s" -ForegroundColor Green
+    }
+    finally {
+        $Process.Refresh()
+        if (-not $Process.HasExited) {
+            Stop-Process -Id $Process.Id -Force -ErrorAction SilentlyContinue
+            $Process.WaitForExit(3000) | Out-Null
+        }
+    }
+
+    Start-Sleep -Milliseconds 400
+}
+
+Write-Host ""
+if ($Failures.Count -gt 0) {
+    Write-Host "FAILED encodings: $($Failures -join ', ')" -ForegroundColor Red
+    Write-Host "Logs: $LogDirectory"
+    exit 1
+}
+
+Write-Host "All UltraVNC smoke checks passed." -ForegroundColor Green
+Write-Host "Logs: $LogDirectory"
+exit 0

From c7c33a3786ab89fd907b28605fc581f118889362 Mon Sep 17 00:00:00 2001
From: xulek <xulek@users.noreply.github.com>
Date: Mon, 7 Sep 2026 07:52:45 +0200
Subject: [PATCH 16/20] docs(v3.3): document real UltraVNC Viewer smoke test

---
 docs/ultravnc.md | 38 +++++++++++++++++++++++++++++++++++++-
 1 file changed, 37 insertions(+), 1 deletion(-)

diff --git a/docs/ultravnc.md b/docs/ultravnc.md
index 6affa93..2c65c26 100644
--- a/docs/ultravnc.md
+++ b/docs/ultravnc.md
@@ -63,7 +63,43 @@ Raw → RRE → Hextile → Zlib → ZRLE → Tight → Raw
 
 For each step it verifies the rectangle header encoding ID and the corresponding wire format. This specifically guards against failures such as advertising RRE while sending Raw bytes, stale zlib state after an encoding switch, or a connection reset caused by stream desynchronization.
 
-This test is not a substitute for running the UltraVNC binary itself: client-specific decoder behavior still needs manual validation before a release. It does, however, make the protocol-level failure modes reproducible in Linux and Windows CI.
+This test is not a substitute for running the UltraVNC binary itself: client-specific decoder behavior still needs validation before a release. It does, however, make the protocol-level failure modes reproducible in Linux and Windows CI.
+
+## Real UltraVNC Viewer smoke test
+
+UltraVNC Viewer exposes a command-line `-encoding` option, so v3.3 also ships a PowerShell smoke-test for the actual `vncviewer.exe` binary.
+
+Start PyVNCServer in one terminal:
+
+```powershell
+pyvncserver serve --log-level DEBUG
+```
+
+Then run in another PowerShell window:
+
+```powershell
+.\scripts\ultravnc_smoke.ps1 `
+  -ViewerPath 'C:\Program Files\uvnc bvba\UltraVNC\vncviewer.exe'
+```
+
+The default sequence is:
+
+```text
+raw → rre → hextile → zlib → tight
+```
+
+For every encoding the script launches a fresh view-only UltraVNC process, keeps it connected for a short observation period and treats an early viewer exit as failure. Viewer logs are written under `%TEMP%\pyvncserver-ultravnc`.
+
+A longer run is useful when validating a release candidate:
+
+```powershell
+.\scripts\ultravnc_smoke.ps1 `
+  -SecondsPerEncoding 10 `
+  -Encodings raw,rre,hextile,zlib,tight
+```
+
+!!! note
+    The script proves that the real UltraVNC process can negotiate and keep the session alive. Visual correctness should still be checked while moving windows, scrolling and changing screen content, especially for Tight and RRE.
 
 ## What to log
 

From c4e4311b297d3ab9c93c69b4e2a426374a8d5cce Mon Sep 17 00:00:00 2001
From: xulek <xulek@users.noreply.github.com>
Date: Mon, 7 Sep 2026 08:46:08 +0200
Subject: [PATCH 17/20] fix(v3.3): avoid duplicate DXCam creation during
 healthcheck

---
 src/vnc_lib/capture_backends.py | 12 ++++++++----
 1 file changed, 8 insertions(+), 4 deletions(-)

diff --git a/src/vnc_lib/capture_backends.py b/src/vnc_lib/capture_backends.py
index 9ab22a1..109718f 100644
--- a/src/vnc_lib/capture_backends.py
+++ b/src/vnc_lib/capture_backends.py
@@ -217,10 +217,14 @@ def _metadata_is_usable(self, camera: Any) -> bool:
         return True
 
     def healthcheck(self) -> bool:
-        try:
-            return self.is_available() and self._get_camera() is not None
-        except Exception:
-            return False
+        # Do not create a DXCamera during backend probing. DXCam keeps one
+        # singleton camera per (device, output, backend), while PyVNCServer's
+        # capture producer runs on a dedicated thread. Creating the camera here
+        # on the server thread makes the producer call dxcam.create() again and
+        # triggers an "instance already exists" warning before first capture.
+        # The first real grab initializes the camera on the capture thread; any
+        # initialization failure is handled by the normal backend failover path.
+        return self.is_available()
 
     def build_metadata(self, width: int, height: int) -> CaptureMetadata:
         try:

From 83106b4350887580b68979a274425b167c9cb237 Mon Sep 17 00:00:00 2001
From: xulek <xulek@users.noreply.github.com>
Date: Mon, 7 Sep 2026 08:46:34 +0200
Subject: [PATCH 18/20] test(v3.3): prevent DXCam creation during backend probe

---
 tests/test_dxgi_metadata.py | 13 +++++++++++++
 1 file changed, 13 insertions(+)

diff --git a/tests/test_dxgi_metadata.py b/tests/test_dxgi_metadata.py
index 567f1f6..7a66ff2 100644
--- a/tests/test_dxgi_metadata.py
+++ b/tests/test_dxgi_metadata.py
@@ -211,3 +211,16 @@ def _get_dxcam_session(self):
     assert metadata.move_rects == [move]
     assert metadata.supports_dirty_regions is True
     assert metadata.supports_move_rects is True
+
+
+def test_dxcam_healthcheck_does_not_create_camera():
+    class FakeOwner:
+        logger = logging.getLogger("test.dxgi.healthcheck")
+        _dxcam_available = True
+
+        def _get_dxcam_session(self):
+            raise AssertionError("healthcheck must not create a DXCamera")
+
+    backend = DXCamCaptureBackend(FakeOwner())
+
+    assert backend.healthcheck() is True

From b7f0b81b2a08f2cb1f007683f386dfd9cd0a5842 Mon Sep 17 00:00:00 2001
From: xulek <xulek@users.noreply.github.com>
Date: Mon, 7 Sep 2026 08:50:04 +0200
Subject: [PATCH 19/20] fix(v3.3): make DXGI capability probe side-effect free

---
 src/vnc_lib/capture_backends.py | 29 ++++++++++++++++++++---------
 1 file changed, 20 insertions(+), 9 deletions(-)

diff --git a/src/vnc_lib/capture_backends.py b/src/vnc_lib/capture_backends.py
index 109718f..432edfe 100644
--- a/src/vnc_lib/capture_backends.py
+++ b/src/vnc_lib/capture_backends.py
@@ -227,6 +227,26 @@ def healthcheck(self) -> bool:
         return self.is_available()
 
     def build_metadata(self, width: int, height: int) -> CaptureMetadata:
+        # Capability probes use a 0x0 size and must be side-effect free. In
+        # particular, do not call dxcam.create() here: the real camera belongs
+        # to the capture-producer thread. Creating it during server startup
+        # would make the producer request the same DXCam singleton again.
+        if width <= 0 or height <= 0:
+            supported = (
+                bool(getattr(self.owner, "enable_dxgi_metadata", True))
+                and float(getattr(self.owner, "scale_factor", 1.0)) == 1.0
+                and self.is_available()
+            )
+            return CaptureMetadata(
+                backend_name=(
+                    "dxcam+dxgi-metadata" if supported else self.name
+                ),
+                dirty_regions=None,
+                move_rects=[],
+                supports_dirty_regions=supported,
+                supports_move_rects=supported,
+            )
+
         try:
             camera = self._get_camera()
         except Exception:
@@ -242,15 +262,6 @@ def build_metadata(self, width: int, height: int) -> CaptureMetadata:
                 supports_move_rects=False,
             )
 
-        if width <= 0 or height <= 0:
-            return CaptureMetadata(
-                backend_name="dxcam+dxgi-metadata",
-                dirty_regions=None,
-                move_rects=[],
-                supports_dirty_regions=True,
-                supports_move_rects=True,
-            )
-
         hints = self._metadata_hook.consume(width, height)
         if hints is None:
             return CaptureMetadata(

From 5a5e27a474c111b068ef32340c232a06b081b53c Mon Sep 17 00:00:00 2001
From: xulek <xulek@users.noreply.github.com>
Date: Mon, 7 Sep 2026 08:50:37 +0200
Subject: [PATCH 20/20] test(v3.3): prevent DXCam creation during capability
 probe

---
 tests/test_dxgi_metadata.py | 36 ++++++++++++++++++++++++++++++++++++
 1 file changed, 36 insertions(+)

diff --git a/tests/test_dxgi_metadata.py b/tests/test_dxgi_metadata.py
index 7a66ff2..9e0db54 100644
--- a/tests/test_dxgi_metadata.py
+++ b/tests/test_dxgi_metadata.py
@@ -224,3 +224,39 @@ def _get_dxcam_session(self):
     backend = DXCamCaptureBackend(FakeOwner())
 
     assert backend.healthcheck() is True
+
+
+def test_dxcam_capability_probe_does_not_create_camera():
+    class ProbeOwner:
+        logger = logging.getLogger("test.dxgi.capability-probe")
+        scale_factor = 1.0
+        enable_dxgi_metadata = True
+        _dxcam_available = True
+
+        def _get_dxcam_session(self):
+            raise AssertionError("capability probe must not create a DXCamera")
+
+    backend = DXCamCaptureBackend(ProbeOwner())
+    metadata = backend.build_metadata(0, 0)
+
+    assert metadata.backend_name == "dxcam+dxgi-metadata"
+    assert metadata.supports_dirty_regions is True
+    assert metadata.supports_move_rects is True
+
+
+def test_dxcam_capability_probe_respects_metadata_disable_flag():
+    class ProbeOwner:
+        logger = logging.getLogger("test.dxgi.capability-probe-disabled")
+        scale_factor = 1.0
+        enable_dxgi_metadata = False
+        _dxcam_available = True
+
+        def _get_dxcam_session(self):
+            raise AssertionError("capability probe must not create a DXCamera")
+
+    backend = DXCamCaptureBackend(ProbeOwner())
+    metadata = backend.build_metadata(0, 0)
+
+    assert metadata.backend_name == "dxcam"
+    assert metadata.supports_dirty_regions is False
+    assert metadata.supports_move_rects is False