From 879080dcf87a80f6a28f8d76197f124fa7bcda18 Mon Sep 17 00:00:00 2001
From: Paul Querna <paul.querna@conductorone.com>
Date: Wed, 29 Jul 2026 17:23:49 +0000
Subject: [PATCH 1/3] test(windows): ConPTY e2e coverage for the PR #124 parity
 fixes
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Adds a #[cfg(windows)] integration suite with no Cargo-feature gate, so
the stock 'cargo test --workspace' CI leg on windows-latest runs it (the
docker/bwrap suites stay feature-gated and Linux-only). Six tests, one
per fix in #124:

1. run | consumer in a cmd.exe pipeline sees prompt EOF (win_spawn
   handle-inheritance allow-list; 90s bound vs the 15min idle timeout
   that masked the pre-fix hang).
2. Interactive pane's startup DSR (ESC[6n) is answered — new dsr_probe
   bin (src/bin) enables VT input on its own console, emits the query,
   and prints the reply; without take_pty_writes write-back it blocks
   forever.
3. tail --follow delivers a fast-exiting child's trailing bytes before
   the stream ends (exit_drained / DRAIN_GRACE).
4. die reaps the descendant tree (verified via Win32_Process with a
   per-test sentinel in the grandchild's command line).
5. signal SIGINT interrupts via ConPTY ETX (ping -t exits).
6. signal SIGBREAK is rejected with an actionable error.

Each test runs an isolated session + socket dir and ties the daemon to
the test process (AGENT_TUI_MONITOR_PARENT_PID) plus a best-effort
'daemon shutdown' on drop, so the file is parallel-safe and leaves no
orphaned daemons.

Verified: cargo check/clippy --all-targets -D warnings for
x86_64-pc-windows-msvc, cargo fmt --check, cargo test -p
agent-tui-integration (Linux aarch64; the new file is cfg-stripped
there). The tests themselves require a Windows host — the windows-latest
CI leg is the first execution environment.

Co-authored-by: c1-squire-dev[bot] <c1-squire-dev[bot]@users.noreply.github.com>
---
 Cargo.lock                                    |   1 +
 crates/agent-tui-integration/Cargo.toml       |   5 +
 .../src/bin/dsr_probe.rs                      |  83 ++++
 .../tests/windows_conpty.rs                   | 409 ++++++++++++++++++
 4 files changed, 498 insertions(+)
 create mode 100644 crates/agent-tui-integration/src/bin/dsr_probe.rs
 create mode 100644 crates/agent-tui-integration/tests/windows_conpty.rs

diff --git a/Cargo.lock b/Cargo.lock
index bdc61d7..03af5b7 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -137,6 +137,7 @@ dependencies = [
  "tracing",
  "tracing-subscriber",
  "uuid",
+ "windows-sys 0.59.0",
 ]
 
 [[package]]
diff --git a/crates/agent-tui-integration/Cargo.toml b/crates/agent-tui-integration/Cargo.toml
index ace53c7..3d0404c 100644
--- a/crates/agent-tui-integration/Cargo.toml
+++ b/crates/agent-tui-integration/Cargo.toml
@@ -40,6 +40,11 @@ chrono = { workspace = true }
 [target.'cfg(unix)'.dependencies]
 nix = { workspace = true }
 
+[target.'cfg(windows)'.dependencies]
+# GetConsoleMode/SetConsoleMode for the dsr_probe test binary's VT-input setup
+# (src/bin/dsr_probe.rs). Same windows-sys version the daemon uses.
+windows-sys = { version = "0.59", features = ["Win32_System_Console", "Win32_Foundation"] }
+
 [dev-dependencies]
 # Optional Docker-API features in the dev tier so `cargo test -p
 # agent-tui-integration --features docker` works without --no-default-features.
diff --git a/crates/agent-tui-integration/src/bin/dsr_probe.rs b/crates/agent-tui-integration/src/bin/dsr_probe.rs
new file mode 100644
index 0000000..e168b11
--- /dev/null
+++ b/crates/agent-tui-integration/src/bin/dsr_probe.rs
@@ -0,0 +1,83 @@
+//! Windows-only test probe for the ConPTY DSR (Device Status Report) write-back
+//! path shipped in PR #124.
+//!
+//! Interactive programs emit `ESC[6n` at startup and block until the terminal
+//! answers with a cursor-position report (`ESC[<row>;<col>R`). The daemon's
+//! engine produces that reply (`Event::PtyWrite` → `take_pty_writes`) and the
+//! reader loop writes it back to the ConPTY master. This probe is the minimal
+//! such child: it enables VT input on its own console (so the reply arrives as
+//! raw bytes, the way a real TUI reads it), emits `ESC[6n`, reads the reply,
+//! and prints it as `DSR-REPLY:<bytes>`. The companion test
+//! (`tests/windows_conpty.rs::interactive_pane_answers_startup_dsr`) asserts
+//! the reply arrives; without the write-back the read below blocks forever.
+//!
+//! Built as a bin of the integration crate so `cargo test` places it next to
+//! `agent-tui.exe` under `target/debug/` on the Windows CI leg. On non-Windows
+//! it compiles to an inert stub so the workspace builds everywhere.
+
+#![deny(unsafe_code)]
+
+#[cfg(windows)]
+mod imp {
+    use std::io::{Read, Write};
+
+    use windows_sys::Win32::Foundation::HANDLE;
+    use windows_sys::Win32::System::Console::{
+        GetConsoleMode, GetStdHandle, STD_INPUT_HANDLE, SetConsoleMode,
+    };
+
+    /// `ENABLE_VIRTUAL_TERMINAL_INPUT` (0x0200): deliver VT sequences — including
+    /// the terminal's DSR reply — as raw input bytes instead of cooked key events.
+    const ENABLE_VIRTUAL_TERMINAL_INPUT: u32 = 0x0200;
+
+    pub(crate) fn run() -> i32 {
+        // SAFETY: standard Win32 console calls on our own process's stdin
+        // handle; the returned handle is borrowed, never closed here.
+        #[allow(unsafe_code)]
+        unsafe {
+            let h: HANDLE = GetStdHandle(STD_INPUT_HANDLE);
+            if h.is_null() {
+                return 2;
+            }
+            let mut mode: u32 = 0;
+            if GetConsoleMode(h, &raw mut mode) == 0 {
+                return 2;
+            }
+            // VT input only: no line buffering, no echo of the reply bytes.
+            if SetConsoleMode(h, ENABLE_VIRTUAL_TERMINAL_INPUT) == 0 {
+                return 2;
+            }
+        }
+
+        print!("\x1b[6n");
+        let _ = std::io::stdout().flush();
+
+        let mut buf = [0u8; 64];
+        let mut n = 0usize;
+        let stdin = std::io::stdin();
+        let mut lock = stdin.lock();
+        while n < buf.len() {
+            match lock.read(&mut buf[n..=n]) {
+                Ok(0) | Err(_) => break,
+                Ok(m) => {
+                    n += m;
+                    if buf[n - 1] == b'R' {
+                        break;
+                    }
+                }
+            }
+        }
+        println!("DSR-REPLY:{}", String::from_utf8_lossy(&buf[..n]));
+        0
+    }
+}
+
+fn main() {
+    #[cfg(windows)]
+    std::process::exit(imp::run());
+    #[cfg(not(windows))]
+    {
+        eprintln!("dsr_probe is a Windows-only test probe");
+        std::process::exit(2);
+    }
+}
diff --git a/crates/agent-tui-integration/tests/windows_conpty.rs b/crates/agent-tui-integration/tests/windows_conpty.rs
new file mode 100644
index 0000000..9c1c4ba
--- /dev/null
+++ b/crates/agent-tui-integration/tests/windows_conpty.rs
@@ -0,0 +1,409 @@
+//! Windows ConPTY end-to-end coverage for the parity work shipped in PR #124.
+//!
+//! These tests are `#[cfg(windows)]` with NO Cargo-feature gate, so the stock
+//! `cargo test --workspace` CI leg on `windows-latest` runs them (the
+//! docker/bwrap suites are feature-gated and stay Linux-only). Each test gets
+//! an isolated session + socket dir, so the file is safe under cargo's
+//! default parallel test threads, and every rig ties its daemon's lifetime to
+//! the test process (`AGENT_TUI_MONITOR_PARENT_PID`) plus a best-effort
+//! `daemon shutdown` on drop.
+//!
+//! Covered (each maps to a fix in #124):
+//!  1. `run | consumer` in a `cmd.exe` pipeline sees prompt EOF (the `win_spawn`
+//!     handle-inheritance allow-list).
+//!  2. An interactive pane's startup DSR (`ESC[6n`) is answered, unblocking
+//!     the child (engine `take_pty_writes` write-back via `dsr_probe`).
+//!  3. `tail --follow` delivers a fast-exiting child's trailing bytes before
+//!     the stream ends (`exit_drained` / `DRAIN_GRACE`).
+//!  4. `die` reaps the whole descendant tree (`taskkill /F /T`).
+//!  5. `signal SIGINT` interrupts via ETX written to the ConPTY input.
+//!  6. `signal SIGBREAK` is rejected with an actionable error (no false
+//!     success).
+
+#![cfg(windows)]
+
+use std::future::Future;
+use std::path::PathBuf;
+use std::process::Stdio;
+use std::time::Duration;
+
+use agent_tui_integration::{agent_tui_binary, workspace_root};
+use anyhow::{Context, Result, bail};
+
+/// Isolated per-test daemon context: unique session, private socket dir and
+/// state home under %TEMP%. Drop shuts the daemon down best-effort and
+/// removes the directory.
+struct Rig {
+    session: String,
+    root: PathBuf,
+    socket_dir: PathBuf,
+    state_home: PathBuf,
+}
+
+impl Rig {
+    fn new(name: &str) -> Result<Self> {
+        let id = uuid::Uuid::new_v4().simple().to_string();
+        let root = std::env::temp_dir().join(format!("agent-tui-win-e2e-{name}-{}", &id[..8]));
+        let socket_dir = root.join("sock");
+        let state_home = root.join("state");
+        std::fs::create_dir_all(&socket_dir)
+            .with_context(|| format!("create {}", socket_dir.display()))?;
+        std::fs::create_dir_all(&state_home)
+            .with_context(|| format!("create {}", state_home.display()))?;
+        Ok(Self {
+            session: format!("e2e-{}", &id[..8]),
+            root,
+            socket_dir,
+            state_home,
+        })
+    }
+
+    /// Per-rig env every child process needs (CLI invocations and the
+    /// cmd.exe pipeline repro alike).
+    fn apply_env(&self, cmd: &mut tokio::process::Command) {
+        cmd.env("AGENT_TUI_SOCKET_DIR", &self.socket_dir)
+            .env("AGENT_TUI_ALLOWED_BINARIES", "*")
+            // Tie any lazy-spawned daemon's lifetime to this test process so
+            // a panicking or SIGKILLed test runner doesn't orphan a daemon.
+            .env(
+                "AGENT_TUI_MONITOR_PARENT_PID",
+                std::process::id().to_string(),
+            )
+            .env("XDG_STATE_HOME", &self.state_home)
+            .stdin(Stdio::null())
+            .stdout(Stdio::piped())
+            .stderr(Stdio::piped());
+    }
+
+    /// Base command with the per-rig env every invocation needs. `json`
+    /// selects the global `--json` flag; text surfaces (`tail`, streaming
+    /// verbs) are asserted on raw stdout instead.
+    fn cli(&self, json: bool) -> tokio::process::Command {
+        let bin = agent_tui_binary().expect("agent-tui binary built by cargo test");
+        let mut cmd = tokio::process::Command::new(bin);
+        cmd.arg("--session")
+            .arg(&self.session)
+            .arg("--socket-dir")
+            .arg(&self.socket_dir);
+        if json {
+            cmd.arg("--json");
+        }
+        self.apply_env(&mut cmd);
+        cmd
+    }
+
+    async fn output(&self, json: bool, args: &[&str]) -> Result<std::process::Output> {
+        let mut cmd = self.cli(json);
+        Ok(cmd.args(args).output().await?)
+    }
+
+    /// `cmd.exe /d /c <script>` with the rig env — used for the pipeline repro,
+    /// where the point is `cmd`'s own handle-inheritance behavior.
+    async fn cmd_exe(&self, script: &str) -> Result<std::process::Output> {
+        let mut cmd = tokio::process::Command::new("cmd.exe");
+        cmd.args(["/d", "/c", script]);
+        self.apply_env(&mut cmd);
+        Ok(cmd.output().await?)
+    }
+
+    async fn spawn_pane(&self, argv: &[&str]) -> Result<String> {
+        let mut spawn_args: Vec<&str> = vec!["spawn", "--"];
+        spawn_args.extend_from_slice(argv);
+        let out = self.output(true, &spawn_args).await?;
+        if !out.status.success() {
+            bail!(
+                "spawn {argv:?} failed: {}",
+                String::from_utf8_lossy(&out.stderr)
+            );
+        }
+        let v: serde_json::Value = serde_json::from_slice(&out.stdout).with_context(|| {
+            format!(
+                "parse spawn response: {}",
+                String::from_utf8_lossy(&out.stdout)
+            )
+        })?;
+        find_pane_id(&v).with_context(|| format!("no pane id in spawn response: {v}"))
+    }
+
+    async fn die(&self, pane: &str) {
+        let _ = self.output(false, &["die", "--pane", pane]).await;
+    }
+}
+
+impl Drop for Rig {
+    fn drop(&mut self) {
+        // Best-effort daemon shutdown (blocking std Command — Drop is sync).
+        if let Ok(bin) = agent_tui_binary() {
+            let _ = std::process::Command::new(bin)
+                .arg("--session")
+                .arg(&self.session)
+                .arg("--socket-dir")
+                .arg(&self.socket_dir)
+                .args(["daemon", "shutdown"])
+                .env("AGENT_TUI_SOCKET_DIR", &self.socket_dir)
+                .env("XDG_STATE_HOME", &self.state_home)
+                .stdin(Stdio::null())
+                .stdout(Stdio::null())
+                .stderr(Stdio::null())
+                .status();
+        }
+        let _ = std::fs::remove_dir_all(&self.root);
+    }
+}
+
+/// First `"pane": "pN"` string anywhere in the envelope (lenient about the
+/// exact nesting so the test doesn't couple to envelope layout).
+fn find_pane_id(v: &serde_json::Value) -> Option<String> {
+    match v {
+        serde_json::Value::Object(map) => {
+            if let Some(serde_json::Value::String(s)) = map.get("pane")
+                && s.starts_with('p')
+                && s.len() > 1
+                && s[1..].chars().all(|c| c.is_ascii_digit())
+            {
+                return Some(s.clone());
+            }
+            map.values().find_map(find_pane_id)
+        }
+        serde_json::Value::Array(a) => a.iter().find_map(find_pane_id),
+        _ => None,
+    }
+}
+
+/// Poll `f` until it returns true or `max` elapses.
+async fn poll_until<F, Fut>(what: &str, max: Duration, mut f: F) -> Result<()>
+where
+    F: FnMut() -> Fut,
+    Fut: Future<Output = Result<bool>>,
+{
+    let start = std::time::Instant::now();
+    loop {
+        if f().await? {
+            return Ok(());
+        }
+        if start.elapsed() >= max {
+            bail!("timed out ({max:?}) waiting for {what}");
+        }
+        tokio::time::sleep(Duration::from_millis(200)).await;
+    }
+}
+
+/// Locate the `dsr_probe` test binary, mirroring `agent_tui_binary`'s
+/// debug-then-release candidate order.
+fn dsr_probe_path() -> Result<PathBuf> {
+    let root = workspace_root()?;
+    for c in [
+        root.join("target").join("debug").join("dsr_probe.exe"),
+        root.join("target").join("release").join("dsr_probe.exe"),
+    ] {
+        if c.exists() {
+            return Ok(c);
+        }
+    }
+    bail!("dsr_probe.exe not built under {}/target", root.display())
+}
+
+/// Count processes whose command line contains `sentinel`, excluding the
+/// querying PowerShell itself (its own command line carries the sentinel).
+async fn sentinel_process_count(sentinel: &str) -> Result<usize> {
+    let ps = format!(
+        "(Get-CimInstance Win32_Process | Where-Object {{ $_.ProcessId -ne $PID -and $_.CommandLine -like '*{sentinel}*' }} | Measure-Object).Count"
+    );
+    let out = tokio::process::Command::new("powershell")
+        .args(["-NoProfile", "-NonInteractive", "-Command", &ps])
+        .output()
+        .await
+        .context("spawn powershell")?;
+    let s = String::from_utf8_lossy(&out.stdout);
+    s.trim()
+        .parse::<usize>()
+        .with_context(|| format!("unexpected powershell output: {s:?}"))
+}
+
+/// Pre-#124, a lazily-spawned daemon inside a `cmd.exe` pipeline inherited an
+/// extra inheritable pipe handle that cmd leaves in the client, so
+/// `agent-tui run ... | consumer` hung until the daemon idle timeout (15 min
+/// default). The `win_spawn` handle-inheritance allow-list makes EOF prompt.
+/// 90s is far under the idle timeout and far over a healthy run on a loaded
+/// CI runner.
+#[tokio::test]
+async fn run_in_cmd_pipeline_consumer_sees_prompt_eof() -> Result<()> {
+    let rig = Rig::new("run-pipe-eof")?;
+    let bin = agent_tui_binary()?;
+    let script = format!(
+        "\"{}\" --session {} --socket-dir \"{}\" run -- cmd /c echo agent-tui-e2e-eof | findstr agent-tui-e2e-eof",
+        bin.display(),
+        rig.session,
+        rig.socket_dir.display()
+    );
+    let out = tokio::time::timeout(Duration::from_secs(90), rig.cmd_exe(&script))
+        .await
+        .context("consumer did not see EOF within 90s (handle-inheritance regression?)")??;
+    assert!(
+        out.status.success(),
+        "pipeline failed: {}",
+        String::from_utf8_lossy(&out.stderr)
+    );
+    let stdout = String::from_utf8_lossy(&out.stdout);
+    assert!(
+        stdout.contains("agent-tui-e2e-eof"),
+        "consumer saw no payload: {stdout}"
+    );
+    Ok(())
+}
+
+/// The engine must answer the ConPTY startup DSR (`ESC[6n`) or interactive
+/// children block waiting for the cursor-position report. `dsr_probe` is the
+/// minimal such child: without the `take_pty_writes` write-back its read
+/// blocks forever and the sentinel never appears.
+#[tokio::test]
+async fn interactive_pane_answers_startup_dsr() -> Result<()> {
+    let rig = Rig::new("dsr")?;
+    let probe = dsr_probe_path()?;
+    let probe_str = probe.to_str().context("probe path not utf-8")?;
+    let pane = rig.spawn_pane(&[probe_str]).await?;
+    poll_until("DSR reply on pane", Duration::from_secs(30), || async {
+        let out = rig
+            .output(false, &["tail", "--pane", &pane, "--strip-ansi"])
+            .await?;
+        Ok(String::from_utf8_lossy(&out.stdout).contains("DSR-REPLY:"))
+    })
+    .await?;
+    let out = rig
+        .output(false, &["tail", "--pane", &pane, "--strip-ansi"])
+        .await?;
+    let text = String::from_utf8_lossy(&out.stdout);
+    // --strip-ansi removes the reply's ESC, leaving "DSR-REPLY:[<row>;<col>R".
+    let reply = text.split("DSR-REPLY:").nth(1).unwrap_or("");
+    assert!(
+        reply.trim_end().starts_with('[') && reply.trim_end().ends_with('R'),
+        "reply is not a cursor-position report: {text}"
+    );
+    rig.die(&pane).await;
+    Ok(())
+}
+
+/// A fast child that writes-then-exits must have its trailing bytes delivered
+/// before the follow stream's terminal gate (`exit_drained` / `DRAIN_GRACE`)
+/// ends the stream — and the stream must end (the ConPTY EOF fix).
+#[tokio::test]
+async fn tail_follow_delivers_trailing_output_after_fast_exit() -> Result<()> {
+    let rig = Rig::new("follow-drain")?;
+    let pane = rig
+        .spawn_pane(&["cmd", "/c", "for /l %i in (1,1,200) do @echo line%i"])
+        .await?;
+    let out = tokio::time::timeout(
+        Duration::from_secs(30),
+        rig.output(
+            false,
+            &["tail", "--pane", &pane, "--follow", "--strip-ansi"],
+        ),
+    )
+    .await
+    .context("tail --follow did not terminate (drain/EOF regression?)")??;
+    let text = String::from_utf8_lossy(&out.stdout);
+    assert!(text.contains("line1"), "early output missing: {text}");
+    assert!(
+        text.contains("line200"),
+        "trailing output dropped at the terminal gate: {text}"
+    );
+    Ok(())
+}
+
+/// `die` must kill the whole descendant tree (`taskkill /F /T`), not just the
+/// direct child. The direct child keeps itself alive with its own `ping -t`
+/// so the tree root still exists at die time; the grandchild carries the
+/// sentinel in its command line so `Win32_Process` can find it.
+#[tokio::test]
+async fn die_reaps_descendant_tree() -> Result<()> {
+    let rig = Rig::new("tree-kill")?;
+    let sentinel = format!(
+        "e2e-tree-{}",
+        &uuid::Uuid::new_v4().simple().to_string()[..8]
+    );
+    let pane = rig
+        .spawn_pane(&[
+            "cmd",
+            "/c",
+            &format!(
+                "start \"\" /b cmd /c \"ping -t 127.0.0.1 >nul & rem {sentinel}\" & ping -t 127.0.0.1 >nul"
+            ),
+        ])
+        .await?;
+    poll_until("grandchild to appear", Duration::from_secs(30), || async {
+        Ok(sentinel_process_count(&sentinel).await? >= 1)
+    })
+    .await?;
+    let out = rig.output(false, &["die", "--pane", &pane]).await?;
+    assert!(
+        out.status.success(),
+        "die failed: {}",
+        String::from_utf8_lossy(&out.stderr)
+    );
+    poll_until(
+        "descendant tree reaped",
+        Duration::from_secs(30),
+        || async { Ok(sentinel_process_count(&sentinel).await? == 0) },
+    )
+    .await
+    .context("grandchild survived die (tree-kill regression?)")?;
+    Ok(())
+}
+
+/// `signal SIGINT` writes ETX to the ConPTY input; conhost raises a real
+/// `CTRL_C_EVENT` and `ping -t` exits.
+#[tokio::test]
+async fn sigint_via_conpty_etx_interrupts_child() -> Result<()> {
+    let rig = Rig::new("sigint")?;
+    let pane = rig.spawn_pane(&["ping", "-t", "127.0.0.1"]).await?;
+    // Give ping a beat to be running before the interrupt lands.
+    tokio::time::sleep(Duration::from_secs(2)).await;
+    let out = rig
+        .output(false, &["signal", "--pane", &pane, "SIGINT"])
+        .await?;
+    assert!(
+        out.status.success(),
+        "signal SIGINT failed: {}",
+        String::from_utf8_lossy(&out.stderr)
+    );
+    let out = rig
+        .output(
+            false,
+            &["wait", "--pane", &pane, "--exit", "--max", "15000"],
+        )
+        .await?;
+    assert!(
+        out.status.success(),
+        "child did not exit after SIGINT: {}",
+        String::from_utf8_lossy(&out.stderr)
+    );
+    rig.die(&pane).await;
+    Ok(())
+}
+
+/// `signal SIGBREAK` has no honest delivery path to a ConPTY child, so it must
+/// fail with an actionable error rather than report a false success.
+#[tokio::test]
+async fn sigbreak_rejected_with_actionable_error() -> Result<()> {
+    let rig = Rig::new("sigbreak")?;
+    let pane = rig.spawn_pane(&["ping", "-t", "127.0.0.1"]).await?;
+    let out = rig
+        .output(false, &["signal", "--pane", &pane, "SIGBREAK"])
+        .await?;
+    assert!(
+        !out.status.success(),
+        "SIGBREAK must not report success: {}",
+        String::from_utf8_lossy(&out.stdout)
+    );
+    let combined = format!(
+        "{}{}",
+        String::from_utf8_lossy(&out.stdout),
+        String::from_utf8_lossy(&out.stderr)
+    );
+    assert!(
+        combined.contains("cannot be delivered"),
+        "expected the honest-error text, got: {combined}"
+    );
+    rig.die(&pane).await;
+    Ok(())
+}

From de2eacf6c544621fdab4c97bee8bc4d1377618eb Mon Sep 17 00:00:00 2001
From: Paul Querna <paul.querna@conductorone.com>
Date: Wed, 29 Jul 2026 17:42:53 +0000
Subject: [PATCH 2/3] test(windows): fix three e2e failures from first CI run
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Iteration 2 from windows-latest CI data (3 passed, 3 failed):

- run-pipe-eof: Rust's CommandLineToArgvW quoting (" for embedded
  quotes) collides with cmd /c's outer-quote stripping ('"...exe" is
  not recognized'). Write the pipeline to a .cmd file instead — cmd
  parses batch lines natively.
