ci: diagnose Windows ARM64 conformance and key the platform golden by family

scripts/conformance/conpty_smoke.py

@@ -0,0 +1,218 @@
+"""Spawn a shell through ConPTY without Pebrel and report its first output.
+
+Separates runtime failures (conpty.dll, OpenConsole.exe, the shell itself)
+from application failures when a platform renders no initial shell output.
+Both the side-loaded pair and the in-box kernel32 ConPTY are exercised.
+"""
+
+from __future__ import annotations
+
+import argparse
+import ctypes
+from ctypes import wintypes as wt
+import json
+import os
+import shutil
+import sys
+import threading
+import time
+
+PROC_THREAD_ATTRIBUTE_PSEUDOCONSOLE = 0x00020016
+EXTENDED_STARTUPINFO_PRESENT = 0x00080000
+STARTF_USESTDHANDLES = 0x00000100
+PSEUDOCONSOLE_WIN32_INPUT_MODE = 0x4
+
+
+class Coord(ctypes.Structure):
+    _fields_ = [("X", ctypes.c_short), ("Y", ctypes.c_short)]
+
+
+class StartupInfo(ctypes.Structure):
+    _fields_ = [
+        ("cb", wt.DWORD), ("lpReserved", wt.LPWSTR), ("lpDesktop", wt.LPWSTR),
+        ("lpTitle", wt.LPWSTR), ("dwX", wt.DWORD), ("dwY", wt.DWORD),
+        ("dwXSize", wt.DWORD), ("dwYSize", wt.DWORD), ("dwXCountChars", wt.DWORD),
+        ("dwYCountChars", wt.DWORD), ("dwFillAttribute", wt.DWORD),
+        ("dwFlags", wt.DWORD), ("wShowWindow", wt.WORD), ("cbReserved2", wt.WORD),
+        ("lpReserved2", ctypes.POINTER(wt.BYTE)), ("hStdInput", wt.HANDLE),
+        ("hStdOutput", wt.HANDLE), ("hStdError", wt.HANDLE),
+    ]
+
+
+class StartupInfoEx(ctypes.Structure):
+    _fields_ = [("StartupInfo", StartupInfo), ("lpAttributeList", wt.LPVOID)]
+
+
+class ProcessInfo(ctypes.Structure):
+    _fields_ = [("hProcess", wt.HANDLE), ("hThread", wt.HANDLE),
+                ("dwProcessId", wt.DWORD), ("dwThreadId", wt.DWORD)]
+
+
+kernel = ctypes.WinDLL("kernel32", use_last_error=True)
+kernel.CreatePipe.argtypes = [ctypes.POINTER(wt.HANDLE), ctypes.POINTER(wt.HANDLE), wt.LPVOID, wt.DWORD]
+kernel.CreatePipe.restype = wt.BOOL
+kernel.ReadFile.argtypes = [wt.HANDLE, wt.LPVOID, wt.DWORD, ctypes.POINTER(wt.DWORD), wt.LPVOID]
+kernel.ReadFile.restype = wt.BOOL
+kernel.WriteFile.argtypes = [wt.HANDLE, wt.LPCVOID, wt.DWORD, ctypes.POINTER(wt.DWORD), wt.LPVOID]
+kernel.WriteFile.restype = wt.BOOL
+kernel.CloseHandle.argtypes = [wt.HANDLE]
+kernel.CloseHandle.restype = wt.BOOL
+kernel.InitializeProcThreadAttributeList.argtypes = [wt.LPVOID, wt.DWORD, wt.DWORD, ctypes.POINTER(ctypes.c_size_t)]
+kernel.InitializeProcThreadAttributeList.restype = wt.BOOL
+kernel.UpdateProcThreadAttribute.argtypes = [wt.LPVOID, wt.DWORD, ctypes.c_size_t, wt.LPVOID, ctypes.c_size_t, wt.LPVOID, wt.LPVOID]
+kernel.UpdateProcThreadAttribute.restype = wt.BOOL
+kernel.DeleteProcThreadAttributeList.argtypes = [wt.LPVOID]
+kernel.CreateProcessW.argtypes = [wt.LPCWSTR, wt.LPWSTR, wt.LPVOID, wt.LPVOID, wt.BOOL, wt.DWORD, wt.LPVOID, wt.LPCWSTR, wt.LPVOID, ctypes.POINTER(ProcessInfo)]
+kernel.CreateProcessW.restype = wt.BOOL
+kernel.WaitForSingleObject.argtypes = [wt.HANDLE, wt.DWORD]
+kernel.WaitForSingleObject.restype = wt.DWORD
+kernel.GetExitCodeProcess.argtypes = [wt.HANDLE, ctypes.POINTER(wt.DWORD)]
+kernel.GetExitCodeProcess.restype = wt.BOOL
+kernel.TerminateProcess.argtypes = [wt.HANDLE, wt.UINT]
+kernel.TerminateProcess.restype = wt.BOOL
+
+CreateFn = ctypes.WINFUNCTYPE(ctypes.c_long, Coord, wt.HANDLE, wt.HANDLE, wt.DWORD, ctypes.POINTER(wt.LPVOID))
+CloseFn = ctypes.WINFUNCTYPE(None, wt.LPVOID)
+
+
+def api(library: str | None) -> tuple[CreateFn, CloseFn, str]:
+    module = ctypes.WinDLL(library, use_last_error=True) if library else kernel
+    create = CreateFn(("CreatePseudoConsole", module))
+    close = CloseFn(("ClosePseudoConsole", module))
+    return create, close, library or "kernel32"
+
+
+def run_case(library: str | None, command: str, flags: int, wait_seconds: float) -> dict[str, object]:
+    result: dict[str, object] = {"library": library or "kernel32", "command": command, "flags": flags}
+    try:
+        create, close, _ = api(library)
+    except OSError as error:
+        result["error"] = f"load failed: {error}"
+        return result
+    conout_read, conout_write = wt.HANDLE(), wt.HANDLE()
+    conin_read, conin_write = wt.HANDLE(), wt.HANDLE()
+    if not kernel.CreatePipe(ctypes.byref(conout_read), ctypes.byref(conout_write), None, 0) or \
+       not kernel.CreatePipe(ctypes.byref(conin_read), ctypes.byref(conin_write), None, 0):
+        result["error"] = f"CreatePipe failed: {ctypes.get_last_error()}"
+        return result
+    hpc = wt.LPVOID()
+    started = time.monotonic()
+    hresult = create(Coord(80, 25), conin_read, conout_write, flags, ctypes.byref(hpc))
+    result["create_hresult"] = f"{hresult & 0xFFFFFFFF:#010x}"
+    result["create_ms"] = round((time.monotonic() - started) * 1000)
+    if hresult != 0:
+        return result
+    # The side-loaded host answers the same primed DA1 reply Pebrel writes.
+    written = wt.DWORD()
+    kernel.WriteFile(conin_write, b"\x1b[?61c", 6, ctypes.byref(written), None)
+
+    size = ctypes.c_size_t(0)
+    kernel.InitializeProcThreadAttributeList(None, 1, 0, ctypes.byref(size))
+    attributes = ctypes.create_string_buffer(size.value)
+    if not kernel.InitializeProcThreadAttributeList(attributes, 1, 0, ctypes.byref(size)):
+        result["error"] = f"InitializeProcThreadAttributeList failed: {ctypes.get_last_error()}"
+        return result
+    if not kernel.UpdateProcThreadAttribute(attributes, 0, PROC_THREAD_ATTRIBUTE_PSEUDOCONSOLE,
+                                            hpc, ctypes.sizeof(wt.LPVOID), None, None):
+        result["error"] = f"UpdateProcThreadAttribute failed: {ctypes.get_last_error()}"
+        return result
+    startup = StartupInfoEx()
+    startup.StartupInfo.cb = ctypes.sizeof(startup)
+    # Null standard handles with USESTDHANDLES: the client binds to the
+    # pseudoconsole instead of inheriting this process's console or pipes.
+    startup.StartupInfo.dwFlags = STARTF_USESTDHANDLES
+    startup.lpAttributeList = ctypes.cast(attributes, wt.LPVOID)
+    process = ProcessInfo()
+    cmdline = ctypes.create_unicode_buffer(command)
+    if not kernel.CreateProcessW(None, cmdline, None, None, False, EXTENDED_STARTUPINFO_PRESENT,
+                                 None, None, ctypes.byref(startup), ctypes.byref(process)):
+        result["error"] = f"CreateProcessW failed: {ctypes.get_last_error()}"
+        return result
+    result["pid"] = process.dwProcessId
+
+    chunks: list[bytes] = []
+
+    def reader() -> None:
+        buffer = ctypes.create_string_buffer(4096)
+        read = wt.DWORD()
+        while kernel.ReadFile(conout_read, buffer, 4096, ctypes.byref(read), None) and read.value:
+            chunks.append(buffer.raw[: read.value])
+
+    thread = threading.Thread(target=reader, daemon=True)
+    thread.start()
+    deadline = time.monotonic() + wait_seconds
+    first_output_ms: int | None = None
+    while time.monotonic() < deadline:
+        if chunks and first_output_ms is None:
+            first_output_ms = round((time.monotonic() - started) * 1000)
+        if kernel.WaitForSingleObject(process.hProcess, 100) == 0:
+            break
+    code = wt.DWORD()
+    kernel.GetExitCodeProcess(process.hProcess, ctypes.byref(code))
+    if code.value == 259:  # STILL_ACTIVE
+        kernel.TerminateProcess(process.hProcess, 1)
+        result["exit"] = "terminated after timeout"
+    else:
+        result["exit"] = code.value
+    # The in-box host flushes trailing output while ClosePseudoConsole drains
+    # the pipe; the reader thread keeps consuming during that call.
+    kernel.CloseHandle(conout_write)
+    close(hpc)
+    thread.join(2.0)
+    output = b"".join(chunks)
+    if chunks and first_output_ms is None:
+        first_output_ms = round((time.monotonic() - started) * 1000)
+    result["first_output_ms"] = first_output_ms
+    result["output_bytes"] = len(output)
+    result["output_head"] = output[:400].decode("utf-8", "replace")
+    kernel.CloseHandle(process.hThread)
+    kernel.CloseHandle(process.hProcess)
+    kernel.DeleteProcThreadAttributeList(attributes)
+    for handle in (conin_write, conin_read, conout_read):
+        kernel.CloseHandle(handle)
+    return result
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--runtime", help="directory holding conpty.dll and OpenConsole.exe")
+    parser.add_argument("--wait", type=float, default=15.0)
+    parser.add_argument("--output")
+    args = parser.parse_args()
+
+    shells = []
+    for name in ("pwsh", "powershell", "cmd"):
+        path = shutil.which(name)
+        if path:
+            shells.append((name, path))
+    libraries: list[str | None] = [None]
+    if args.runtime:
+        libraries.insert(0, os.path.join(os.path.abspath(args.runtime), "conpty.dll"))
+
+    report: dict[str, object] = {
+        "architecture": os.environ.get("PROCESSOR_ARCHITECTURE"),
+        "python": sys.version,
+        "shells": dict(shells),
+        "cases": [],
+    }
+    commands = {
+        "pwsh": '"{path}" -NoLogo -NoProfile -Command "Write-Output CONPTY_SMOKE_OK"',
+        "powershell": '"{path}" -NoLogo -NoProfile -Command "Write-Output CONPTY_SMOKE_OK"',
+        "cmd": '"{path}" /d /c "echo CONPTY_SMOKE_OK"',
+    }
+    for library in libraries:
+        for name, path in shells:
+            for flags in (PSEUDOCONSOLE_WIN32_INPUT_MODE, 0):
+                case = run_case(library, commands[name].format(path=path), flags, args.wait)
+                case["shell"] = name
+                print(json.dumps(case, ensure_ascii=False), flush=True)
+                report["cases"].append(case)
+    if args.output:
+        os.makedirs(os.path.dirname(os.path.abspath(args.output)), exist_ok=True)
+        with open(args.output, "w", encoding="utf-8") as handle:
+            json.dump(report, handle, ensure_ascii=False, indent=2)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/conformance/golden/windows.json

