diff --git a/scripts/conformance/conpty_smoke.py b/scripts/conformance/conpty_smoke.py index 7a26cf3..d781cf8 100644 --- a/scripts/conformance/conpty_smoke.py +++ b/scripts/conformance/conpty_smoke.py @@ -19,6 +19,7 @@ import time PROC_THREAD_ATTRIBUTE_PSEUDOCONSOLE = 0x00020016 EXTENDED_STARTUPINFO_PRESENT = 0x00080000 +CREATE_UNICODE_ENVIRONMENT = 0x00000400 STARTF_USESTDHANDLES = 0x00000100 PSEUDOCONSOLE_WIN32_INPUT_MODE = 0x4 @@ -124,9 +125,14 @@ def run_case( wait_seconds: float, done_marker: bytes = b"CONPTY_SMOKE_OK", steps: tuple[str, ...] = (), + env: dict[str, str] | None = None, + cwd: str | None = None, ) -> dict[str, object]: """Interactive shells never exit; the marker ends the wait once it renders. + `env` replaces the inherited block (CREATE_UNICODE_ENVIRONMENT) and `cwd` + the inherited directory, so Pebrel's own child environment can be replayed. + `steps` replays Pebrel's post-spawn sequence 30ms after the first output: `resize` calls ResizePseudoConsole, `probe` attaches to the client console and reads its cursor, mirroring the align-sync and align stages. @@ -174,8 +180,14 @@ def run_case( 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)): + creation_flags = EXTENDED_STARTUPINFO_PRESENT + env_block = None + if env is not None: + creation_flags |= CREATE_UNICODE_ENVIRONMENT + block = "".join(f"{name}={value}\0" for name, value in sorted(env.items(), key=lambda item: item[0].upper())) + env_block = ctypes.create_unicode_buffer(block + "\0") + if not kernel.CreateProcessW(None, cmdline, None, None, False, creation_flags, + env_block, cwd, ctypes.byref(startup), ctypes.byref(process)): result["error"] = f"CreateProcessW failed: {ctypes.get_last_error()}" return result result["pid"] = process.dwProcessId diff --git a/scripts/conformance/explicit_command_boot.py b/scripts/conformance/explicit_command_boot.py index 8409167..21b7777 100644 --- a/scripts/conformance/explicit_command_boot.py +++ b/scripts/conformance/explicit_command_boot.py @@ -3,6 +3,11 @@ Separates "the shell hangs at startup" from "the PTY read path is dead": cmd, PowerShell 5.1 and pwsh are launched through the same ConPTY and environment path as the default tab, and every pane is read for 15s. + +The `cmd-env` case captures the exact environment block Pebrel hands its +children; PowerShell 5.1 is then replayed outside the application with that +block and/or that working directory, and the differing variables are bisected +when the block alone reproduces a silent start. """ from __future__ import annotations @@ -16,18 +21,24 @@ import time from pathlib import Path sys.path.insert(0, os.fspath(Path(__file__).resolve().parent.parent)) +sys.path.insert(0, os.fspath(Path(__file__).resolve().parent)) +import conpty_smoke # noqa: E402 from conformance.harness import RuntimeClient # noqa: E402 +MARKER = "PEBREL_SMOKE_OK" +POWERSHELL_51 = ["powershell.exe", "-NoLogo", "-NoProfile", "-NoExit", "-Command", f"Write-Output {MARKER}"] +REPLAY_WAIT = 10.0 + app = Path(sys.argv[1]).resolve() output = Path(sys.argv[2]) -commands = { - "cmd": ["cmd.exe", "/d", "/k", "echo PEBREL_SMOKE_OK"], - "powershell": ["powershell.exe", "-NoLogo", "-NoProfile", "-NoExit", "-Command", "Write-Output PEBREL_SMOKE_OK"], - "powershell-profile": ["powershell.exe", "-NoLogo", "-NoExit", "-Command", "Write-Output PEBREL_SMOKE_OK"], - "pwsh": ["pwsh.exe", "-NoLogo", "-NoProfile", "-NoExit", "-Command", "Write-Output PEBREL_SMOKE_OK"], -} +runtime_library = app.parent / "runtime" / "conpty.dll" report: dict[str, object] = {} -for name, command in commands.items(): +app_env: dict[str, str] | None = None +app_cwd: str | None = None + + +def boot(name: str, command: list[str], env_dump: Path | None = None) -> dict[str, object]: + global app_env with tempfile.TemporaryDirectory(prefix="pebrel-cmd-", ignore_cleanup_errors=True) as root: config = Path(root) / "config" work = Path(root) / "work" @@ -38,6 +49,9 @@ for name, command in commands.items(): "auto_check_updates=0\nai_hooks=0\nwindowing_behavior=use_new\n", encoding="utf-8", ) + if env_dump is not None: + env_dump = Path(root) / env_dump + command = [*command, f"set > {env_dump}"] env = dict(os.environ, PEBREL_CONFIG_DIR=os.fspath(config), NEBULA_CONFIG_DIR=os.fspath(config), NEBULA_BOOT_TRACE="1") env.pop("PEBREL_RUNTIME_ENDPOINT", None) log = open(Path(root) / "app.log", "wb") @@ -76,14 +90,27 @@ for name, command in commands.items(): entry["cwd"] = pane.get("cwd") if text.strip() and first_text_ms is None: first_text_ms = round((time.monotonic() - started) * 1000) - if "PEBREL_SMOKE_OK" in text: + if MARKER in text: + break + if env_dump is not None and env_dump.is_file() and env_dump.stat().st_size > 0: + time.sleep(1.0) break except Exception as error: entry["read_error"] = str(error) time.sleep(0.25) entry["first_text_ms"] = first_text_ms - entry["marker"] = "PEBREL_SMOKE_OK" in text + entry["marker"] = MARKER in text entry["text"] = text.strip()[:600] + if env_dump is not None and env_dump.is_file(): + parsed: dict[str, str] = {} + for line in env_dump.read_text(encoding="mbcs", errors="replace").splitlines(): + name, separator, value = line.partition("=") + if separator and name and not name.startswith("="): + parsed[name] = value + if parsed: + # Windows names are case-insensitive; os.environ upper-cases them. + app_env = {name.upper(): value for name, value in parsed.items()} + entry["app_env_count"] = len(parsed) # The shell keeps the work directory as its cwd; take the whole tree down. subprocess.run(["taskkill", "/T", "/F", "/PID", str(process.pid)], capture_output=True) process.wait(10) @@ -92,7 +119,84 @@ for name, command in commands.items(): line for line in (Path(root) / "app.log").read_text(encoding="utf-8", errors="replace").splitlines() if "prepaint" not in line and "resize-trace" not in line ][:40] - report[name] = entry print(name, json.dumps({k: v for k, v in entry.items() if k != "app_log"}, ensure_ascii=False)[:400], flush=True) + return entry + + +def replay(name: str, env: dict[str, str] | None, cwd: str | None) -> bool: + """Run PowerShell 5.1 through the smoke path; True when the marker rendered.""" + library = os.fspath(runtime_library) if runtime_library.is_file() else None + command = subprocess.list2cmdline(POWERSHELL_51) + result = conpty_smoke.run_case(library, command, 0, REPLAY_WAIT, done_marker=MARKER.encode(), env=env, cwd=cwd) + passed = result.get("marker_ms") is not None + summary = {k: result.get(k) for k in ("error", "first_output_ms", "marker_ms", "output_bytes")} + print("replay", name, "PASS" if passed else "SILENT", json.dumps(summary), flush=True) + report.setdefault("replays", {})[name] = summary | {"passed": passed} # type: ignore[index] + return passed + + +def bisect(base: dict[str, str], target: dict[str, str]) -> None: + """Find which of the variables that differ between base and target break PS 5.1.""" + differing = sorted({*base, *target}, key=str.upper) + differing = [name for name in differing if base.get(name) != target.get(name)] + report["env_diff"] = {name: {"runner": base.get(name), "app": target.get(name)} for name in differing} + print(f"bisect over {len(differing)} differing variables", flush=True) + + def with_changes(names: list[str]) -> dict[str, str]: + env = dict(base) + for name in names: + if name in target: + env[name] = target[name] + else: + env.pop(name, None) + return env + + if replay("runner-env-plus-all-diffs", with_changes(differing), None): + report["bisect"] = "applying every differing variable to the runner environment still passes" + return + suspects = list(differing) + rounds = 0 + while len(suspects) > 1 and rounds < 12: + rounds += 1 + half = suspects[: len(suspects) // 2] + if not replay(f"round{rounds}-first-half({len(half)})", with_changes(half), None): + suspects = half + continue + rest = suspects[len(suspects) // 2 :] + if not replay(f"round{rounds}-second-half({len(rest)})", with_changes(rest), None): + suspects = rest + continue + report["bisect"] = f"no single half reproduces; remaining suspects: {suspects}" + return + report["bisect"] = {"culprits": suspects, "values": {name: target.get(name) for name in suspects}} + print("bisect result:", report["bisect"], flush=True) + + +def redacted(env: dict[str, str] | None) -> dict[str, str] | None: + if env is None: + return None + secret = ("TOKEN", "SECRET", "PASSWORD", "_KEY") + return {name: ("" if any(part in name.upper() for part in secret) else value) for name, value in env.items()} + + +# Same shape as the panes' working directory (under TEMP), but it outlives boot(). +app_cwd = os.fspath(Path(tempfile.mkdtemp(prefix="pebrel-cmd-")) / "work") +os.mkdir(app_cwd) +report["cmd-env"] = boot("cmd-env", ["cmd.exe", "/d", "/k"], env_dump=Path("app-env.txt")) +report["powershell"] = boot("powershell", POWERSHELL_51) +report["runner_env"] = redacted(dict(os.environ)) +report["app_env"] = redacted(app_env) +if app_env is not None: + control = replay("inherited-env-inherited-cwd", None, None) + env_only = replay("app-env-inherited-cwd", app_env, None) + cwd_only = replay("inherited-env-app-cwd", None, app_cwd) + replay("app-env-app-cwd", app_env, app_cwd) + if control and not env_only: + bisect(dict(os.environ), app_env) + if isinstance(report.get("env_diff"), dict): + report["env_diff"] = { + name: {side: redacted({name: value})[name] if value is not None else None for side, value in sides.items()} # type: ignore[index] + for name, sides in report["env_diff"].items() # type: ignore[union-attr] + } output.parent.mkdir(parents=True, exist_ok=True) output.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")