- dsr: probe read 0 bytes (reply never became readable stdin). Probe is
  now conpty_probe with MODE-SET/DSR-READ diagnostics, and the test
  parses the tail JSON envelope's data.text instead of raw stdout.
- sigint: the signal RPC succeeded (ETX written) but ping never exited —
  conhost's ETX -> CTRL_C_EVENT translation is mode- and Windows-
  version-dependent (author's Win11 e2e saw it land; windows-latest CI
  did not). Repin the test on the deterministic daemon-side half: ETX
  byte delivery to the child's stdin, via conpty_probe read-stdin.

Co-authored-by: c1-squire-dev[bot] <c1-squire-dev[bot]@users.noreply.github.com>
---
 .../src/bin/conpty_probe.rs                   | 136 ++++++++++++++++++
 .../src/bin/dsr_probe.rs                      |  83 -----------
 .../tests/windows_conpty.rs                   | 112 ++++++++++-----
 3 files changed, 215 insertions(+), 116 deletions(-)
 create mode 100644 crates/agent-tui-integration/src/bin/conpty_probe.rs
 delete mode 100644 crates/agent-tui-integration/src/bin/dsr_probe.rs

diff --git a/crates/agent-tui-integration/src/bin/conpty_probe.rs b/crates/agent-tui-integration/src/bin/conpty_probe.rs
new file mode 100644
index 0000000..e59c39e
--- /dev/null
+++ b/crates/agent-tui-integration/src/bin/conpty_probe.rs
@@ -0,0 +1,136 @@
+//! Windows-only test probes for the ConPTY paths shipped in PR #124.
+//!
+//! Two modes, used by `tests/windows_conpty.rs`:
+//!
+//! `conpty_probe dsr`
+//!   The minimal interactive child that blocks on the terminal's answer to the
+//!   startup DSR (`ESC[6n` → `ESC[<row>;<col>R`). Enables VT input on its own
+//!   console (so the reply arrives as raw bytes, the way a real TUI reads it),
+//!   emits the query, reads the reply, and prints it as `DSR-REPLY:<bytes>`
+//!   with `DSR-READ:` diagnostics. Without the daemon's `take_pty_writes`
+//!   write-back the read never produces the reply.
+//!
+//! `conpty_probe read-stdin`
+//!   Reads up to 16 stdin bytes (stopping after `0x03`) and prints them as
+//!   `STDIN-HEX:<hex>`. Proves the daemon's ConPTY-input write path
+//!   (`signal SIGINT` writes ETX) end to end, independent of conhost's
+//!   mode-dependent ETX → `CTRL_C_EVENT` translation.
+//!
+//! Built as a bin of the integration crate so `cargo test` places it next to
+//! `agent-tui.exe` under `target/debug/` on the Windows CI leg. On non-Windows
+//! it compiles to an inert stub so the workspace builds everywhere.
+
+#![deny(unsafe_code)]
+
+#[cfg(windows)]
+mod imp {
+    use std::fmt::Write as _;
+    use std::io::{Read, Write};
+
+    use windows_sys::Win32::Foundation::HANDLE;
+    use windows_sys::Win32::System::Console::{
+        GetConsoleMode, GetStdHandle, STD_INPUT_HANDLE, SetConsoleMode,
+    };
+
+    /// `ENABLE_VIRTUAL_TERMINAL_INPUT` (0x0200): deliver VT sequences — including
+    /// terminal replies and ETX — as raw input bytes instead of cooked key
+    /// events or processed-input control events.
+    const ENABLE_VIRTUAL_TERMINAL_INPUT: u32 = 0x0200;
+
+    /// Enable VT input on our own console. Returns false (with a printed
+    /// diagnostic) if any step fails — on a ConPTY client these handles are
+    /// console handles, so failure here is itself interesting test data.
+    fn enable_vt_input() -> bool {
+        // SAFETY: standard Win32 console calls on our own process's stdin
+        // handle; the returned handle is borrowed, never closed here.
+        #[allow(unsafe_code)]
+        unsafe {
+            let h: HANDLE = GetStdHandle(STD_INPUT_HANDLE);
+            if h.is_null() {
+                println!("MODE-SET:fail:null-handle");
+                return false;
+            }
+            let mut mode: u32 = 0;
+            if GetConsoleMode(h, &raw mut mode) == 0 {
+                println!("MODE-SET:fail:get-mode:{}", std::io::Error::last_os_error());
+                return false;
+            }
+            if SetConsoleMode(h, ENABLE_VIRTUAL_TERMINAL_INPUT) == 0 {
+                println!("MODE-SET:fail:set-mode:{}", std::io::Error::last_os_error());
+                return false;
+            }
+            println!("MODE-SET:ok:was-{mode:#x}");
+            true
+        }
+    }
+
+    /// Read up to `buf.len()` bytes, stopping early at `stop` or EOF/error.
+    /// Returns (bytes-read, status string) for diagnostics.
+    fn read_until(buf: &mut [u8], stop: u8) -> (usize, String) {
+        let mut n = 0usize;
+        let stdin = std::io::stdin();
+        let mut lock = stdin.lock();
+        while n < buf.len() {
+            match lock.read(&mut buf[n..=n]) {
+                Ok(0) => return (n, "eof".to_string()),
+                Ok(m) => {
+                    n += m;
+                    if buf[n - 1] == stop {
+                        return (n, "stop-byte".to_string());
+                    }
+                }
+                Err(e) => return (n, format!("err:{e}")),
+            }
+        }
+        (n, "full".to_string())
+    }
+
+    fn dsr() -> i32 {
+        if !enable_vt_input() {
+            return 2;
+        }
+        print!("\x1b[6n");
+        let _ = std::io::stdout().flush();
+        let mut buf = [0u8; 64];
+        let (n, status) = read_until(&mut buf, b'R');
+        println!("DSR-READ:n={n}:status={status}");
+        println!("DSR-REPLY:{}", String::from_utf8_lossy(&buf[..n]));
+        0
+    }
+
+    fn read_stdin() -> i32 {
+        if !enable_vt_input() {
+            return 2;
+        }
+        let mut buf = [0u8; 16];
+        let (n, status) = read_until(&mut buf, 0x03);
+        let mut hex = String::with_capacity(n * 2);
+        for b in &buf[..n] {
+            let _ = write!(hex, "{b:02x}");
+        }
+        println!("STDIN-READ:n={n}:status={status}");
+        println!("STDIN-HEX:{hex}");
+        0
+    }
+
+    pub(crate) fn run() -> i32 {
+        match std::env::args().nth(1).as_deref() {
+            Some("dsr") => dsr(),
+            Some("read-stdin") => read_stdin(),
+            other => {
+                eprintln!("usage: conpty_probe <dsr|read-stdin>, got {other:?}");
+                2
+            }
+        }
+    }
+}
+
+fn main() {
+    #[cfg(windows)]
+    std::process::exit(imp::run());
+    #[cfg(not(windows))]
+    {
+        eprintln!("conpty_probe is a Windows-only test probe");
+        std::process::exit(2);
+    }
+}
diff --git a/crates/agent-tui-integration/src/bin/dsr_probe.rs b/crates/agent-tui-integration/src/bin/dsr_probe.rs
deleted file mode 100644
index e168b11..0000000
--- a/crates/agent-tui-integration/src/bin/dsr_probe.rs
+++ /dev/null
@@ -1,83 +0,0 @@
-//! Windows-only test probe for the ConPTY DSR (Device Status Report) write-back
-//! path shipped in PR #124.
-//!
-//! Interactive programs emit `ESC[6n` at startup and block until the terminal
-//! answers with a cursor-position report (`ESC[<row>;<col>R`). The daemon's
-//! engine produces that reply (`Event::PtyWrite` → `take_pty_writes`) and the
-//! reader loop writes it back to the ConPTY master. This probe is the minimal
-//! such child: it enables VT input on its own console (so the reply arrives as
-//! raw bytes, the way a real TUI reads it), emits `ESC[6n`, reads the reply,
-//! and prints it as `DSR-REPLY:<bytes>`. The companion test
-//! (`tests/windows_conpty.rs::interactive_pane_answers_startup_dsr`) asserts
-//! the reply arrives; without the write-back the read below blocks forever.
-//!
-//! Built as a bin of the integration crate so `cargo test` places it next to
-//! `agent-tui.exe` under `target/debug/` on the Windows CI leg. On non-Windows
-//! it compiles to an inert stub so the workspace builds everywhere.
-
-#![deny(unsafe_code)]
-
-#[cfg(windows)]
-mod imp {
-    use std::io::{Read, Write};
-
-    use windows_sys::Win32::Foundation::HANDLE;
-    use windows_sys::Win32::System::Console::{
-        GetConsoleMode, GetStdHandle, STD_INPUT_HANDLE, SetConsoleMode,
-    };
-
-    /// `ENABLE_VIRTUAL_TERMINAL_INPUT` (0x0200): deliver VT sequences — including
-    /// the terminal's DSR reply — as raw input bytes instead of cooked key events.
-    const ENABLE_VIRTUAL_TERMINAL_INPUT: u32 = 0x0200;
-
-    pub(crate) fn run() -> i32 {
-        // SAFETY: standard Win32 console calls on our own process's stdin
-        // handle; the returned handle is borrowed, never closed here.
-        #[allow(unsafe_code)]
-        unsafe {
-            let h: HANDLE = GetStdHandle(STD_INPUT_HANDLE);
-            if h.is_null() {
-                return 2;
-            }
-            let mut mode: u32 = 0;
-            if GetConsoleMode(h, &raw mut mode) == 0 {
-                return 2;
-            }
-            // VT input only: no line buffering, no echo of the reply bytes.
-            if SetConsoleMode(h, ENABLE_VIRTUAL_TERMINAL_INPUT) == 0 {
-                return 2;
-            }
-        }
-
-        print!("\x1b[6n");
-        let _ = std::io::stdout().flush();
-
-        let mut buf = [0u8; 64];
-        let mut n = 0usize;
-        let stdin = std::io::stdin();
-        let mut lock = stdin.lock();
-        while n < buf.len() {
-            match lock.read(&mut buf[n..=n]) {
-                Ok(0) | Err(_) => break,
-                Ok(m) => {
-                    n += m;
-                    if buf[n - 1] == b'R' {
-                        break;
-                    }
-                }
-            }
-        }
-        println!("DSR-REPLY:{}", String::from_utf8_lossy(&buf[..n]));
-        0
-    }
-}
-
-fn main() {
-    #[cfg(windows)]
-    std::process::exit(imp::run());
-    #[cfg(not(windows))]
-    {
-        eprintln!("dsr_probe is a Windows-only test probe");
-        std::process::exit(2);
-    }
-}
diff --git a/crates/agent-tui-integration/tests/windows_conpty.rs b/crates/agent-tui-integration/tests/windows_conpty.rs
index 9c1c4ba..54efa82 100644
--- a/crates/agent-tui-integration/tests/windows_conpty.rs
+++ b/crates/agent-tui-integration/tests/windows_conpty.rs
@@ -12,11 +12,11 @@
 //!  1. `run | consumer` in a `cmd.exe` pipeline sees prompt EOF (the `win_spawn`
 //!     handle-inheritance allow-list).
 //!  2. An interactive pane's startup DSR (`ESC[6n`) is answered, unblocking