@@ -36,7 +36,7 @@
   "cases.split.status": "passed",
   "cases.ssh_loop.reason": "runtime_api_has_no_ssh_open_method",
   "cases.ssh_loop.status": "skipped",
-  "platform": "windows-x86_64",
+  "platform": "windows",
   "schema_version": 1,
   "summary.failed": 0,
   "summary.passed": 9,

scripts/conformance/harness.py

@@ -936,7 +936,10 @@ def load_rules(golden_dir: Path) -> dict[str, Any]:
 
 def stable_flat(report: dict[str, Any], golden_dir: Path) -> dict[str, Any]:
     rules = load_rules(golden_dir)
-    return filter_flat(flatten(report), list(rules.get("volatile", {})))
+    flat = filter_flat(flatten(report), list(rules.get("volatile", {})))
+    # One golden per family serves every architecture of that platform.
+    flat["platform"] = platform_family(str(report["platform"]))
+    return flat
 
 
 def compare_platform_golden(report: dict[str, Any], golden_dir: Path) -> list[str]:

scripts/conformance/tests/test_harness.py

@@ -154,6 +154,30 @@ class ComparisonTests(unittest.TestCase):
             errors = compare_reports([first, second], golden)
             self.assertTrue(any("tail_contains_marker" in error for error in errors))
 
