fix: shut down JSONRPC cleanly on EOF (Fixes #529) (#559)

Treat frame-boundary stdin EOF as normal server shutdown instead of
repeatedly logging empty headers. Cancel admitted subprocess probes
through ownership cleanup, without waiting indefinitely for blocked
transport I/O.

- Distinguish clean EOF from truncated frames and terminal transport
failures; preserve errors that race shutdown.
- Use bounded FIFO output with a dedicated owned stdout handle, explicit
saturation errors, and documented cancellation semantics.
- Supervise probe admission and cleanup; report cleanup failures or the
three-second cleanup deadline explicitly.
- Close test-client stdin and bound normal exit, forced termination, and
reader teardown.
- Add deterministic transport/output/supervisor coverage and native EOF,
unread-output, broken-pipe, and active-descendant regressions.

Validation: Windows and Linux affected suites and CI-feature native
tests pass; 80 concurrent lifecycle regression cases pass. Workspace
formatting/Clippy and targeted all-target Clippy pass. Independent
Reviewer returned no findings. macOS and hosted quality results remain
pending.

Multi-header framing and envelope validation remain separate work under
#532.

Fixes #529

---------

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: karthiknadig

.github/skills/rust-coding-skill/SKILL.md

@@ -98,3 +98,7 @@ Before every Rust commit, run targeted tests and invoke the `rust-precommit` ski
 ## Learnings
 
 Do not execute freshly written scripts as concurrent Unix subprocess fixtures: spawning can fail with `ETXTBSY` (Text file busy). Prefer an existing interpreter such as `/bin/sh -c` with an inline script, or the existing test executable. Assert the typed runner outcome before checking an optional parsed result, so a spawn failure cannot masquerade as a successful negative parsing or timeout test.
+
+For real-pipe EOF/EPIPE tests, create the pipe inside an isolated test subprocess when other test threads spawn children. Unix `CLOEXEC` closes descriptors at exec, not fork: a concurrent child can temporarily retain a reader, allowing the only write to succeed before the final reader disappears. A readiness handshake alone does not prevent this race. Keep the operation's measured deadline separate from setup, and make an outer fixture deadline cover readiness, waits both before and after forced termination, reader joins, and fallback `Drop` cleanup.
+
+Use a per-worktree Cargo target directory when validating stacked changes so native fixtures cannot execute another worktree's stale binary. On WSL, run timing-sensitive Linux binaries from the native Linux filesystem rather than a Windows mount, where page faults can stall in filesystem RPC. When launching instrumented PET with `env_clear()`, retain `LLVM_PROFILE_FILE` exactly so child coverage reaches the collector instead of an uncollected default profile.

crates/pet-conda/src/conda_info.rs

@@ -107,6 +107,13 @@ impl CondaInfo {
                     None
                 }
             }
+            Err(ProcessError::Cancelled) => {
+                trace!(
+                    "Cancelled Conda probe during process shutdown: {:?}",
+                    executable
+                );
+                None
+            }
             Err(err) => {
                 if !is_missing_default_conda(using_default, &err) {
                     warn!(
@@ -191,6 +198,7 @@ mod tests {
             );
         }
         for error in [
+            ProcessError::Cancelled,
             ProcessError::Spawn(io::Error::from(io::ErrorKind::NotFound)),
             ProcessError::Io(io::Error::from(io::ErrorKind::BrokenPipe)),
             ProcessError::Timeout(Duration::from_secs(15)),

crates/pet-jsonrpc/src/lib.rs

@@ -2,10 +2,12 @@
 // Licensed under the MIT License.
 
 use serde::{Deserialize, Serialize};
-use std::io::{self, Write};
 
+mod output;
 pub mod server;
 
+pub use output::{initialize_output, output_error, shutdown_output};
+
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(untagged)]
 pub enum RequestId {
@@ -24,51 +26,52 @@ impl From<u32> for RequestId {
 #[serde(rename_all = "camelCase")]
 #[derive(Debug)]
 struct AnyMethodMessage<T> {
-    pub jsonrpc: String,
+    pub jsonrpc: &'static str,
     pub method: &'static str,
     pub params: Option<T>,
 }
 
 pub fn send_message<T: serde::Serialize>(method: &'static str, params: Option<T>) {
     let payload = AnyMethodMessage {
-        jsonrpc: "2.0".to_string(),
+        jsonrpc: "2.0",
         method,
         params,
     };
-    let message = serde_json::to_string(&payload).unwrap();
-    print!(
-        "Content-Length: {}\r\nContent-Type: application/vscode-jsonrpc; charset=utf-8\r\n\r\n{}",
-        message.len(),
-        message
-    );
-    let _ = io::stdout().flush();
+    output::send(&payload);
 }
+
 pub fn send_reply<T: serde::Serialize>(id: &RequestId, payload: Option<T>) {
-    let payload = serde_json::json!({
-        "jsonrpc": "2.0",
-        "result": payload,
-        "id": id
+    #[derive(Serialize)]
+    struct Reply<'a, T> {
+        jsonrpc: &'static str,
+        result: Option<T>,
+        id: &'a RequestId,
+    }
+
+    output::send(&Reply {
+        jsonrpc: "2.0",
+        result: payload,
+        id,
     });
-    let message = serde_json::to_string(&payload).unwrap();
-    print!(
-        "Content-Length: {}\r\nContent-Type: application/vscode-jsonrpc; charset=utf-8\r\n\r\n{}",
-        message.len(),
-        message
-    );
-    let _ = io::stdout().flush();
 }
 
 pub fn send_error(id: Option<&RequestId>, code: i32, message: String) {
-    let payload = serde_json::json!({
-        "jsonrpc": "2.0",
-        "error": { "code": code, "message": message },
-        "id": id
+    #[derive(Serialize)]
+    struct ErrorBody {
+        code: i32,
+        message: String,
+    }
+
+    #[derive(Serialize)]
+    struct ErrorReply<'a> {
+        jsonrpc: &'static str,
+        error: ErrorBody,
+        id: Option<&'a RequestId>,
+    }
+
+    output::send(&ErrorReply {
+        jsonrpc: "2.0",
+        error: ErrorBody { code, message },
+        id,
     });
-    let message = serde_json::to_string(&payload).unwrap();
-    print!(
-        "Content-Length: {}\r\nContent-Type: application/vscode-jsonrpc; charset=utf-8\r\n\r\n{}",
-        message.len(),
-        message
-    );
-    let _ = io::stdout().flush();
 }