-//!     the child (engine `take_pty_writes` write-back via `dsr_probe`).
+//!     the child (engine `take_pty_writes` write-back via `conpty_probe`).
 //!  3. `tail --follow` delivers a fast-exiting child's trailing bytes before
 //!     the stream ends (`exit_drained` / `DRAIN_GRACE`).
 //!  4. `die` reaps the whole descendant tree (`taskkill /F /T`).
-//!  5. `signal SIGINT` interrupts via ETX written to the ConPTY input.
+//!  5. `signal SIGINT` delivers ETX to the child's ConPTY input.
 //!  6. `signal SIGBREAK` is rejected with an actionable error (no false
 //!     success).
 
@@ -188,19 +188,33 @@ where
     }
 }
 
-/// Locate the `dsr_probe` test binary, mirroring `agent_tui_binary`'s
+/// Locate the `conpty_probe` test binary, mirroring `agent_tui_binary`'s
 /// debug-then-release candidate order.
-fn dsr_probe_path() -> Result<PathBuf> {
+fn conpty_probe_path() -> Result<PathBuf> {
     let root = workspace_root()?;
     for c in [
-        root.join("target").join("debug").join("dsr_probe.exe"),
-        root.join("target").join("release").join("dsr_probe.exe"),
+        root.join("target").join("debug").join("conpty_probe.exe"),
+        root.join("target").join("release").join("conpty_probe.exe"),
     ] {
         if c.exists() {
             return Ok(c);
         }
     }
-    bail!("dsr_probe.exe not built under {}/target", root.display())
+    bail!("conpty_probe.exe not built under {}/target", root.display())
+}
+
+/// Extract the `text` field from a `tail --strip-ansi` JSON envelope.
+fn tail_text(out: &std::process::Output) -> Result<String> {
+    let v: serde_json::Value = serde_json::from_slice(&out.stdout).with_context(|| {
+        format!(
+            "parse tail envelope: {}",
+            String::from_utf8_lossy(&out.stdout)
+        )
+    })?;
+    v.pointer("/data/text")
+        .and_then(serde_json::Value::as_str)
+        .map(str::to_owned)
+        .with_context(|| format!("no data.text in tail envelope: {v}"))
 }
 
 /// Count processes whose command line contains `sentinel`, excluding the
@@ -230,13 +244,23 @@ async fn sentinel_process_count(sentinel: &str) -> Result<usize> {
 async fn run_in_cmd_pipeline_consumer_sees_prompt_eof() -> Result<()> {
     let rig = Rig::new("run-pipe-eof")?;
     let bin = agent_tui_binary()?;
-    let script = format!(
-        "\"{}\" --session {} --socket-dir \"{}\" run -- cmd /c echo agent-tui-e2e-eof | findstr agent-tui-e2e-eof",
-        bin.display(),
-        rig.session,
-        rig.socket_dir.display()
-    );
-    let out = tokio::time::timeout(Duration::from_secs(90), rig.cmd_exe(&script))
+    // Write the pipeline to a .cmd file instead of passing it as a `cmd /c`
+    // argument: Rust's CommandLineToArgvW quoting (\" for embedded quotes)
+    // collides with cmd's own outer-quote stripping and produced
+    // '"...agent-tui.exe" is not recognized' on the first CI run. Inside a
+    // batch file cmd parses the line natively, so quotes just work.
+    let script_path = rig.root.join("pipeline.cmd");
+    std::fs::write(
+        &script_path,
+        format!(
+            "\"{}\" --session {} --socket-dir \"{}\" run -- cmd /c echo agent-tui-e2e-eof | findstr agent-tui-e2e-eof\r\n",
+            bin.display(),
+            rig.session,
+            rig.socket_dir.display()
+        ),
+    )?;
+    let script = script_path.to_str().context("script path not utf-8")?;
+    let out = tokio::time::timeout(Duration::from_secs(90), rig.cmd_exe(script))
         .await
         .context("consumer did not see EOF within 90s (handle-inheritance regression?)")??;
     assert!(
@@ -253,15 +277,15 @@ async fn run_in_cmd_pipeline_consumer_sees_prompt_eof() -> Result<()> {
 }
 
 /// The engine must answer the ConPTY startup DSR (`ESC[6n`) or interactive
-/// children block waiting for the cursor-position report. `dsr_probe` is the
-/// minimal such child: without the `take_pty_writes` write-back its read
-/// blocks forever and the sentinel never appears.
+/// children block waiting for the cursor-position report. `conpty_probe dsr`
+/// is the minimal such child: without the `take_pty_writes` write-back its
+/// read blocks forever and the sentinel never appears.
 #[tokio::test]
 async fn interactive_pane_answers_startup_dsr() -> Result<()> {
     let rig = Rig::new("dsr")?;
-    let probe = dsr_probe_path()?;
+    let probe = conpty_probe_path()?;
     let probe_str = probe.to_str().context("probe path not utf-8")?;
-    let pane = rig.spawn_pane(&[probe_str]).await?;
+    let pane = rig.spawn_pane(&[probe_str, "dsr"]).await?;
     poll_until("DSR reply on pane", Duration::from_secs(30), || async {
         let out = rig
             .output(false, &["tail", "--pane", &pane, "--strip-ansi"])
@@ -272,7 +296,7 @@ async fn interactive_pane_answers_startup_dsr() -> Result<()> {
     let out = rig
         .output(false, &["tail", "--pane", &pane, "--strip-ansi"])
         .await?;
-    let text = String::from_utf8_lossy(&out.stdout);
+    let text = tail_text(&out)?;
     // --strip-ansi removes the reply's ESC, leaving "DSR-REPLY:[<row>;<col>R".
     let reply = text.split("DSR-REPLY:").nth(1).unwrap_or("");
     assert!(
@@ -350,14 +374,27 @@ async fn die_reaps_descendant_tree() -> Result<()> {
     Ok(())
 }
 
-/// `signal SIGINT` writes ETX to the ConPTY input; conhost raises a real
-/// `CTRL_C_EVENT` and `ping -t` exits.
+/// `signal SIGINT` writes ETX (`0x03`) to the ConPTY input — the daemon-side
+/// half of the interrupt contract. Asserted as byte delivery to the child's
+/// stdin (via `conpty_probe read-stdin` in VT-input mode): whether conhost
+/// then translates ETX into a `CTRL_C_EVENT` is conhost-mode- and
+/// Windows-version-dependent (a `ping -t` child did NOT exit on the
+/// windows-latest CI runner, though the author's Windows 11 e2e saw the
+/// interrupt land), so the suite pins the deterministic half.
 #[tokio::test]
-async fn sigint_via_conpty_etx_interrupts_child() -> Result<()> {
+async fn sigint_writes_etx_to_conpty_input() -> Result<()> {
     let rig = Rig::new("sigint")?;
-    let pane = rig.spawn_pane(&["ping", "-t", "127.0.0.1"]).await?;
-    // Give ping a beat to be running before the interrupt lands.
-    tokio::time::sleep(Duration::from_secs(2)).await;
+    let probe = conpty_probe_path()?;
+    let probe_str = probe.to_str().context("probe path not utf-8")?;
+    let pane = rig.spawn_pane(&[probe_str, "read-stdin"]).await?;
+    // Let the probe set its console mode and reach its blocking read.
+    poll_until("probe mode-set", Duration::from_secs(15), || async {
+        let out = rig
+            .output(false, &["tail", "--pane", &pane, "--strip-ansi"])
+            .await?;
+        Ok(String::from_utf8_lossy(&out.stdout).contains("MODE-SET:ok"))
+    })
+    .await?;
     let out = rig
         .output(false, &["signal", "--pane", &pane, "SIGINT"])
         .await?;
@@ -366,16 +403,25 @@ async fn sigint_via_conpty_etx_interrupts_child() -> Result<()> {
         "signal SIGINT failed: {}",
         String::from_utf8_lossy(&out.stderr)
     );
+    poll_until(
+        "ETX byte on child stdin",
+        Duration::from_secs(15),
+        || async {
+            let out = rig
+                .output(false, &["tail", "--pane", &pane, "--strip-ansi"])
+                .await?;
+            Ok(String::from_utf8_lossy(&out.stdout).contains("STDIN-HEX:"))
+        },
+    )
+    .await?;
     let out = rig
-        .output(
-            false,
-            &["wait", "--pane", &pane, "--exit", "--max", "15000"],
-        )
+        .output(false, &["tail", "--pane", &pane, "--strip-ansi"])
         .await?;
+    let text = tail_text(&out)?;
+    let hex = text.split("STDIN-HEX:").nth(1).unwrap_or("");
     assert!(
-        out.status.success(),
-        "child did not exit after SIGINT: {}",
-        String::from_utf8_lossy(&out.stderr)
+        hex.trim_end().ends_with("03"),
+        "ETX (0x03) not delivered to child stdin: {text}"
     );
     rig.die(&pane).await;
     Ok(())

From 04d51194f21951433827941e30fc02375cd42364 Mon Sep 17 00:00:00 2001
From: Paul Querna <paul.querna@conductorone.com>
Date: Wed, 29 Jul 2026 17:48:02 +0000
Subject: [PATCH 3/3] =?UTF-8?q?test(windows):=20print=20DSR=20reply=20as?=
 =?UTF-8?q?=20hex=20=E2=80=94=20it=20was=20arriving=20all=20along?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

CI diagnostics showed the probe DID receive the cursor-position report
(DSR-READ:n=6:status=stop-byte, i.e. ESC[1;1R) — but printing it raw
into the pane means the pane's own terminal parser swallows the escape
sequence, so the stripped text showed an empty reply. Print
DSR-REPLY-HEX instead and assert the decoded shape
(ESC [ <digits> ; <digits> R).

Co-authored-by: c1-squire-dev[bot] <c1-squire-dev[bot]@users.noreply.github.com>
---
 .../src/bin/conpty_probe.rs                   | 10 ++++++-
 .../tests/windows_conpty.rs                   | 27 +++++++++++++++----
 2 files changed, 31 insertions(+), 6 deletions(-)

diff --git a/crates/agent-tui-integration/src/bin/conpty_probe.rs b/crates/agent-tui-integration/src/bin/conpty_probe.rs
index e59c39e..264c929 100644
--- a/crates/agent-tui-integration/src/bin/conpty_probe.rs
+++ b/crates/agent-tui-integration/src/bin/conpty_probe.rs
@@ -94,7 +94,15 @@ mod imp {
         let mut buf = [0u8; 64];
         let (n, status) = read_until(&mut buf, b'R');
         println!("DSR-READ:n={n}:status={status}");
-        println!("DSR-REPLY:{}", String::from_utf8_lossy(&buf[..n]));
+        // Print the reply as HEX, not raw: the reply bytes are themselves a VT
+        // sequence (ESC[<row>;<col>R) and the pane's own terminal parser
+        // swallows them — which is exactly what hid a *successful* reply on
+        // the first CI iteration (n=6, stop-byte, empty text).
+        let mut hex = String::with_capacity(n * 2);
+        for b in &buf[..n] {
+            let _ = write!(hex, "{b:02x}");
+        }
+        println!("DSR-REPLY-HEX:{hex}");
         0
     }
 
diff --git a/crates/agent-tui-integration/tests/windows_conpty.rs b/crates/agent-tui-integration/tests/windows_conpty.rs
index 54efa82..00f8d4d 100644
--- a/crates/agent-tui-integration/tests/windows_conpty.rs
+++ b/crates/agent-tui-integration/tests/windows_conpty.rs
@@ -290,23 +290,40 @@ async fn interactive_pane_answers_startup_dsr() -> Result<()> {
         let out = rig
             .output(false, &["tail", "--pane", &pane, "--strip-ansi"])
             .await?;
-        Ok(String::from_utf8_lossy(&out.stdout).contains("DSR-REPLY:"))
+        Ok(String::from_utf8_lossy(&out.stdout).contains("DSR-REPLY-HEX:"))
     })
     .await?;
     let out = rig
         .output(false, &["tail", "--pane", &pane, "--strip-ansi"])
         .await?;
     let text = tail_text(&out)?;
-    // --strip-ansi removes the reply's ESC, leaving "DSR-REPLY:[<row>;<col>R".
-    let reply = text.split("DSR-REPLY:").nth(1).unwrap_or("");
+    let hex = text.split("DSR-REPLY-HEX:").nth(1).unwrap_or("").trim_end();
+    let reply = decode_hex(hex).with_context(|| format!("bad reply hex in: {text}"))?;
     assert!(
-        reply.trim_end().starts_with('[') && reply.trim_end().ends_with('R'),
-        "reply is not a cursor-position report: {text}"
+        reply.len() >= 6
+            && reply[0] == 0x1b
+            && reply[1] == b'['
+            && reply[reply.len() - 1] == b'R'
+            && reply[2..reply.len() - 1]
+                .iter()
+                .all(|b| b.is_ascii_digit() || *b == b';'),
+        "reply is not a cursor-position report (ESC[<row>;<col>R): {text}"
     );
     rig.die(&pane).await;
     Ok(())
 }
 
+/// Strict hex decode (even length, lowercase pairs as the probe emits).
+fn decode_hex(s: &str) -> Result<Vec<u8>> {
+    if s.len() % 2 != 0 {
+        bail!("odd-length hex: {s:?}");
+    }
+    (0..s.len())
+        .step_by(2)
+        .map(|i| u8::from_str_radix(&s[i..i + 2], 16).map_err(Into::into))
+        .collect()
+}
+
 /// A fast child that writes-then-exits must have its trailing bytes delivered
 /// before the follow stream's terminal gate (`exit_drained` / `DRAIN_GRACE`)
 /// ends the stream — and the stream must end (the ConPTY EOF fix).