tenant.openagents/omega

1#!/usr/bin/env python3
2"""Observe whether upstream Zed publishes an application accessibility tree.
3
4The owner direction of 2026-07-25 permits omitting assistive technology from
5this release *only where upstream Zed omits it too*. That is a condition, not a
6blanket exemption, so it has to be observed against a real installed upstream
7build before any waiver may cite it.
8
9This is a separate command on purpose. It launches upstream Zed, which writes
10to Zed's own data directory, and the installed-observation collector proves
11that Omega leaves that directory untouched. Running the probe inside the
12collector would put the harness's own writes inside the window the isolation
13check measures. Run this first, then capture the Zed data digest, then install
14and exercise the candidate.
15
16Usage:
17    observe-upstream-accessibility-parity --output <path>
18"""
19
20from __future__ import annotations
21
22import argparse
23import json
24import subprocess
25import sys
26import time
27from datetime import datetime, timezone
28from pathlib import Path
29
30PARITY_SCHEMA = "openagents.omega.upstream-accessibility-parity.v1"
31
32#: macOS supplies these window-frame controls itself. A window whose only
33#: direct children are these publishes no application accessibility tree.
34WINDOW_FRAME_CONTROLS = {"close button", "full screen button", "minimize button"}
35
36
37class ParityError(Exception):
38    pass
39
40
41def now() -> str:
42    return datetime.now(timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z")
43
44
45def osascript(script: str, timeout: int = 90) -> tuple[bool, str]:
46    try:
47        result = subprocess.run(
48            ["osascript", "-e", script], capture_output=True, text=True, timeout=timeout
49        )
50    except (OSError, subprocess.TimeoutExpired):
51        return False, ""
52    return result.returncode == 0, result.stdout.strip()
53
54
55#: See the same constant in `collect-omega-installed-observations`.
56NO_WINDOW_IN_TREE = -2
57
58#: Compiled `script/omega-window-list.swift`, set from `--window-list`.
59WINDOW_LIST_TOOL: Path | None = None
60
61
62def window_server_windows(owner: str) -> list[dict]:
63    """Ordinary windows the window server attributes to a process by name.
64
65    Asked only to tell two states apart: an application that is not showing a
66    window, and an application that is showing one the accessibility tree does
67    not carry. Without that distinction the probe reports the second as a
68    harness failure and no waiver can ever cite it.
69    """
70    if WINDOW_LIST_TOOL is None:
71        return []
72    try:
73        result = subprocess.run(
74            [str(WINDOW_LIST_TOOL)], capture_output=True, text=True, timeout=60
75        )
76    except (OSError, subprocess.SubprocessError):
77        return []
78    if result.returncode != 0:
79        return []
80    try:
81        windows = json.loads(result.stdout)
82    except json.JSONDecodeError:
83        return []
84    return [
85        window
86        for window in windows
87        if window.get("layer") == 0
88        and str(window.get("owner", "")).casefold() == owner.casefold()
89    ]
90
91
92def ax_children(process: str) -> tuple[bool, int, list[str]]:
93    script = f"""
94    tell application "System Events"
95        if not (exists process "{process}") then return "NOPROC"
96        tell process "{process}"
97            if (count of windows) is 0 then return "NOWINDOW"
98            set w to window 1
99            set out to ""
100            try
101                set out to out & "contents=" & (count of (entire contents of w)) & linefeed
102            on error
103                set out to out & "contents=-1" & linefeed
104            end try
105            try
106                repeat with e in UI elements of w
107                    set out to out & "child=" & (role description of e) & linefeed
108                end repeat
109            end try
110            return out
111        end tell
112    end tell
113    """
114    ok, output = osascript(script)
115    if ok and output == "NOWINDOW" and window_server_windows(process):
116        # The upstream build is running and the window server is showing its
117        # window. The accessibility tree carries no window for it. That is the
118        # weakest publication there is, and it is the observation this probe
119        # exists to make — not a failure to look.
120        return True, NO_WINDOW_IN_TREE, []
121    if not ok or output in {"NOPROC", "NOWINDOW"}:
122        return False, -1, []
123    contents = -1
124    children: list[str] = []
125    for line in output.split("\n"):
126        line = line.strip()
127        if line.startswith("contents="):
128            try:
129                contents = int(line.split("=", 1)[1])
130            except ValueError:
131                contents = -1
132        elif line.startswith("child="):
133            children.append(line.split("=", 1)[1])
134    return True, contents, children
135
136
137def main() -> int:
138    parser = argparse.ArgumentParser()
139    parser.add_argument("--app", type=Path, default=Path("/Applications/Zed.app"))
140    parser.add_argument("--process", default="zed")
141    parser.add_argument(
142        "--window-list",
143        type=Path,
144        default=None,
145        help=(
146            "compiled script/omega-window-list.swift. Lets the probe tell an "
147            "application showing no window from one whose window the "
148            "accessibility tree does not carry."
149        ),
150    )
151    parser.add_argument(
152        "--owner",
153        default="Zed",
154        help="the window-server owner name of the upstream build",
155    )
156    parser.add_argument("--output", type=Path, required=True)
157    args = parser.parse_args()
158
159    global WINDOW_LIST_TOOL
160    WINDOW_LIST_TOOL = args.window_list
161
162    if not args.app.is_dir():
163        raise ParityError(f"{args.app} is not installed, so parity cannot be observed")
164    version = subprocess.run(
165        [
166            "defaults",
167            "read",
168            str(args.app / "Contents" / "Info.plist"),
169            "CFBundleShortVersionString",
170        ],
171        capture_output=True,
172        text=True,
173        timeout=20,
174    )
175    if version.returncode != 0 or not version.stdout.strip():
176        raise ParityError("the upstream build has no readable version")
177
178    running = subprocess.run(
179        ["pgrep", "-f", f"{args.app}/Contents/MacOS/"], capture_output=True
180    )
181    already_running = running.returncode == 0
182    if not already_running:
183        subprocess.run(["open", "-a", str(args.app)], check=True, timeout=30)
184        time.sleep(15)
185    server_windows = window_server_windows(args.owner)
186    reachable, contents, children = ax_children(args.process)
187    if not already_running:
188        osascript(f'tell application "{args.app.stem}" to quit')
189        time.sleep(4)
190    if not reachable:
191        raise ParityError("the upstream window could not be read")
192
193    publishes = contents > 0 or any(
194        child not in WINDOW_FRAME_CONTROLS for child in children
195    )
196    record = {
197        "schema": PARITY_SCHEMA,
198        "observed_at": now(),
199        "product": "Zed",
200        "version": version.stdout.strip(),
201        "method": (
202            "System Events AXUIElement read of the front window: the size of "
203            "`entire contents` and the role description of every direct child, "
204            "with the window-server window list read alongside it so an "
205            "application showing no window is not confused with one whose "
206            "window the tree does not carry"
207        ),
208        "window_server_windows": [
209            {"id": window.get("id"), "name": window.get("name")}
210            for window in server_windows
211        ],
212        "observed": (
213            (
214                "the accessibility tree carries no window for the running "
215                f"process, while the window server is showing "
216                f"{len(server_windows)} window(s) for it; "
217                if contents == NO_WINDOW_IN_TREE
218                else ""
219            )
220            + f"entire contents = {contents}; direct children = "
221            f"{', '.join(children) if children else 'none'}"
222        ),
223        "publishes_accessibility_tree": publishes,
224    }
225    args.output.parent.mkdir(parents=True, exist_ok=True)
226    args.output.write_text(json.dumps(record, indent=2, sort_keys=True) + "\n", encoding="utf-8")
227    print(json.dumps(record, indent=2, sort_keys=True))
228    return 0
229
230
231if __name__ == "__main__":
232    try:
233        sys.exit(main())
234    except (ParityError, subprocess.SubprocessError, OSError) as error:
235        print(f"error: {error}", file=sys.stderr)
236        sys.exit(2)
237