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<uint, ProcessEntry>();
+        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<uint>();
+        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<Process>();
+        var outputs = new List<Task<string>>();
+        var errors = new List<Task<string>>();
+        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