earakely-scale · Nov 7, 2026 · Nov 7, 2026 · Nov 7, 2026 · Nov 7, 2026 · Nov 7, 2026
@@ -228,7 +228,7 @@ jobs:
         if: always()
         run: |
           .venv/bin/python .github/scripts/check_skip_policy.py junit-integration-slow.xml \
-            --allow model_endpoint_configured remote_sandbox default_a2a_agent mcp_server_sources
+            --allow model_endpoint_configured remote_sandbox default_a2a_agent mcp_server_sources windows_media
       # The tests' times, for .github/scripts/slow_test_durations.py to rebalance the shards with.
       - uses: actions/upload-artifact@cf430e030ddbb5b0abf93d22962f4752f3646cd9 # v7.0.2
         if: always()

@@ -1,12 +1,15 @@
-"""The base class's defaults for the OS operations on Linux and macOS, which use POSIX alone. The driver runs as the
-session user inside the desktop session, so both act directly: a command inherits the session's environment, a file
-lands as that user."""
+"""The base class's defaults for the OS operations on Linux, macOS and Windows. The driver runs as the session user
+inside the desktop session, so both act directly: a command inherits the session's environment, a file lands as that
+user. A command runs in a process group of its own, which a timeout kills whole: on Windows, a new process group with no
+console window, killed with its children by taskkill."""
 from __future__ import annotations
 
 import asyncio
 import contextlib
 import os
 import signal
+import subprocess
+import sys
 import tempfile
 from pathlib import Path
 
@@ -22,6 +25,10 @@
 #: How much of a stream is read at a time.
 READ_BYTES = 65536
 _draining: set = set()