+    def test_platform_golden_serves_every_architecture_of_a_family(self) -> None:
+        from conformance.harness import compare_platform_golden, stable_flat
+
+        report = {
+            "schema_version": 1,
+            "platform": "windows-x86_64",
+            "cases": {"boot": {"status": "passed", "duration_ms": 3}},
+            "summary": {"total": 1, "passed": 1, "failed": 0, "skipped": 0},
+        }
+        with tempfile.TemporaryDirectory() as directory:
+            golden = Path(directory)
+            (golden / "whitelist.json").write_text(
+                json.dumps({"volatile": {"cases.*.duration_ms": "timing"}, "cross_platform": {}}),
+                encoding="utf-8",
+            )
+            expected = stable_flat(report, golden)
+            self.assertEqual(expected["platform"], "windows")
+            (golden / "windows.json").write_text(json.dumps(expected), encoding="utf-8")
+            arm64 = json.loads(json.dumps(report))
+            arm64["platform"] = "windows-aarch64"
+            self.assertEqual(compare_platform_golden(arm64, golden), [])
+            arm64["cases"]["boot"]["status"] = "failed"
+            self.assertTrue(compare_platform_golden(arm64, golden))
+
 
 class ArchiveSafetyTests(unittest.TestCase):
     def test_new_portable_package_resolves_only_the_product_executable(self) -> None: