From 6a8a6f164f7127006ac5a2d21cd91e8a0e601c7a Mon Sep 17 00:00:00 2001 From: Joey Wang Date: Thu, 8 Oct 2026 16:18:12 -0400 Subject: [PATCH] Fix Codex lifecycle notifications from detached Windows hooks Recover the live Codex ancestor console while keeping hook control streams captured. Add isolated ConPTY delivery coverage and publish the candidate as plugin 0.4.3. Co-Authored-By: Warp --- .agents/plugins/marketplace.json | 2 +- .github/workflows/test.yml | 5 + README.md | 19 +- plugins/warp/.codex-plugin/plugin.json | 2 +- plugins/warp/scripts/common.ps1 | 90 +---- plugins/warp/scripts/console-writer.cs | 262 +++++++++++++ plugins/warp/scripts/on-session-start.ps1 | 2 +- plugins/warp/scripts/on-session-start.sh | 2 +- tests/console-parent.cs | 109 ++++++ tests/notification_conpty_test.py | 424 ++++++++++++++++++++++ tests/test-hooks.ps1 | 13 +- 11 files changed, 836 insertions(+), 94 deletions(-) create mode 100644 plugins/warp/scripts/console-writer.cs create mode 100644 tests/console-parent.cs create mode 100644 tests/notification_conpty_test.py diff --git a/.agents/plugins/marketplace.json b/.agents/plugins/marketplace.json index 95bfcd9..7fc13a6 100644 --- a/.agents/plugins/marketplace.json +++ b/.agents/plugins/marketplace.json @@ -14,7 +14,7 @@ "installation": "AVAILABLE", "authentication": "ON_INSTALL" }, - "version": "0.4.2", + "version": "0.4.3", "category": "productivity", "tags": ["notifications", "terminal", "warp"] }, diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 29fbb0e..0cf524b 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -23,6 +23,11 @@ jobs: runs-on: windows-latest steps: - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.12" - name: Run PowerShell tests shell: powershell run: .\tests\test-hooks.ps1 + - name: Verify ConPTY notification delivery + run: python tests/notification_conpty_test.py -v diff --git a/README.md b/README.md index e1f5407..3a9f1ab 100644 --- a/README.md +++ b/README.md @@ -13,6 +13,7 @@ plugins/orchestration/hooks/hooks.json plugins/orchestration/scripts/ Oz parent-message listener, drain, and lifecycle scripts for POSIX shells and Windows PowerShell plugins/orchestration/skills/ Oz orchestration skills tests/test-hooks.sh Shell tests +tests/notification_conpty_test.py Isolated Windows ConPTY delivery tests ``` ## Plugins - `warp`: `SessionStart`, `Stop`, `PermissionRequest`, `UserPromptSubmit`, `PostToolUse` notifications for Warp. @@ -40,8 +41,20 @@ The POSIX notifier uses `/dev/tty` when a controlling terminal is available. Codex 0.155.0 and later detach hooks from that terminal, so the notifier falls back to an ancestor's terminal device. Linux uses `/proc` and `readlink`; macOS uses `ps`. Discovery is bounded, only terminal devices are used, and failure -produces a diagnostic without failing the hook. Notification payloads and the -Windows transport are unchanged. +produces a diagnostic without failing the hook. Notification payloads are unchanged. +On Windows, the notifier locates a live native Codex ancestor and checks whether +it shares that process's console. Detached hooks temporarily attach to the Codex +console while preserving their captured standard streams, then restore their +previous attachment when another live owner remains. Notifications never go to +hook stdout; an unavailable console produces a payload-free stderr diagnostic. +The Windows test suite checks actual OSC delivery on an isolated ConPTY, including +detached `cmd /C` launches, concurrent hooks, Unicode, and captured control output: +```sh +python tests/notification_conpty_test.py -v +``` +Set `WARP_CONPTY_DLL` to Warp's bundled `conpty.dll` to test its duplex transport +instead of Windows' system ConPTY. Keep the DLL's companion assets adjacent and +available on `PATH`. ## One-off notifier testing in an Oz Docker task Building the Codex CLI sidecar does not include this checkout. The CLI image contains Codex itself; Warp normally installs this plugin at runtime. @@ -60,7 +73,7 @@ This modifies only that container's installed copy. It does not publish the plugin or make newly created tasks use local changes. ## Versioning `plugins/warp/scripts/on-session-start.sh` emits `PLUGIN_VERSION`. -Current plugin version: `0.4.2`. +Current plugin version: `0.4.3`. Keep it in sync with Warp's Codex plugin manager minimum version. ## Skills diff --git a/plugins/warp/.codex-plugin/plugin.json b/plugins/warp/.codex-plugin/plugin.json index 80b0245..aa944ef 100644 --- a/plugins/warp/.codex-plugin/plugin.json +++ b/plugins/warp/.codex-plugin/plugin.json @@ -1,7 +1,7 @@ { "name": "warp", "description": "Warp terminal integration for Codex - native notifications, and more to come", - "version": "0.4.2", + "version": "0.4.3", "author": { "name": "Warp", "url": "https://warp.dev" diff --git a/plugins/warp/scripts/common.ps1 b/plugins/warp/scripts/common.ps1 index efcf857..eac3810 100644 --- a/plugins/warp/scripts/common.ps1 +++ b/plugins/warp/scripts/common.ps1 @@ -122,86 +122,7 @@ function Initialize-WarpConsoleWriter { return } - Add-Type -TypeDefinition @" -using System; -using System.Runtime.InteropServices; - -public static class WarpConsoleWriter -{ - private const uint GenericWrite = 0x40000000; - private const uint FileShareRead = 0x00000001; - private const uint FileShareWrite = 0x00000002; - private const uint OpenExisting = 3; - - [DllImport("kernel32.dll", CharSet = CharSet.Unicode, SetLastError = true)] - private static extern IntPtr CreateFileW( - string fileName, - uint desiredAccess, - uint shareMode, - IntPtr securityAttributes, - uint creationDisposition, - uint flagsAndAttributes, - IntPtr templateFile); - - [DllImport("kernel32.dll", CharSet = CharSet.Unicode, SetLastError = true)] - private static extern bool WriteConsoleW( - IntPtr consoleOutput, - string buffer, - uint charsToWrite, - out uint charsWritten, - IntPtr reserved); - - [DllImport("kernel32.dll")] - private static extern bool CloseHandle(IntPtr handle); - - public static bool Write(string message) - { - IntPtr handle = CreateFileW( - "CONOUT$", - GenericWrite, - FileShareRead | FileShareWrite, - IntPtr.Zero, - OpenExisting, - 0, - IntPtr.Zero); - - if (handle == new IntPtr(-1)) - { - return false; - } - - try - { - int offset = 0; - while (offset < message.Length) - { - int chunkLength = Math.Min(16384, message.Length - offset); - if (offset + chunkLength < message.Length - && char.IsHighSurrogate(message[offset + chunkLength - 1])) - { - chunkLength--; - } - - string chunk = message.Substring(offset, chunkLength); - uint written; - if (!WriteConsoleW(handle, chunk, (uint)chunk.Length, out written, IntPtr.Zero) - || written == 0) - { - return false; - } - - offset += (int)written; - } - - return true; - } - finally - { - CloseHandle(handle); - } - } -} -"@ + Add-Type -Path (Join-Path $PSScriptRoot "console-writer.cs") } function Send-WarpNotification { @@ -216,10 +137,11 @@ function Send-WarpNotification { try { Initialize-WarpConsoleWriter - [void][WarpConsoleWriter]::Write($message) + if (-not [WarpConsoleWriter]::Write($message)) { + [Console]::Error.WriteLine("[codex-warp] notification delivery failed: $([WarpConsoleWriter]::LastError)") + } } catch { - # Hook stdout is reserved for Codex hook control JSON. If there is no - # attached console device, drop the notification rather than emitting - # OSC text to captured stdout and making the hook fail. + # Hook stdout is reserved for Codex hook control JSON, not notification bytes. + [Console]::Error.WriteLine("[codex-warp] notification delivery failed: console interop unavailable") } } diff --git a/plugins/warp/scripts/console-writer.cs b/plugins/warp/scripts/console-writer.cs new file mode 100644 index 0000000..03d83d2 --- /dev/null +++ b/plugins/warp/scripts/console-writer.cs @@ -0,0 +1,262 @@ +using System; +using System.Collections.Generic; +using System.Runtime.InteropServices; + +public static class WarpConsoleWriter +{ + private const uint SnapshotProcesses = 2; + private const uint QueryLimitedInformation = 0x1000; + private const uint Synchronize = 0x100000; + private const uint WaitTimeout = 258; + private static readonly IntPtr InvalidHandle = new IntPtr(-1); + + public static string LastError { get; private set; } + + [StructLayout(LayoutKind.Sequential, CharSet = CharSet.Unicode)] + private struct ProcessEntry + { + public uint Size; + public uint Usage; + public uint ProcessId; + public UIntPtr DefaultHeap; + public uint ModuleId; + public uint Threads; + public uint ParentId; + public int Priority; + public uint Flags; + [MarshalAs(UnmanagedType.ByValTStr, SizeConst = 260)] + public string Executable; + } + + [StructLayout(LayoutKind.Sequential)] + private struct FileTime + { + public uint Low; + public uint High; + public ulong Value { get { return ((ulong)High << 32) | Low; } } + } + + [DllImport("kernel32.dll", SetLastError = true)] + private static extern IntPtr CreateToolhelp32Snapshot(uint flags, uint processId); + [DllImport("kernel32.dll", CharSet = CharSet.Unicode, SetLastError = true)] + private static extern bool Process32FirstW(IntPtr snapshot, ref ProcessEntry entry); + [DllImport("kernel32.dll", CharSet = CharSet.Unicode, SetLastError = true)] + private static extern bool Process32NextW(IntPtr snapshot, ref ProcessEntry entry); + [DllImport("kernel32.dll", SetLastError = true)] + private static extern IntPtr OpenProcess(uint access, bool inherit, uint processId); + [DllImport("kernel32.dll")] + private static extern uint GetCurrentProcessId(); + [DllImport("kernel32.dll")] + private static extern IntPtr GetCurrentProcess(); + [DllImport("kernel32.dll", SetLastError = true)] + private static extern bool GetProcessTimes(IntPtr process, out FileTime created, + out FileTime exited, out FileTime kernel, out FileTime user); + [DllImport("kernel32.dll")] + private static extern uint WaitForSingleObject(IntPtr handle, uint milliseconds); + [DllImport("kernel32.dll", SetLastError = true)] + private static extern uint GetConsoleProcessList([Out] uint[] processes, uint count); + [DllImport("kernel32.dll", SetLastError = true)] + private static extern bool AttachConsole(uint processId); + [DllImport("kernel32.dll", SetLastError = true)] + private static extern bool FreeConsole(); + [DllImport("kernel32.dll")] + private static extern IntPtr GetStdHandle(int which); + [DllImport("kernel32.dll", SetLastError = true)] + private static extern bool SetStdHandle(int which, IntPtr handle); + [DllImport("kernel32.dll", CharSet = CharSet.Unicode, SetLastError = true)] + private static extern IntPtr CreateFileW(string name, uint access, uint share, + IntPtr security, uint disposition, uint flags, IntPtr template); + [DllImport("kernel32.dll", CharSet = CharSet.Unicode, SetLastError = true)] + private static extern bool WriteConsoleW(IntPtr output, string text, uint length, + out uint written, IntPtr reserved); + [DllImport("kernel32.dll")] + private static extern bool CloseHandle(IntPtr handle); + + private static bool CreationTime(IntPtr process, out ulong time) + { + FileTime created, exited, kernel, user; + bool ok = GetProcessTimes(process, out created, out exited, out kernel, out user); + time = created.Value; + return ok; + } + + private static uint[] ConsoleProcesses() + { + uint[] ids = new uint[64]; + uint count = GetConsoleProcessList(ids, (uint)ids.Length); + if (count > ids.Length && count <= 1024) + { + ids = new uint[count]; + count = GetConsoleProcessList(ids, (uint)ids.Length); + } + if (count > ids.Length) + return new uint[0]; + Array.Resize(ref ids, (int)count); + return ids; + } + + private static bool IsCodex(string executable) + { + return String.Equals(executable, "codex.exe", StringComparison.OrdinalIgnoreCase) + || (executable.StartsWith("codex-", StringComparison.OrdinalIgnoreCase) + && executable.EndsWith(".exe", StringComparison.OrdinalIgnoreCase)); + } + + private static IntPtr FindCodex(out uint targetId) + { + targetId = 0; + ulong childCreated; + if (!CreationTime(GetCurrentProcess(), out childCreated)) + return IntPtr.Zero; + IntPtr snapshot = CreateToolhelp32Snapshot(SnapshotProcesses, 0); + if (snapshot == InvalidHandle) + return IntPtr.Zero; + var entries = new Dictionary(); + try + { + var entry = new ProcessEntry(); + entry.Size = (uint)Marshal.SizeOf(typeof(ProcessEntry)); + if (!Process32FirstW(snapshot, ref entry)) + return IntPtr.Zero; + do { entries[entry.ProcessId] = entry; } + while (Process32NextW(snapshot, ref entry)); + } + finally { CloseHandle(snapshot); } + + uint current = GetCurrentProcessId(); + var visited = new HashSet(); + for (int depth = 0; depth < 32; depth++) + { + ProcessEntry child, parent; + if (!visited.Add(current) || !entries.TryGetValue(current, out child) + || !entries.TryGetValue(child.ParentId, out parent)) + break; + IntPtr process = OpenProcess(QueryLimitedInformation | Synchronize, false, parent.ProcessId); + if (process == IntPtr.Zero) + break; + ulong parentCreated; + if (WaitForSingleObject(process, 0) != WaitTimeout + || !CreationTime(process, out parentCreated) || parentCreated > childCreated) + { + CloseHandle(process); + break; + } + if (IsCodex(parent.Executable)) + { + // Keep the process object alive while AttachConsole consumes its numeric PID. + targetId = parent.ProcessId; + return process; + } + childCreated = parentCreated; + current = parent.ProcessId; + CloseHandle(process); + } + return IntPtr.Zero; + } + + public static bool Write(string message) + { + LastError = "no live Codex ancestor"; + uint targetId; + IntPtr target = FindCodex(out targetId); + if (target == IntPtr.Zero) + return false; + + IntPtr input = GetStdHandle(-10); + IntPtr output = GetStdHandle(-11); + IntPtr error = GetStdHandle(-12); + IntPtr restoreProcess = IntPtr.Zero; + uint restoreId = 0; + bool changedAttachment = false; + bool attachedTarget = false; + try + { + uint[] original = ConsoleProcesses(); + if (Array.IndexOf(original, targetId) < 0) + { + foreach (uint id in original) + { + if (id == GetCurrentProcessId()) + continue; + restoreProcess = OpenProcess(QueryLimitedInformation | Synchronize, false, id); + if (restoreProcess != IntPtr.Zero + && WaitForSingleObject(restoreProcess, 0) == WaitTimeout) + { + restoreId = id; + break; + } + if (restoreProcess != IntPtr.Zero) + CloseHandle(restoreProcess); + restoreProcess = IntPtr.Zero; + } + if (original.Length > 0 && !FreeConsole()) + { + LastError = "could not detach private hook console"; + return false; + } + changedAttachment = true; + if (WaitForSingleObject(target, 0) != WaitTimeout || !AttachConsole(targetId)) + { + LastError = "could not attach Codex console"; + return false; + } + attachedTarget = true; + SetStdHandle(-10, input); + SetStdHandle(-11, output); + SetStdHandle(-12, error); + if (Array.IndexOf(ConsoleProcesses(), targetId) < 0) + { + LastError = "attached console does not contain Codex"; + return false; + } + } + + IntPtr console = CreateFileW("CONOUT$", 0x40000000, 3, IntPtr.Zero, 3, 0, IntPtr.Zero); + if (console == InvalidHandle) + { + LastError = "could not open Codex console output"; + return false; + } + try + { + int offset = 0; + while (offset < message.Length) + { + int length = Math.Min(16384, message.Length - offset); + if (offset + length < message.Length + && char.IsHighSurrogate(message[offset + length - 1])) + length--; + uint written; + if (!WriteConsoleW(console, message.Substring(offset, length), + (uint)length, out written, IntPtr.Zero) || written == 0) + { + LastError = "could not write Codex console notification"; + return false; + } + offset += (int)written; + } + LastError = null; + return true; + } + finally { CloseHandle(console); } + } + finally + { + if (changedAttachment) + { + if (attachedTarget) + FreeConsole(); + if (restoreProcess != IntPtr.Zero + && WaitForSingleObject(restoreProcess, 0) == WaitTimeout) + AttachConsole(restoreId); + // AttachConsole can replace standard handles; hook control JSON must stay piped. + SetStdHandle(-10, input); + SetStdHandle(-11, output); + SetStdHandle(-12, error); + } + if (restoreProcess != IntPtr.Zero) + CloseHandle(restoreProcess); + CloseHandle(target); + } + } +} diff --git a/plugins/warp/scripts/on-session-start.ps1 b/plugins/warp/scripts/on-session-start.ps1 index 258cf0d..a83d3de 100644 --- a/plugins/warp/scripts/on-session-start.ps1 +++ b/plugins/warp/scripts/on-session-start.ps1 @@ -7,7 +7,7 @@ if (-not (Test-ShouldUseStructured)) { exit 0 } -$pluginVersion = "0.4.2" +$pluginVersion = "0.4.3" $inputJson = Read-HookInput $body = New-WarpPayload $inputJson "session_start" @{ plugin_version = $pluginVersion diff --git a/plugins/warp/scripts/on-session-start.sh b/plugins/warp/scripts/on-session-start.sh index f34329c..0e85767 100755 --- a/plugins/warp/scripts/on-session-start.sh +++ b/plugins/warp/scripts/on-session-start.sh @@ -5,7 +5,7 @@ set -euo pipefail # Bump on every release; keep in sync with Warp's Codex plugin manager. -PLUGIN_VERSION="0.4.2" +PLUGIN_VERSION="0.4.3" SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" source "$SCRIPT_DIR/should-use-structured.sh" diff --git a/tests/console-parent.cs b/tests/console-parent.cs new file mode 100644 index 0000000..710e811 --- /dev/null +++ b/tests/console-parent.cs @@ -0,0 +1,109 @@ +using System; +using System.Collections.Generic; +using System.Diagnostics; +using System.IO; +using System.Runtime.InteropServices; +using System.Text; +using System.Threading.Tasks; + +public static class ConsoleParent +{ + [DllImport("kernel32.dll")] + private static extern IntPtr GetStdHandle(int which); + [DllImport("kernel32.dll")] + private static extern bool GetConsoleMode(IntPtr handle, out uint mode); + [DllImport("kernel32.dll")] + private static extern bool SetConsoleMode(IntPtr handle, uint mode); + + public static int Main(string[] args) + { + if (args[0] == "--orphan") + return StartOrphanHook(args[1], args[2]); + string hook = args[0]; + string payload = File.ReadAllText(args[1], Encoding.UTF8); + string results = args[2]; + bool detached = args[3] == "detached"; + bool wrapper = args[4] == "cmd"; + int count = Int32.Parse(args[5]); + bool exitedParent = args.Length > 6 && args[6] == "dead"; + var children = new List(); + var outputs = new List>(); + var errors = new List>(); + uint mode; + IntPtr console = GetStdHandle(-11); + if (GetConsoleMode(console, out mode)) + SetConsoleMode(console, mode | 4); + try + { + for (int i = 0; i < count; i++) + { + string powershell = "powershell.exe -NoProfile -ExecutionPolicy Bypass -File \"" + hook + "\""; + var start = new ProcessStartInfo { + FileName = exitedParent ? System.Reflection.Assembly.GetExecutingAssembly().Location + : wrapper ? "cmd.exe" : "powershell.exe", + Arguments = exitedParent ? "--orphan \"" + hook + "\" \"" + args[1] + "\"" + : wrapper ? "/C \"" + powershell + "\"" + : "-NoProfile -ExecutionPolicy Bypass -File \"" + hook + "\"", + UseShellExecute = false, + CreateNoWindow = detached, + RedirectStandardInput = true, + RedirectStandardOutput = true, + RedirectStandardError = true, + StandardOutputEncoding = Encoding.UTF8, + StandardErrorEncoding = Encoding.UTF8 + }; + start.EnvironmentVariables["WARP_CLI_AGENT_PROTOCOL_VERSION"] = "1"; + start.EnvironmentVariables["WARP_CLIENT_VERSION"] = "test"; + Process child = Process.Start(start); + children.Add(child); + outputs.Add(child.StandardOutput.ReadToEndAsync()); + errors.Add(child.StandardError.ReadToEndAsync()); + byte[] input = Encoding.UTF8.GetBytes(payload); + if (!exitedParent) + child.StandardInput.BaseStream.Write(input, 0, input.Length); + child.StandardInput.Close(); + } + for (int i = 0; i < children.Count; i++) + { + Process child = children[i]; + if (!child.WaitForExit(30000)) + { + child.Kill(); + return 2; + } + string result = "{\"exit_code\":" + child.ExitCode + + ",\"stdout_base64\":\"" + Convert.ToBase64String(Encoding.UTF8.GetBytes(outputs[i].Result)) + + "\",\"stderr_base64\":\"" + Convert.ToBase64String(Encoding.UTF8.GetBytes(errors[i].Result)) + "\"}"; + File.WriteAllText(Path.Combine(results, "result-" + i + ".json"), + result, new UTF8Encoding(false)); + } + return 0; + } + finally + { + foreach (Process child in children) + { + if (!child.HasExited) + child.Kill(); + child.Dispose(); + } + } + } + + private static int StartOrphanHook(string hook, string inputPath) + { + var start = new ProcessStartInfo { + FileName = "powershell.exe", + Arguments = "-NoProfile -ExecutionPolicy Bypass -File \"" + hook + "\"", + UseShellExecute = false, + CreateNoWindow = true, + RedirectStandardInput = true + }; + start.EnvironmentVariables["CONPTY_TEST_PARENT_PID"] = Process.GetCurrentProcess().Id.ToString(); + Process child = Process.Start(start); + byte[] input = Encoding.UTF8.GetBytes(File.ReadAllText(inputPath, Encoding.UTF8)); + child.StandardInput.BaseStream.Write(input, 0, input.Length); + // The coordinator retains the capture pipes; this parent exits before the hook emits. + return 0; + } +} diff --git a/tests/notification_conpty_test.py b/tests/notification_conpty_test.py new file mode 100644 index 0000000..e874984 --- /dev/null +++ b/tests/notification_conpty_test.py @@ -0,0 +1,424 @@ +import base64 +import ctypes +from ctypes import wintypes as w +import json +import os +from pathlib import Path +import re +import shutil +import subprocess +import sys +import tempfile +import threading +import time +import unittest + + +ROOT = Path(__file__).resolve().parents[1] +IS_WINDOWS = sys.platform == "win32" + +if IS_WINDOWS: + kernel = ctypes.WinDLL("kernel32", use_last_error=True) + + class Coord(ctypes.Structure): + _fields_ = [("x", ctypes.c_short), ("y", ctypes.c_short)] + + class Overlapped(ctypes.Structure): + _fields_ = [("internal", ctypes.c_size_t), ("internal_high", ctypes.c_size_t), + ("offset", w.DWORD), ("offset_high", w.DWORD), ("event", w.HANDLE)] + + class StartupInfo(ctypes.Structure): + _fields_ = [ + ("cb", w.DWORD), ("reserved", w.LPWSTR), ("desktop", w.LPWSTR), + ("title", w.LPWSTR), ("x", w.DWORD), ("y", w.DWORD), + ("x_size", w.DWORD), ("y_size", w.DWORD), + ("x_chars", w.DWORD), ("y_chars", w.DWORD), + ("fill", w.DWORD), ("flags", w.DWORD), ("show", w.WORD), + ("reserved_size", w.WORD), ("reserved_ptr", ctypes.c_void_p), + ("stdin", w.HANDLE), ("stdout", w.HANDLE), ("stderr", w.HANDLE), + ] + + class StartupInfoEx(ctypes.Structure): + _fields_ = [("startup", StartupInfo), ("attributes", ctypes.c_void_p)] + + class ProcessInfo(ctypes.Structure): + _fields_ = [("process", w.HANDLE), ("thread", w.HANDLE), + ("pid", w.DWORD), ("tid", w.DWORD)] + + class BasicLimits(ctypes.Structure): + _fields_ = [ + ("process_time", ctypes.c_int64), ("job_time", ctypes.c_int64), + ("flags", w.DWORD), ("min_working_set", ctypes.c_size_t), + ("max_working_set", ctypes.c_size_t), ("active_processes", w.DWORD), + ("affinity", ctypes.c_size_t), ("priority", w.DWORD), ("scheduling", w.DWORD), + ] + + class ExtendedLimits(ctypes.Structure): + _fields_ = [ + ("basic", BasicLimits), ("io_counters", ctypes.c_uint64 * 6), + ("process_memory", ctypes.c_size_t), ("job_memory", ctypes.c_size_t), + ("peak_process_memory", ctypes.c_size_t), ("peak_job_memory", ctypes.c_size_t), + ] + + def api(name, args, result=w.BOOL): + function = getattr(kernel, name) + function.argtypes = args + function.restype = result + return function + + close = api("CloseHandle", [w.HANDLE]) + create_pipe = api("CreatePipe", [ctypes.POINTER(w.HANDLE), ctypes.POINTER(w.HANDLE), + ctypes.c_void_p, w.DWORD]) + read_file = api("ReadFile", [w.HANDLE, ctypes.c_void_p, w.DWORD, + ctypes.POINTER(w.DWORD), ctypes.c_void_p]) + write_file = api("WriteFile", [w.HANDLE, ctypes.c_void_p, w.DWORD, + ctypes.POINTER(w.DWORD), ctypes.c_void_p]) + create_event = api("CreateEventW", [ctypes.c_void_p, w.BOOL, w.BOOL, w.LPCWSTR], w.HANDLE) + overlapped_result = api("GetOverlappedResult", + [w.HANDLE, ctypes.c_void_p, ctypes.POINTER(w.DWORD), w.BOOL]) + cancel_io = api("CancelIoEx", [w.HANDLE, ctypes.c_void_p]) + wait = api("WaitForSingleObject", [w.HANDLE, w.DWORD], w.DWORD) + terminate = api("TerminateProcess", [w.HANDLE, w.UINT]) + create_job = api("CreateJobObjectW", [ctypes.c_void_p, w.LPCWSTR], w.HANDLE) + set_job_info = api("SetInformationJobObject", + [w.HANDLE, w.DWORD, ctypes.c_void_p, w.DWORD]) + assign_job = api("AssignProcessToJobObject", [w.HANDLE, w.HANDLE]) + resume = api("ResumeThread", [w.HANDLE], w.DWORD) + get_exit = api("GetExitCodeProcess", [w.HANDLE, ctypes.POINTER(w.DWORD)]) + init_attributes = api("InitializeProcThreadAttributeList", + [ctypes.c_void_p, w.DWORD, w.DWORD, ctypes.POINTER(ctypes.c_size_t)]) + set_attribute = api("UpdateProcThreadAttribute", + [ctypes.c_void_p, w.DWORD, ctypes.c_size_t, ctypes.c_void_p, + ctypes.c_size_t, ctypes.c_void_p, ctypes.c_void_p]) + delete_attributes = api("DeleteProcThreadAttributeList", [ctypes.c_void_p], None) + create_process = api("CreateProcessW", + [w.LPCWSTR, w.LPWSTR, ctypes.c_void_p, ctypes.c_void_p, w.BOOL, + w.DWORD, ctypes.c_void_p, w.LPCWSTR, ctypes.POINTER(StartupInfo), + ctypes.POINTER(ProcessInfo)]) + + def checked(ok): + if not ok: + raise ctypes.WinError(ctypes.get_last_error()) + + class ConPty: + def __init__(self, command): + self.handles = [] + self.process = ProcessInfo() + self.job = None + self.console = w.HANDLE() + self.output = bytearray() + self.condition = threading.Condition() + self.eof = False + self.reader = None + dll_path = os.environ.get("WARP_CONPTY_DLL") + self.duplex = bool(dll_path) + dll = ctypes.WinDLL(dll_path, use_last_error=True) if dll_path else kernel + self.dll = dll + create = dll.CreatePseudoConsole + create.argtypes = [Coord, w.HANDLE, w.HANDLE, w.DWORD, ctypes.POINTER(w.HANDLE)] + create.restype = ctypes.c_long + self.close_console = dll.ClosePseudoConsole + self.close_console.argtypes = [w.HANDLE] + self.close_console.restype = None + + try: + if dll_path: + # Warp's bundled ConPTY expects one duplex endpoint for both directions. + name = r"\\.\pipe\codex-warp-test-" + os.urandom(12).hex() + named = api("CreateNamedPipeW", + [w.LPCWSTR, w.DWORD, w.DWORD, w.DWORD, w.DWORD, + w.DWORD, w.DWORD, ctypes.c_void_p], w.HANDLE) + file = api("CreateFileW", [w.LPCWSTR, w.DWORD, w.DWORD, ctypes.c_void_p, + w.DWORD, w.DWORD, w.HANDLE], w.HANDLE) + connect = api("ConnectNamedPipe", [w.HANDLE, ctypes.c_void_p]) + server = named(name, 3 | 0x40000000, 0, 1, 65536, 65536, 0, None) + checked(server != ctypes.c_void_p(-1).value) + self.handles.append(server) + client = file(name, 0xC0000000, 0, None, 3, 0x40000000, None) + checked(client != ctypes.c_void_p(-1).value) + self.handles.append(client) + connection = Overlapped() + connection.event = create_event(None, True, False, None) + checked(connection.event) + try: + if not connect(server, ctypes.byref(connection)): + checked(ctypes.get_last_error() == 535) + finally: + close(connection.event) + self.input_write = self.output_read = server + input_read = output_write = client + else: + input_read, self.input_write = w.HANDLE(), w.HANDLE() + self.output_read, output_write = w.HANDLE(), w.HANDLE() + checked(create_pipe(ctypes.byref(input_read), ctypes.byref(self.input_write), None, 0)) + self.handles.extend([input_read, self.input_write]) + checked(create_pipe(ctypes.byref(self.output_read), ctypes.byref(output_write), None, 0)) + self.handles.extend([self.output_read, output_write]) + hr = create(Coord(120, 40), input_read, output_write, 0, ctypes.byref(self.console)) + if hr < 0: + raise OSError("CreatePseudoConsole failed: HRESULT %08x" % (hr & 0xFFFFFFFF)) + + for endpoint in {input_read.value if hasattr(input_read, "value") else input_read, + output_write.value if hasattr(output_write, "value") else output_write}: + close(endpoint) + self.handles = [handle for handle in self.handles + if (handle.value if hasattr(handle, "value") else handle) != endpoint] + + self.reader = threading.Thread(target=self._read, daemon=True) + self.reader.start() + self._write(b"\x1b[?1;2c") + size = ctypes.c_size_t() + init_attributes(None, 1, 0, ctypes.byref(size)) + attributes = ctypes.create_string_buffer(size.value) + checked(init_attributes(attributes, 1, 0, ctypes.byref(size))) + try: + checked(set_attribute(attributes, 0, 0x00020016, self.console, + ctypes.sizeof(w.HANDLE), None, None)) + startup = StartupInfoEx() + startup.startup.cb = ctypes.sizeof(startup) + startup.startup.flags = 0x100 + startup.attributes = ctypes.cast(attributes, ctypes.c_void_p) + cmdline = ctypes.create_unicode_buffer(subprocess.list2cmdline(command)) + checked(create_process(None, cmdline, None, None, False, 0x80004, + None, str(ROOT), + ctypes.cast(ctypes.byref(startup), ctypes.POINTER(StartupInfo)), + ctypes.byref(self.process))) + self.job = create_job(None, None) + checked(self.job) + limits = ExtendedLimits() + limits.basic.flags = 0x2000 + checked(set_job_info(self.job, 9, ctypes.byref(limits), ctypes.sizeof(limits))) + checked(assign_job(self.job, self.process.process)) + checked(resume(self.process.thread) != 0xFFFFFFFF) + finally: + delete_attributes(attributes) + if dll_path: + show = dll.ConptyShowHidePseudoConsole + show.argtypes = [w.HANDLE, w.BOOL] + show.restype = ctypes.c_long + show(self.console, True) + release = dll.ConptyReleasePseudoConsole + release.argtypes = [w.HANDLE] + release.restype = ctypes.c_long + release(self.console) + except BaseException: + self.close() + raise + + def _io(self, function, handle, buffer, length, transferred): + if not self.duplex: + return function(handle, buffer, length, ctypes.byref(transferred), None) + operation = Overlapped() + operation.event = create_event(None, True, False, None) + checked(operation.event) + try: + if function(handle, buffer, length, ctypes.byref(transferred), + ctypes.byref(operation)): + return True + if ctypes.get_last_error() != 997: + return False + if wait(operation.event, 45000) != 0: + cancel_io(handle, ctypes.byref(operation)) + overlapped_result(handle, ctypes.byref(operation), + ctypes.byref(transferred), True) + return False + return overlapped_result(handle, ctypes.byref(operation), + ctypes.byref(transferred), True) + finally: + close(operation.event) + + def _write(self, data): + written = w.DWORD() + checked(self._io(write_file, self.input_write, data, len(data), written)) + + def _read(self): + buffer = ctypes.create_string_buffer(4096) + read = w.DWORD() + while self._io(read_file, self.output_read, buffer, len(buffer), read): + if not read.value: + break + chunk = buffer.raw[:read.value] + with self.condition: + self.output.extend(chunk) + self.condition.notify_all() + if b"\x1b[c" in chunk: + self._write(b"\x1b[?1;2c") + if b"\x1b[6n" in chunk: + self._write(b"\x1b[1;1R") + with self.condition: + self.eof = True + self.condition.notify_all() + + def complete(self, timeout=40): + if wait(self.process.process, int(timeout * 1000)) != 0: + raise TimeoutError("console fixture did not finish") + code = w.DWORD() + checked(get_exit(self.process.process, ctypes.byref(code))) + if code.value: + raise AssertionError("console fixture exited %d" % code.value) + + def notifications(self, count, timeout=5): + pattern = rb"\x1b\]777;notify;warp://cli-agent;([^\x07]*)\x07" + deadline = time.monotonic() + timeout + with self.condition: + while True: + matches = re.findall(pattern, bytes(self.output)) + if len(matches) >= count or self.eof: + return [json.loads(match) for match in matches] + remaining = deadline - time.monotonic() + if remaining <= 0: + return [json.loads(match) for match in matches] + self.condition.wait(remaining) + + def close(self): + if self.job: + close(self.job) + self.job = None + if self.process.process: + if wait(self.process.process, 0) != 0: + terminate(self.process.process, 1) + wait(self.process.process, 5000) + close(self.process.thread) + close(self.process.process) + self.process = ProcessInfo() + if self.console: + self.close_console(self.console) + self.console = w.HANDLE() + if self.duplex: + for handle in self.handles: + cancel_io(handle, None) + if self.reader: + self.reader.join(timeout=5) + for handle in self.handles: + close(handle) + self.handles = [] + if self.reader: + self.reader.join(timeout=5) + + +@unittest.skipUnless(IS_WINDOWS, "requires Windows console APIs") +class ConPtyNotificationTests(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.temp = tempfile.TemporaryDirectory(prefix="codex warp conpty ") + cls.addClassCleanup(cls.temp.cleanup) + cls.root = Path(cls.temp.name) + cls.codex = cls.root / "codex.exe" + source = ROOT / "tests" / "console-parent.cs" + compile_command = ( + "Add-Type -Path '" + str(source).replace("'", "''") + + "' -OutputAssembly '" + str(cls.codex).replace("'", "''") + + "' -OutputType ConsoleApplication" + ) + subprocess.run(["powershell.exe", "-NoProfile", "-Command", compile_command], + check=True, capture_output=True, timeout=30) + cls.scripts = cls.root / "plugin scripts" + shutil.copytree(ROOT / "plugins" / "warp" / "scripts", cls.scripts) + (cls.scripts / "control-output.ps1").write_text( + ". \"$PSScriptRoot\\on-session-start.ps1\"\n" + "[Console]::Out.WriteLine('{\"control\":true}')\n" + "[Console]::Error.WriteLine('control-stderr')\n", + encoding="utf-8", + ) + + (cls.scripts / "dead-parent.ps1").write_text( + "$parent = Get-Process -Id ([int]$env:CONPTY_TEST_PARENT_PID) -ErrorAction SilentlyContinue\n" + "$deadline = [DateTime]::UtcNow.AddSeconds(10)\n" + "while ($null -ne $parent -and -not $parent.HasExited) {\n" + " if ([DateTime]::UtcNow -gt $deadline) { throw 'parent did not exit' }\n" + " Start-Sleep -Milliseconds 20\n" + "}\n" + ". \"$PSScriptRoot\\on-session-start.ps1\"\n", + encoding="utf-8", + ) + + def run_hook(self, hook, detached=True, wrapper=True, count=1, owner=True, role=None): + with tempfile.TemporaryDirectory(dir=self.root) as directory: + result_root = Path(directory) + payload = result_root / "input.json" + payload.write_text(json.dumps({ + "session_id": "test-session", "cwd": "C:\\work\\project", + "prompt": "Check café and \U0001f600", "last_assistant_message": "Done café \U0001f600", + }, ensure_ascii=False), encoding="utf-8") + executable = self.codex + if not owner: + executable = self.root / "unrelated-parent.exe" + if not executable.exists(): + shutil.copyfile(self.codex, executable) + command = [ + str(executable), str(self.scripts / hook), str(payload), directory, + "detached" if detached else "inherited", + "cmd" if wrapper else "direct", str(count), + ] + if role: + command.append(role) + pty = ConPty(command) + try: + pty.complete() + results = [ + json.loads((result_root / ("result-%d.json" % i)).read_text(encoding="utf-8-sig")) + for i in range(count) + ] + for result in results: + result["stdout"] = base64.b64decode(result.pop("stdout_base64")).decode("utf-8") + result["stderr"] = base64.b64decode(result.pop("stderr_base64")).decode("utf-8") + received = pty.notifications(count if owner and role != "dead" else 0) + return results, received + finally: + pty.close() + + def test_inherited_and_detached_delivery(self): + for detached in (False, True): + for wrapper in (False, True): + with self.subTest(detached=detached, wrapper=wrapper): + results, received = self.run_hook( + "on-session-start.ps1", detached=detached, wrapper=wrapper, + ) + self.assertEqual(results, [{"exit_code": 0, "stdout": "", "stderr": ""}]) + self.assertEqual(len(received), 1, "no lifecycle event on ConPTY output") + self.assertEqual(received[0]["event"], "session_start") + self.assertEqual(received[0]["session_id"], "test-session") + + def test_prompt_and_stop_delivery_with_unicode(self): + for hook, event, field, text in ( + ("on-prompt-submit.ps1", "prompt_submit", "query", "Check café and \U0001f600"), + ("on-stop.ps1", "stop", "response", "Done café \U0001f600"), + ): + with self.subTest(hook=hook): + results, received = self.run_hook(hook) + self.assertEqual(results, [{"exit_code": 0, "stdout": "", "stderr": ""}]) + self.assertEqual(len(received), 1) + self.assertEqual(received[0]["event"], event) + self.assertEqual(received[0][field], text) + + def test_concurrent_detached_hooks(self): + results, received = self.run_hook("on-session-start.ps1", count=4) + self.assertEqual(results, [{"exit_code": 0, "stdout": "", "stderr": ""}] * 4) + self.assertEqual(len(received), 4) + self.assertTrue(all(payload["event"] == "session_start" for payload in received)) + + def test_control_output_stays_captured(self): + results, received = self.run_hook("control-output.ps1") + self.assertEqual(results, [{ + "exit_code": 0, "stdout": '{"control":true}\r\n', "stderr": "control-stderr\r\n", + }]) + self.assertEqual(len(received), 1) + self.assertEqual(received[0]["event"], "session_start") + + def test_unrelated_ancestor_is_not_used(self): + results, received = self.run_hook("on-session-start.ps1", owner=False) + self.assertEqual(results[0]["exit_code"], 0) + self.assertEqual(results[0]["stdout"], "") + self.assertIn("no live Codex ancestor", results[0]["stderr"]) + self.assertEqual(received, []) + + def test_exited_closest_codex_does_not_route_to_live_ancestor(self): + results, received = self.run_hook("dead-parent.ps1", role="dead") + self.assertEqual(results[0]["exit_code"], 0) + self.assertEqual(results[0]["stdout"], "") + self.assertIn("no live Codex ancestor", results[0]["stderr"]) + self.assertEqual(received, []) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test-hooks.ps1 b/tests/test-hooks.ps1 index abc890c..afa73db 100644 --- a/tests/test-hooks.ps1 +++ b/tests/test-hooks.ps1 @@ -38,9 +38,16 @@ function Invoke-PowerShellHook { "-File", $ScriptPath ) + $ScriptArguments - $output = @($InputJson | & powershell.exe @powershellArguments 2>&1) - if ($LASTEXITCODE -ne 0) { - throw "$ScriptPath exited with code $LASTEXITCODE.`n$($output -join "`n")" + $previousPreference = $ErrorActionPreference + try { + $ErrorActionPreference = "Continue" + $output = @($InputJson | & powershell.exe @powershellArguments 2>&1) + $exitCode = $LASTEXITCODE + } finally { + $ErrorActionPreference = $previousPreference + } + if ($exitCode -ne 0) { + throw "$ScriptPath exited with code $exitCode.`n$($output -join "`n")" } return $output