feat: capture Windows UIA action evidence

README.md

@@ -163,6 +163,15 @@ with Recorder(
     input("Perform the task, then press Enter...")
 ```
 
+For native Windows applications, pass
+`capture_structural_observations=True`. Each mouse-down action then carries a
+versioned, JSON-safe UI Automation fingerprint through
+`Action.structural_observation` (role, name, AutomationId, window, ancestry,
+bounds, process, and supported native patterns). This evidence is stored in the
+existing `recording.db`; it does not create another recorder or upload path.
+Do not enable it for RDP or Citrix client windows: those sessions are observed
+externally through pixels and OCR.
+
 `owner` matches the application (macOS: window owner name; Windows: process
 executable name) and `title` optionally disambiguates among its windows; both
 are case-insensitive substrings, mirroring how `openadapt-flow`'s

openadapt_capture/__init__.py

@@ -56,6 +56,9 @@
     MouseUpEvent,
     ScreenEvent,
     ScreenFrameEvent,
+    StructuralBounds,
+    StructuralNode,
+    StructuralObservationV1,
 )
 
 # Event processing
@@ -155,6 +158,9 @@
     "MouseClickEvent",
     "MouseDoubleClickEvent",
     "MouseDragEvent",
+    "StructuralBounds",
+    "StructuralNode",
+    "StructuralObservationV1",
     # Keyboard events
     "KeyDownEvent",
     "KeyUpEvent",

openadapt_capture/capture.py

@@ -34,6 +34,7 @@
     MouseMoveEvent,
     MouseScrollEvent,
     MouseUpEvent,
+    StructuralObservationV1,
 )
 from openadapt_capture.processing import process_events
 
@@ -43,6 +44,22 @@
     from openadapt_capture.browser_events import BrowserEvent
 
 
+def _parse_structural_observation(raw: object) -> StructuralObservationV1 | None:
+    """Parse only the versioned native-observation contract.
+
+    Historical ``element_state`` JSON remains readable but is not silently
+    promoted to trusted structural evidence.
+    """
+    if not isinstance(raw, dict):
+        return None
+    if raw.get("schema_version") != 1 or raw.get("observer") != "windows_uia":
+        return None
+    try:
+        return StructuralObservationV1.model_validate(raw)
+    except Exception:
+        return None
+
+
 def _convert_action_event(db_event) -> PydanticActionEvent | None:
     """Convert a SQLAlchemy ActionEvent to a Pydantic event.
 
@@ -69,6 +86,9 @@ def _convert_action_event(db_event) -> PydanticActionEvent | None:
                 x=db_event.mouse_x or 0,
                 y=db_event.mouse_y or 0,
                 button=button,
+                structural_observation=_parse_structural_observation(
+                    getattr(db_event, "element_state", None)
+                ),
             )
         elif db_event.mouse_pressed is False:
             return MouseUpEvent(
@@ -369,6 +389,11 @@ def button(self) -> str | None:
             return btn.value if hasattr(btn, "value") else str(btn)
         return None
 
+    @property
+    def structural_observation(self) -> StructuralObservationV1 | None:
+        """Native accessibility evidence captured at action time, if present."""
+        return getattr(self.event, "structural_observation", None)
+
     @property
     def screenshot(self) -> "Image" | None:
         """Get the screenshot at the time of this action.

openadapt_capture/config.py

@@ -99,6 +99,7 @@ class Settings(BaseSettings):
     "capture_audio": "RECORD_AUDIO",
     "capture_images": "RECORD_IMAGES",
     "capture_window_data": "RECORD_WINDOW_DATA",
+    "capture_structural_observations": "RECORD_READ_ACTIVE_ELEMENT_STATE",
     "capture_browser_events": "RECORD_BROWSER_EVENTS",
     "window_owner": "RECORD_WINDOW_OWNER",
     "window_title": "RECORD_WINDOW_TITLE",
@@ -124,6 +125,7 @@ class RecordingConfig:
     capture_audio: bool | None = None
     capture_images: bool | None = None
     capture_window_data: bool | None = None
+    capture_structural_observations: bool | None = None
     capture_browser_events: bool | None = None
     # Window-scoped capture selectors (see window_capture.WindowTarget).
     window_owner: str | None = None

openadapt_capture/events.py

@@ -62,6 +62,37 @@ class BaseEvent(BaseModel):
     model_config = {"use_enum_values": True}
 
 
+class StructuralBounds(BaseModel):
+    """Element bounds in the native screen coordinate space."""
+
+    left: float
+    top: float
+    right: float
+    bottom: float
+
+
+class StructuralNode(BaseModel):
+    """Stable, JSON-safe subset of a native accessibility node."""
+
+    role: str = ""
+    name: str = ""
+    automation_id: str = ""
+    class_name: str = ""
+    bounds: StructuralBounds | None = None
+    supported_patterns: list[str] = Field(default_factory=list)
+
+
+class StructuralObservationV1(BaseModel):
+    """Versioned Windows UI Automation evidence captured with an action."""
+
+    schema_version: Literal[1] = 1
+    observer: Literal["windows_uia"] = "windows_uia"
+    target: StructuralNode
+    ancestors: list[StructuralNode] = Field(default_factory=list)
+    window_name: str = ""
+    process_id: int | None = None
+
+
 # =============================================================================
 # Mouse Events
 # =============================================================================
@@ -88,6 +119,7 @@ class MouseDownEvent(BaseEvent):
     x: float = Field(description="Mouse X position in pixels")
     y: float = Field(description="Mouse Y position in pixels")
     button: str = Field(description="Native mouse button name")
+    structural_observation: StructuralObservationV1 | None = None
 
 
 class MouseUpEvent(BaseEvent):
@@ -203,6 +235,7 @@ class MouseClickEvent(BaseEvent):
     x: float = Field(description="Mouse X position in pixels")
     y: float = Field(description="Mouse Y position in pixels")
     button: str = Field(description="Native mouse button name")
+    structural_observation: StructuralObservationV1 | None = None
     children: list[MouseDownEvent | MouseUpEvent] = Field(
         default_factory=list, description="Child events that were merged"
     )
@@ -219,6 +252,7 @@ class MouseDoubleClickEvent(BaseEvent):
     x: float = Field(description="Mouse X position in pixels")
     y: float = Field(description="Mouse Y position in pixels")
     button: str = Field(description="Native mouse button name")
+    structural_observation: StructuralObservationV1 | None = None
     children: list[MouseDownEvent | MouseUpEvent] = Field(
         default_factory=list, description="Child events that were merged"
     )
@@ -237,6 +271,7 @@ class MouseDragEvent(BaseEvent):
     dx: float = Field(description="Horizontal displacement (end_x - start_x)")
     dy: float = Field(description="Vertical displacement (end_y - start_y)")
     button: str = Field(description="Native mouse button name")
+    structural_observation: StructuralObservationV1 | None = None
     children: list[MouseDownEvent | MouseMoveEvent | MouseUpEvent] = Field(
         default_factory=list, description="Child events that were merged"
     )

openadapt_capture/processing.py

@@ -373,6 +373,7 @@ def calculate_distance(x1: float, y1: float, x2: float, y2: float) -> float:
                             x=down.x,
                             y=down.y,
                             button=down.button,
+                            structural_observation=down.structural_observation,
                             children=[down, up, next_down, next_up],
                         )
                         result.append(double_click)
@@ -387,6 +388,7 @@ def calculate_distance(x1: float, y1: float, x2: float, y2: float) -> float:
                     x=down.x,
                     y=down.y,
                     button=down.button,
+                    structural_observation=down.structural_observation,
                     children=[down, up],
                 )
                 result.append(single_click)
@@ -451,6 +453,7 @@ def calculate_distance(x1: float, y1: float, x2: float, y2: float) -> float:
                         dx=event.x - down_event.x,
                         dy=event.y - down_event.y,
                         button=down_event.button,
+                        structural_observation=down_event.structural_observation,
                         children=[down_event] + moves + [event],
                     )
                     result.append(drag)

openadapt_capture/recorder.py

@@ -856,9 +856,13 @@ def trigger_action_event(
     x = action_event_args.get("mouse_x")
     y = action_event_args.get("mouse_y")
     if x is not None and y is not None:
-        if config.RECORD_READ_ACTIVE_ELEMENT_STATE:
+        if (
+            config.RECORD_READ_ACTIVE_ELEMENT_STATE
+            and action_event_args.get("name") == "click"
+            and action_event_args.get("mouse_pressed") is True
+        ):
             # element lookup needs GLOBAL coordinates: translate afterwards.
-            element_state = window.get_active_element_state(x, y)
+            element_state = window.get_active_element_observation(x, y)
         else:
             element_state = {}
         action_event_args["element_state"] = element_state
@@ -2176,6 +2180,7 @@ def __init__(
         capture_audio: bool | None = None,
         capture_images: bool | None = None,
         capture_window_data: bool | None = None,
+        capture_structural_observations: bool | None = None,
         capture_browser_events: bool | None = None,
         capture_full_video: bool | None = None,
         video_encoding: str | None = None,
@@ -2209,6 +2214,7 @@ def __init__(
             capture_audio=capture_audio,
             capture_images=capture_images,
             capture_window_data=capture_window_data,
+            capture_structural_observations=capture_structural_observations,
             capture_browser_events=capture_browser_events,
             capture_full_video=capture_full_video,
             video_encoding=video_encoding,

openadapt_capture/window/__init__.py

@@ -93,3 +93,18 @@ def get_active_element_state(x: int, y: int) -> dict | None:
     except Exception as exc:
         logger.warning(f"{exc=}")
         return None
+
+
+def get_active_element_observation(x: int, y: int) -> dict | None:
+    """Return versioned structural evidence for the element at a point.
+
+    Platforms without a versioned observer deliberately return ``None``;
+    legacy element-state dictionaries are not promoted to structural evidence.
+    """
+    if impl is None or not hasattr(impl, "get_active_element_observation"):
+        return None
+    try:
+        return impl.get_active_element_observation(x, y)
+    except Exception as exc:
+        logger.warning(f"{exc=}")
+        return None

openadapt_capture/window/_windows.py

@@ -95,6 +95,115 @@ def get_active_element_state(x: int, y: int) -> dict:
     return properties
 
 
+def _safe_text(value: object) -> str:
+    """Return a stable textual property without leaking object reprs."""
+    if value is None:
+        return ""
+    if isinstance(value, str):
+        return value
+    return str(value)
+
+
+def _element_info_value(element: object, name: str) -> object | None:
+    info = getattr(element, "element_info", None)
+    if info is None:
+        return None
+    try:
+        return getattr(info, name, None)
+    except Exception:
+        return None
+
+
+def _safe_int(value: object) -> int | None:
+    try:
+        return int(value) if value is not None else None
+    except (TypeError, ValueError):
+        return None
+
+
+def _supported_patterns(element: object) -> list[str]:
+    patterns = (
+        ("invoke", "iface_invoke"),
+        ("value", "iface_value"),
+        ("selection_item", "iface_selection_item"),
+        ("toggle", "iface_toggle"),
+        ("expand_collapse", "iface_expand_collapse"),
+    )
+    supported = []
+    for label, attribute in patterns:
+        try:
+            if getattr(element, attribute, None) is not None:
+                supported.append(label)
+        except Exception:
+            continue
+    return supported
+
+
+def _structural_node(element: object) -> dict:
+    try:
+        properties = element.get_properties()
+    except Exception:
+        try:
+            properties = get_properties(element)
+        except Exception:
+            properties = {}
+
+    rectangle = properties.get("rectangle")
+    bounds = dictify_rect(rectangle) if rectangle is not None else None
+    texts = properties.get("texts") or []
+    name = properties.get("name") or (texts[0] if texts else "")
+    role = properties.get("control_type") or _element_info_value(
+        element, "control_type"
+    )
+    return {
+        "role": _safe_text(role),
+        "name": _safe_text(name),
+        "automation_id": _safe_text(
+            properties.get("automation_id")
+            or _element_info_value(element, "automation_id")
+        ),
+        "class_name": _safe_text(
+            properties.get("class_name")
+            or _element_info_value(element, "class_name")
+        ),
+        "bounds": bounds,
+        "supported_patterns": _supported_patterns(element),
+    }
+
+
+def get_active_element_observation(x: int, y: int) -> dict:
+    """Capture a versioned, JSON-safe UIA fingerprint at action time."""
+    active_window = get_active_window()
+    target = active_window.from_point(x, y)
+
+    ancestors = []
+    current = target
+    for _ in range(5):
+        try:
+            current = current.parent()
+        except Exception:
+            break
+        if current is None:
+            break
+        ancestors.append(_structural_node(current))
+        if current == active_window:
+            break
+
+    window_node = _structural_node(active_window)
+    process_id = _element_info_value(target, "process_id")
+    if process_id is None:
+        process_id = _element_info_value(active_window, "process_id")
+
+    return {
+        "schema_version": 1,
+        "observer": "windows_uia",
+        "target": _structural_node(target),
+        "ancestors": ancestors,
+        "window_name": window_node["name"],
+        "process_id": _safe_int(process_id),
+    }
+
+
 def get_active_window() -> "pywinauto.application.WindowSpecification":
     """Get the active window object.
 

pyproject.toml

@@ -44,6 +44,8 @@ dependencies = [
     "matplotlib>=3.10.8",
     # Native macOS window/accessibility observation via permissive PyObjC.
     "pyobjc-framework-ApplicationServices>=12.2.1; sys_platform == 'darwin'",
+    # Native Windows window resolution and UI Automation observation.
+    "pywinauto>=0.6.8; sys_platform == 'win32'",
 ]
 
 [project.optional-dependencies]