+WINDOWS = sys.platform == "win32"
+#: How a command starts: in a session, or on Windows, a process group, of its own, with no console window there.
+_SPAWN = ({"creationflags": subprocess.CREATE_NEW_PROCESS_GROUP | subprocess.CREATE_NO_WINDOW} if WINDOWS else
+          {"start_new_session": True})
 
 
 async def run_in_session(argv: list[str], cwd: str | None, env: dict[str, str] | None, timeout_s: float, *,
@@ -33,15 +40,14 @@ async def run_in_session(argv: list[str], cwd: str | None, env: dict[str, str] |
     try:
         process = await asyncio.create_subprocess_exec(
             *argv, cwd=workdir, env={**os.environ, **(env or {})}, stdin=asyncio.subprocess.DEVNULL,
-            stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, start_new_session=True)
+            stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, **_SPAWN)
     except (FileNotFoundError, PermissionError, NotADirectoryError) as e:
         raise InvalidArgument(f"cannot run {argv[0]!r}: {e.strerror or e}") from None
     outputs = [_Capped(output_bytes), _Capped(output_bytes)]
     readers = [asyncio.ensure_future(out.read(stream)) for out, stream in zip(outputs, (process.stdout, process.stderr))]
     timed_out = not await _exited(process, timeout_s)
     if timed_out:
-        with contextlib.suppress(ProcessLookupError):
-            os.killpg(process.pid, signal.SIGKILL)
+        await _killed(process.pid)
         await _exited(process, DRAIN_S)
     _, pending = await asyncio.wait(readers, timeout=DRAIN_S)
     for out in outputs:
@@ -54,6 +60,17 @@ async def run_in_session(argv: list[str], cwd: str | None, env: dict[str, str] |
                       stderr=bytes(stderr.kept), timed_out=timed_out, truncated=stdout.cut or stderr.cut)
 
 
+async def _killed(pid: int) -> None:
+    """The process group ``pid`` leads killed: on Windows, the process and its children."""
+    if WINDOWS:
+        killer = await asyncio.create_subprocess_exec("taskkill", "/F", "/T", "/PID", str(pid),
+                                                      stdout=asyncio.subprocess.DEVNULL, stderr=asyncio.subprocess.DEVNULL)
+        await killer.wait()
+        return
+    with contextlib.suppress(ProcessLookupError):
+        os.killpg(pid, signal.SIGKILL)
+
+
 async def _exited(process: asyncio.subprocess.Process, timeout_s: float) -> bool:
     """Whether the process exits within ``timeout_s``. Read from ``returncode``, which is set when the process exits:
     ``Process.wait()`` also waits for the pipes to close, which a background child can hold open."""

@@ -72,6 +72,25 @@
 READ_CHUNK = ("import base64,sys;f=open(sys.argv[1],'rb');f.seek(int(sys.argv[2]));"
               "print(base64.b64encode(f.read(int(sys.argv[3]))).decode())")
 CHUNK_BYTES = 512 << 10
+#: The probe's commands on a Windows computer, which has no POSIX shell or tools: Python one-liners on the python3 the
+#: fixture runs on, writing bytes, so output carries no CRLF, and taskkill and the account checks Windows has.
+WIN_ECHO = ("python3", "-c", "import sys;sys.stdout.buffer.write(b'probe\\n');sys.stderr.buffer.write(b'oops\\n');"
+            "sys.exit(3)")
+WIN_SLEEP = ("python3", "-c", "import time;time.sleep(30)")
+WIN_NOW = ("python3", "-c", "import time;print(time.time_ns())")
+WIN_CAT = ("python3", "-c", "import sys;sys.stdout.buffer.write(open(sys.argv[1],'rb').read())")
+WIN_TAIL = ("python3", "-c", "import sys;sys.stdout.buffer.write(b''.join(open(sys.argv[1],'rb').readlines()[-5:]))")
+WIN_REMOVE = ("python3", "-c", "import os,shutil,sys;[shutil.rmtree(p,ignore_errors=True) if os.path.isdir(p) else "
+              "os.path.exists(p) and os.remove(p) for p in sys.argv[1:]]")
+WIN_RMDIR = ("python3", "-c", "import os,sys;p=sys.argv[1];os.path.isdir(p) and not os.listdir(p) and os.rmdir(p)")
+#: Starts the fixture detached, its output to its log, in a process group of its own with no console window; its pid.
+WIN_START = ("python3", "-c", "import subprocess,sys;log=open(sys.argv[3],'wb');print(subprocess.Popen("
+             "[sys.executable,sys.argv[1],sys.argv[2]],stdout=log,stderr=subprocess.STDOUT,stdin=subprocess.DEVNULL,"
+             "creationflags=0x08000200).pid)")
+#: A file's owner, its path in the command, since PowerShell takes what follows -Command as more of the command.
+WIN_OWNER = "(Get-Acl -LiteralPath '{}').Owner"
+#: The well-known SID of the local Administrators group, which an unprivileged user's token holds no enabled entry of.
+ADMINISTRATORS = "S-1-5-32-544"
 #: The checks only ``full`` runs, recorded unverified when the fixture can't be staged.
 FULL_CHECKS = (("P3", "fixture colours land where the mapping says"), ("P4", "clicks show their effect"),
                ("P5", "refusals leave the fixture as it was"), ("P6", "typed and pressed keys arrive"),
@@ -98,6 +117,7 @@ def __init__(self, driver: Driver, session: ClientSession, ready_timeout_s: floa
         self.fixture_pid: str | None = None
         self.page = self.features.get("screen") == "page"
         self.fixture = "page_fixture.py" if self.page else "fixture.py"
+        self.windows = (self.features.get("platform") or {}).get("os") == "windows"
         self.desk = ""
         self.tab: str | None = None
         self.opened: list[str] = []
@@ -188,7 +208,7 @@ async def act(self, name: str, args: dict) -> None:
             raise AssertionError(f"{name} {args} failed: {result.content[0].text}")
 
     async def state(self) -> dict:
-        out = await self.driver.run("cat", f"{STAGE}/state/state.json")
+        out = await self._cat(f"{STAGE}/state/state.json")
         return json.loads(out["stdout"])
 
     async def until(self, holds: Callable[[Any], bool],
@@ -394,24 +414,26 @@ async def _exec(self) -> None:
         cap = self.driver.params(COMPUTER_EXEC).get("max_output_bytes")
 
         async def output() -> str | None:
-            out = await self.driver.run("sh", "-c", "echo probe; echo oops >&2; exit 3")
+            out = await self.driver.run(*(WIN_ECHO if self.windows else ("sh", "-c", "echo probe; echo oops >&2; exit 3")))
             want = {"exit_code": 3, "stdout": "probe\n", "stderr": "oops\n", "timed_out": False}
             return None if {k: out.get(k) for k in want} == want else f"it answered {out}"
 
         async def timeout() -> str | None:
             started = time.monotonic()
-            out = await self.driver.run("sh", "-c", "sleep 30", timeout_s=1)
+            out = await self.driver.run(*(WIN_SLEEP if self.windows else ("sh", "-c", "sleep 30")), timeout_s=1)
             took = time.monotonic() - started
             return None if out.get("timed_out") and took < 15 else f"after {took:.1f} s it answered {out}"
 
         async def once() -> str | None:
-            request = {"argv": ["sh", "-c", "date +%s%N"], "request_id": f"{PROBE_ROLE}-once-{time.time_ns()}"}
+            request = {"argv": list(WIN_NOW if self.windows else ("sh", "-c", "date +%s%N")),
+                       "request_id": f"{PROBE_ROLE}-once-{time.time_ns()}"}
             first = await self.driver.invoke(COMPUTER_EXEC, "run", request)
             again = await self.driver.invoke(COMPUTER_EXEC, "run", request)
             return None if first == again else f"it ran twice: {first} then {again}"
 
         async def capped() -> str | None:
-            out = await self.driver.run("sh", "-c", f"head -c {cap + 1000} /dev/zero | tr '\\0' a")
+            out = await self.driver.run(*(("python3", "-c", f"import sys;sys.stdout.buffer.write(b'a'*{cap + 1000})")
+                                          if self.windows else ("sh", "-c", f"head -c {cap + 1000} /dev/zero | tr '\\0' a")))
             return None if out.get("truncated") and len(out["stdout"]) == cap else (
                 f"{len(out['stdout'])} bytes came back, truncated {out.get('truncated')}")
 
@@ -546,28 +568,32 @@ async def _stage(self) -> str | None:
         added = await self.driver.add_file(f"{STAGE}/{self.fixture}", source)
         if "error" in added:
             return f"data/add refused it: {added['error'].get('message')}"
-        await self.driver.run("sh", "-c", f"rm -rf {STAGE}/state")
-        started = await self.driver.run(
-            "sh", "-c", f"python3 {STAGE}/{self.fixture} {STAGE}/state > {STAGE}/fixture.log 2>&1 & echo $!")
+        if self.windows:
+            await self.driver.run(*WIN_REMOVE, f"{STAGE}/state")
+            started = await self.driver.run(*WIN_START, f"{STAGE}/{self.fixture}", f"{STAGE}/state", f"{STAGE}/fixture.log")
+        else:
+            await self.driver.run("sh", "-c", f"rm -rf {STAGE}/state")
+            started = await self.driver.run(
+                "sh", "-c", f"python3 {STAGE}/{self.fixture} {STAGE}/state > {STAGE}/fixture.log 2>&1 & echo $!")
         self.fixture_pid = started["stdout"].strip() or None
         deadline = time.monotonic() + FIXTURE_START_S
         while time.monotonic() < deadline:
             if self.page and self.tab is None and (problem := await self._open_desk()):
                 return problem
-            layout = await self.driver.run("cat", f"{STAGE}/state/layout.json")
+            layout = await self._cat(f"{STAGE}/state/layout.json")
             if layout.get("exit_code") == 0 and layout["stdout"].strip():
                 self.layout = json.loads(layout["stdout"])
                 return None
-            state = await self.driver.run("cat", f"{STAGE}/state/state.json")
+            state = await self._cat(f"{STAGE}/state/state.json")
             if state.get("exit_code") == 0 and "error" in (failed := json.loads(state["stdout"] or "{}")):
                 return failed["error"]
             await asyncio.sleep(0.5)
-        log = await self.driver.run("tail", "-n", "5", f"{STAGE}/fixture.log")
+        log = await self.driver.run(*(WIN_TAIL if self.windows else ("tail", "-n", "5")), f"{STAGE}/fixture.log")
         return f"no layout after {FIXTURE_START_S:g} s: {log['stdout'].strip() or log['stderr'].strip()}"
 
     async def _open_desk(self) -> str | None:
         """Once the page fixture's server has its port, the desk opened in a new tab, which is the probe's."""
-        port = await self.driver.run("cat", f"{STAGE}/state/port")
+        port = await self._cat(f"{STAGE}/state/port")
         if port.get("exit_code") != 0 or not port["stdout"].strip():
             return None
         self.desk = f"http://127.0.0.1:{port['stdout'].strip()}/"
@@ -588,10 +614,14 @@ async def _unstage(self) -> None:
                         await self.tool("computer_tabs", {"action": "close", "tab": tab})
                         left.remove(tab)
         if self.fixture_pid:
-            await self.driver.run("kill", self.fixture_pid)
-        await self.driver.run("rm", "-rf", STAGE)
+            await self.driver.run(*(("taskkill", "/F", "/T", "/PID") if self.windows else ("kill",)), self.fixture_pid)
+        await self.driver.run(*(WIN_REMOVE if self.windows else ("rm", "-rf")), STAGE)
         if self.upload:
-            await self.driver.run("rm", "-f", f"{UPLOADS}/{self.upload}")
+            await self.driver.run(*(WIN_REMOVE if self.windows else ("rm", "-f")), f"{UPLOADS}/{self.upload}")
+
+    async def _cat(self, path: str) -> dict:
+        """``path``'s bytes as the computer's command answers them."""
+        return await self.driver.run(*(WIN_CAT if self.windows else ("cat",)), path)
 
     async def _base(self) -> tuple[str, float, object]:
         """A fresh frame of the fixture: its id, the display's scale, and its image."""
@@ -605,6 +635,21 @@ def _at(self, box: str, scale: float, dx: float | None = None, dy: float | None
                 math.floor((y + (h / 2 if dy is None else dy)) * scale))
 
     async def _security(self) -> None:
+        async def one_account() -> str | None:
+            account = (await self.driver.run("whoami"))["stdout"].strip()
+            owner = (await self.driver.run("powershell", "-NoProfile", "-Command",
+                                           WIN_OWNER.format(f"{STAGE}/{self.fixture}")))["stdout"].strip()
+            groups = (await self.driver.run("whoami", "/groups"))["stdout"]
+            if any(ADMINISTRATORS in line and "Enabled group" in line for line in groups.splitlines()):
+                return f"commands run as {account}, an administrator"
+            return None if owner.lower() == account.lower() else (
+                f"data/add wrote as {owner}, commands run as {account}")
+
+        if self.windows:
+            await self.check("P9", "commands and data/add run as one unprivileged user", one_account, COMPUTER_EXEC,
+                             *DATA_PLANE)
+            return
+
         async def one_user() -> str | None:
             uid = (await self.driver.run("id", "-u"))["stdout"].strip()
             owner = (await self.driver.run(*OWNER, f"{STAGE}/{self.fixture}"))["stdout"].strip()
@@ -993,8 +1038,9 @@ async def limited() -> str | None:
             for recording_id in made:
                 with contextlib.suppress(Exception):
                     await self.driver.invoke(COMPUTER_RECORDING, "end", {"recording_id": recording_id})
-                await self.driver.run("rm", "-rf", f"{RECORDINGS_DIR}/{recording_id}")
-            await self.driver.run("sh", "-c", f"rmdir {RECORDINGS_DIR} 2>/dev/null || true")
+                await self.driver.run(*(WIN_REMOVE if self.windows else ("rm", "-rf")), f"{RECORDINGS_DIR}/{recording_id}")
+            await self.driver.run(*((*WIN_RMDIR, RECORDINGS_DIR) if self.windows else
+                                    ("sh", "-c", f"rmdir {RECORDINGS_DIR} 2>/dev/null || true")))
 
     async def _segments_match(self, wire: dict, fps: int) -> str | None:
         """What is wrong with a recording without a grant, read back from the computer, or None."""

@@ -219,13 +219,13 @@ def _computer_require_init(self) -> None:
         if self._computer is None:
             raise TypeError(f"{type(self).__qualname__}.__init__ must call super().__init__()")
 
-    # The defaults for the OS operations on Linux and macOS, POSIX alike.
+    # The defaults for the OS operations on Linux, macOS and Windows.
 
-    @default_for("run_in_session", os=["linux", "macos"])
+    @default_for("run_in_session", os=["linux", "macos", "windows"])
     async def _computer_linux_run_in_session(self, argv: list, cwd: str | None, env: dict | None, timeout_s: float):
         return await _linux.run_in_session(argv, cwd, env, timeout_s, output_bytes=self.settings.exec_output_bytes)
 
-    @default_for("write_files", os=["linux", "macos"])
+    @default_for("write_files", os=["linux", "macos", "windows"])
     async def _computer_linux_write_files(self, files: dict) -> None:
         await asyncio.to_thread(_linux.write_files, files)
 

@@ -28,14 +28,17 @@ def as_dict(self) -> dict:
 
 def detect(environ: dict | None = None, os_release: Path = Path("/etc/os-release")) -> Platform:
     """The running platform: the OS, ``/etc/os-release``'s ID and VERSION_ID on Linux and the macOS version on macOS,
-    the machine, and the display: on Linux from ``XDG_SESSION_TYPE``, ``none`` for a computer with no display, else
-    ``DISPLAY`` (x11) or ``WAYLAND_DISPLAY`` (wayland); on macOS Aqua."""
+    Windows' marketing version (``11`` from build 22000) on Windows, the machine, and the display: on Linux from
+    ``XDG_SESSION_TYPE``, ``none`` for a computer with no display, else ``DISPLAY`` (x11) or ``WAYLAND_DISPLAY``
+    (wayland); on macOS Aqua; on Windows the Desktop Window Manager (dwm)."""
     environ = os.environ if environ is None else environ
     name = _OS.get(sys.platform, sys.platform)
     release = _os_release(os_release) if name == "linux" else {}
     if name == "macos":
         release = {"VERSION_ID": _platform.mac_ver()[0] or None}
-    display = "aqua" if name == "macos" else None
+    if name == "windows":
+        release = {"VERSION_ID": "11" if sys.getwindowsversion().build >= 22000 else "10"}
+    display = {"macos": "aqua", "windows": "dwm"}.get(name)
     if name == "linux":
         session = (environ.get("XDG_SESSION_TYPE") or "").strip().lower()
         if session in ("x11", "wayland", "none"):

@@ -34,6 +34,7 @@
     type_text,
 )
 from agentenv_protocol.computer.conformance import CheckResult, ProbeReport, run_probe  # noqa: E402
+from agentenv_protocol.computer import _linux  # noqa: E402
 from agentenv_protocol.computer.conformance import checks  # noqa: E402
 from agentenv_protocol.computer.conformance.client import Driver  # noqa: E402
 from agentenv_protocol import CONTROL_TOKEN_ENV  # noqa: E402
@@ -167,7 +168,7 @@ async def test_each_app_a_driver_fronts_that_exports_is_asked_for_its_export_onc
 
 @pytest.mark.asyncio
 async def test_a_driver_without_exec_or_data_add_passes_basic_and_leaves_full_unverified():
-    report = await _probe(MinimalDriver(FakeScreen(640, 400), platform=Platform(os="windows")))
+    report = await _probe(MinimalDriver(FakeScreen(640, 400), platform=Platform(os="freebsd")))
     assert report.passed, _failed(report)
     unverified = [c for c in report.checks if c.status == "unverified"]
     assert {c.check for c in unverified} == {"P3", "P4", "P5", "P6", "P7", "P9"}
@@ -176,6 +177,26 @@ async def test_a_driver_without_exec_or_data_add_passes_basic_and_leaves_full_un
     assert report.extensions["urn:agentenv:computer/v1"] == "verified"
 
 
+@pytest.mark.asyncio
+async def test_a_windows_computer_is_probed_with_the_commands_windows_has(monkeypatch):
+    """Its exec checks run Python one-liners rather than a POSIX shell; here they run on this machine's python3."""
+    sent = []
+    run = _linux.run_in_session
+
+    async def recorded(argv, *args, **kwargs):
+        sent.append(list(argv))
+        return await run(argv, *args, **kwargs)
+
+    monkeypatch.setattr(_linux, "run_in_session", recorded)
+    report = await _probe(MinimalDriver(FakeScreen(640, 400), platform=Platform(os="windows", arch="amd64")), "basic")
+
+    assert report.passed, _failed(report)
+    exec_checks = [c for c in report.checks if "urn:agentenv:computer-exec/v1" in c.extensions]
+    assert len(exec_checks) >= 4 and all(c.status == "passed" for c in exec_checks)
+    assert list(checks.WIN_ECHO) in sent and list(checks.WIN_SLEEP) in sent and list(checks.WIN_NOW) in sent
+    assert not [argv for argv in sent if argv[0] == "sh"]
+
+
 class NeverReady(FixtureDriver):
     async def ready_checks(self):
         return [ReadyCheck("session", False, "no desktop session")]