Preserve raw worker stdin at the server boundary (#72)

* Add Zod readline discard recovery test

Teach the standalone Zod worker to emit readline_discard for buffered stdin after an interrupt so stale active-turn input is accounted for before a follow-up tail runs.

Validation: cargo check; cargo build; python3 tests/run_integration_tests.py --binary target/debug/mcp-repl; cargo clippy --all-targets --all-features -- -D warnings; cargo test --quiet; cargo +nightly fmt.

* Validate protocol session_end reasons

Reject worker sideband session_end frames that carry unrecognized reasons, and validate optional session_end message_b64 payloads before accepting the frame as final.

Add a Zod test hook and MCP-level regression coverage so malformed protocol-worker session_end data fails fast instead of being treated as a clean session shutdown.

Validation: cargo test --quiet --test zod_protocol zod_worker_invalid_session_end_reason_is_protocol_error; cargo test --quiet --test zod_protocol; cargo check; cargo build; python3 tests/run_integration_tests.py --binary target/debug/mcp-repl; cargo clippy --all-targets --all-features -- -D warnings; cargo test --quiet; cargo +nightly fmt

* Add Zod shutdown conformance hooks

Add slow-shutdown and hang-shutdown commands to the standalone Zod worker so public protocol tests can exercise delayed graceful shutdown and pending shutdown states. Cover both through MCP-level tests and update the active protocol plan command list.

Validation:
- cargo test --quiet --test zod_protocol shutdown (red before fixture change)
- cargo test --quiet --test zod_protocol
- cargo check
- cargo build
- python3 tests/run_integration_tests.py --binary target/debug/mcp-repl
- cargo clippy --all-targets --all-features -- -D warnings
- cargo test --quiet
- cargo +nightly fmt

* Split Zod interrupt observations

Add an interrupt-report fixture command that records sideband interrupt
notifications separately from OS interrupt delivery. Cover it through the
public Zod protocol test and document the command in the active protocol plan.

Validation:
- cargo +nightly fmt
- cargo check
- cargo build
- cargo test --test zod_protocol zod_worker_reports_sideband_and_os_interrupt_facts -- --nocapture
- cargo test --test zod_protocol --quiet
- python3 tests/run_integration_tests.py --binary target/debug/mcp-repl
- cargo clippy --all-targets --all-features -- -D warnings
- cargo test --quiet

* Preserve control-tail newline bytes

Consume only the leading Ctrl-C/Ctrl-D byte when splitting control-prefixed input so an immediate newline remains part of the tail payload. Add a public Zod MCP regression that proves the worker observes that newline before the tail command.

Update existing tests that intend bare controls to send unterminated control bytes, and update transcript snapshots to show those bare inputs explicitly.

Validation:
- cargo check
- cargo build
- python3 tests/run_integration_tests.py --binary target/debug/mcp-repl
- cargo clippy --all-targets --all-features -- -D warnings
- cargo test --quiet
- cargo +nightly fmt

* Preserve protocol worker stdin bytes

Move CR/CRLF normalization out of the shared request path and keep it in the built-in R/Python drivers, so custom protocol workers receive the exact client stdin bytes plus the existing single appended LF rule.

Add public Zod MCP coverage for supplied CRLF bytes and a trailing bare carriage return. Update the protocol plan to state the byte-preservation contract.

Validation:
- cargo test --test zod_protocol zod_worker_preserves_crlf_stdin_and_appended_newline -- --exact
- cargo test --test write_stdin_edge_cases write_stdin_accepts_crlf_input -- --exact
- cargo check
- cargo build
- python3 tests/run_integration_tests.py --binary target/debug/mcp-repl
- cargo clippy --all-targets --all-features -- -D warnings
- cargo test --quiet
- cargo +nightly fmt

* Fix reset teardown output and Ctrl-C bundle reuse

Snapshot pending output before reset shutdown so teardown-only worker output
does not leak into the reset reply. Also classify Ctrl-C plus newline/CRLF
as follow-up input for timeout bundle reuse; only bare Ctrl-C reuses the
full timeout reply.

Validation:
- cargo check
- cargo build
- python3 tests/run_integration_tests.py --binary target/debug/mcp-repl
- cargo clippy --all-targets --all-features -- -D warnings
- cargo test --quiet
- cargo +nightly fmt

* Cover reset EOF for active Zod stdin

Add a Zod command that blocks on a second stdin read and verify repl_reset
closes the active stdin stream without sending interpreter shutdown text.

Validation:
- cargo check
- cargo build
- python3 tests/run_integration_tests.py --binary target/debug/mcp-repl
- cargo clippy --all-targets --all-features -- -D warnings
- cargo test --quiet
- cargo +nightly fmt

* Fix Unix Python CRLF stdin normalization

Normalize CRLF and bare CR input before the built-in Unix Python backend
writes request bytes to its PTY. Keep custom protocol worker payloads
byte-preserving by applying the normalization only when the protocol driver is
serving the built-in Python bridge.

Add a Unix public regression test that sends a CRLF-formatted multiline Python
block through the MCP repl tool and asserts it executes without an injected
blank line or IndentationError.

Review finding:
- [P2] Normalize CRLF before Unix Python PTY writes — /Users/tomasz/github/posit-dev/mcp-repl/src/worker_process.rs:2624-2624
  On Unix, the built-in Python backend uses `ProtocolBackendDriver`, whose `prepare_input_payload` now preserves stdin bytes. Since the previous normalization was removed before this call, CRLF input is written to the PTY as `\r\n`; the terminal maps `\r` to another newline, so a block like `if True:\r\n    print("A")` reaches Python with a blank line and raises `IndentationError`. Custom protocol workers can preserve bytes, but built-in Python still needs newline normalization before writing to the PTY.

Response:
- Added `python_crlf_multiline_block_executes` to cover CRLF multiline Python input through the public MCP tool path.
- Updated `ProtocolBackendDriver::prepare_input_payload` so only the built-in Unix Python bridge normalizes CRLF/CR before PTY writes; custom protocol workers still preserve stdin bytes.

Validation:
- cargo test --test python_backend python_crlf_multiline_block_executes -- --nocapture (failed before fix, passed after fix)
- cargo check
- cargo build
- python3 tests/run_integration_tests.py --binary target/debug/mcp-repl
- cargo clippy --all-targets --all-features -- -D warnings
- cargo test --quiet
- cargo +nightly fmt

* Fix Zod command fixture argument count

Move the shutdown-log path into CommandState so run_command stays within the clippy argument limit after the stdin-close reset coverage was adapted.

Validation:
- cargo test --test zod_protocol --quiet
- cargo test --test python_backend --quiet crlf
- cargo test --quiet timeout_bundle_reuse_treats_newline_ctrl_c_as_follow_up_input
- cargo check
- cargo build
- python3 tests/run_integration_tests.py --binary target/debug/mcp-repl
- cargo clippy --all-targets --all-features -- -D warnings
- cargo test --quiet
- cargo +nightly fmt

* Preserve worker stdin bytes at server boundary

Keep server-side input payload preparation limited to the MCP UTF-8 string
boundary and the explicit final-newline append rule. Backend/runtime-specific
CRLF interpretation belongs to the worker instead of the server.

This also removes the Unix Python CRLF multiline expectation, keeps CRLF
acceptance scoped to Windows, and rewords the Zod protocol regression around
raw client byte preservation.

Validation:
- cargo check
- cargo build
- python3 tests/run_integration_tests.py --binary target/debug/mcp-repl
- cargo clippy --all-targets --all-features -- -D warnings
- cargo test --quiet
- cargo +nightly fmt

* Fix CI stdin edge cases

Normalize CRLF input in the built-in R backend before writing bytes to R so
Windows requests do not leave carriage returns in R source. Keep protocol
worker byte preservation intact by leaving the protocol driver unchanged.

Make the files-mode timeout poll regression deterministic by replacing a
short sleep with an explicit file gate before releasing the request.

Validation:
- cargo check
- cargo build
- python3 tests/run_integration_tests.py --binary target/debug/mcp-repl
- cargo clippy --all-targets --all-features -- -D warnings
- cargo test --quiet
- cargo +nightly fmt

Author: t-kalinowski

docs/plans/active/worker-server-protocol-zod.md

@@ -664,7 +664,9 @@ emits an unsatisfied `readline_start`, the server uses that explicit
 event to finalize the poll reply.
 
 A later non-empty `repl()` call after an unsatisfied `readline_start` is
-written to worker stdin after the same trailing-newline normalization.
+written to worker stdin after the same trailing-newline rule: if the input is
+non-empty and does not end in `\n`, append one `\n`. The server does not
+otherwise canonicalize supplied stdin bytes such as `\r\n` or bare `\r`.
 The worker/runtime decides whether those bytes answer an
 interpreter-level prompt, continue an incomplete expression, or start a
 new top-level evaluation.
@@ -751,8 +753,14 @@ a small deterministic command language through stdin:
   bytes.
 - `sleep <millis>` delays the next unsatisfied prompt so timeout and
   poll behavior can be tested.
-- `interruptible <millis>` delays until either the timer finishes or an
-  OS interrupt arrives.
+- `interruptible <millis>` delays until either the timer finishes or a
+  sideband or OS interrupt arrives.
+- `interrupt-report <millis>` delays while recording sideband and OS
+  interrupt delivery as separate output facts.
+- `slow-shutdown <millis>` delays a later `exit` or EOF `session_end`
+  so reset and shutdown graceful-exit timing can be tested.
+- `hang-shutdown` accepts a later `exit` or EOF but never emits
+  `session_end`, so reset and shutdown OS escalation can be tested.
 - `image` emits a tiny deterministic PNG if Zod advertises image
   support.
 - `exit` emits `session_end`.

src/ipc.rs

@@ -461,6 +461,14 @@ impl ServerIpcConnection {
                         reason,
                         message_b64,
                     } => {
+                        if let Err(err) =
+                            validate_session_end(reason.as_deref(), message_b64.as_deref())
+                        {
+                            let mut guard = reader_inbox.lock().unwrap();
+                            latch_protocol_error(&mut guard, err);
+                            reader_cvar.notify_all();
+                            break;
+                        }
                         let mut guard = reader_inbox.lock().unwrap();
                         guard.session_end = true;
                         guard.session_end_final = true;
@@ -1907,6 +1915,23 @@ fn latch_protocol_error(guard: &mut ServerIpcInbox, message: impl Into<String>)
     });
 }
 
+fn validate_session_end(reason: Option<&str>, message_b64: Option<&str>) -> Result<(), String> {
+    if let Some(reason) = reason {
+        match reason {
+            "shutdown" | "reset" | "runtime_exit" | "crash" | "protocol_error" => {}
+            other => return Err(format!("invalid session_end reason: {other}")),
+        }
+    }
+    if let Some(message_b64) = message_b64
+        && base64::engine::general_purpose::STANDARD
+            .decode(message_b64)
+            .is_err()
+    {
+        return Err("invalid session_end message_b64 base64".to_string());
+    }
+    Ok(())
+}
+
 fn take_latched_protocol_error(guard: &mut ServerIpcInbox) -> Option<String> {
     guard.protocol_error.take().map(|error| error.message)
 }

src/server/response.rs

@@ -186,15 +186,6 @@ pub(crate) fn timeout_bundle_reuse_for_input(input: &str) -> TimeoutBundleReuse
         return TimeoutBundleReuse::FullReply;
     };
     let tail = &input[first.len_utf8()..];
-    let tail = if let Some(rest) = tail.strip_prefix("\r\n") {
-        rest
-    } else if let Some(rest) = tail.strip_prefix('\n') {
-        rest
-    } else if let Some(rest) = tail.strip_prefix('\r') {
-        rest
-    } else {
-        tail
-    };
 
     match first {
         '\u{3}' if tail.is_empty() => TimeoutBundleReuse::FullReply,

src/server/tests.rs

@@ -197,6 +197,22 @@ fn timeout_bundle_reuse_treats_blank_lines_as_fresh_input() {
     ));
 }
 
+#[test]
+fn timeout_bundle_reuse_treats_newline_ctrl_c_as_follow_up_input() {
+    assert!(matches!(
+        super::response::timeout_bundle_reuse_for_input("\u{3}"),
+        super::response::TimeoutBundleReuse::FullReply
+    ));
+    assert!(matches!(
+        super::response::timeout_bundle_reuse_for_input("\u{3}\n"),
+        super::response::TimeoutBundleReuse::FollowUpInput
+    ));
+    assert!(matches!(
+        super::response::timeout_bundle_reuse_for_input("\u{3}\r\n"),
+        super::response::TimeoutBundleReuse::FollowUpInput
+    ));
+}
+
 fn split_lines(text: &str) -> Vec<String> {
     if text.is_empty() {
         return Vec::new();

src/worker_process.rs

@@ -475,7 +475,8 @@ fn driver_refresh_worker_ready(
 
 impl BackendDriver for RBackendDriver {
     fn prepare_input_payload(&self, text: &str) -> Vec<u8> {
-        let mut payload = text.as_bytes().to_vec();
+        let normalized = normalize_input_newlines(text);
+        let mut payload = normalized.into_bytes();
         if !payload.is_empty() && !payload.ends_with(b"\n") {
             payload.push(b'\n');
         }
@@ -1234,17 +1235,7 @@ pub(crate) fn split_write_stdin_control_prefix(
         _ => return None,
     };
 
-    let tail = &input[first.len_utf8()..];
-    let tail = if let Some(rest) = tail.strip_prefix("\r\n") {
-        rest
-    } else if let Some(rest) = tail.strip_prefix('\n') {
-        rest
-    } else if let Some(rest) = tail.strip_prefix('\r') {
-        rest
-    } else {
-        tail
-    };
-    Some((action, tail))
+    Some((action, &input[first.len_utf8()..]))
 }
 
 fn worker_context_event_payload(
@@ -2575,7 +2566,6 @@ impl WorkerManager {
         worker_timeout: Duration,
         server_timeout: Duration,
     ) -> Result<RequestState, WorkerError> {
-        let text = normalize_input_newlines(&text);
         let started_at = std::time::Instant::now();
         let prompt = self.current_prompt_hint();
         self.remember_prompt(prompt);
@@ -3480,13 +3470,23 @@ impl WorkerManager {
                 MISSING_INHERITED_SANDBOX_STATE_MESSAGE.to_string(),
             ));
         }
+        self.maybe_emit_pending_server_notice();
+        let pre_shutdown_output = self
+            .process
+            .is_some()
+            .then(|| self.drain_sealed_formatted_output());
         if let Some(process) = self.process.take() {
             let _ = process.shutdown_graceful(timeout);
+            self.pending_output_tape.clear();
         }
         self.guardrail.busy.store(false, Ordering::Relaxed);
-        self.maybe_emit_pending_server_notice();
 
-        let reply = self.build_session_reset_reply_files("new session started");
+        let reply = match pre_shutdown_output {
+            Some(output) => {
+                self.build_session_reset_reply_files_from_formatted("new session started", output)
+            }
+            None => self.build_session_reset_reply_files("new session started"),
+        };
         self.clear_preserved_prefixes();
         self.reset_output_state_files(true);
         self.note_respawn_during_write();
@@ -3653,14 +3653,32 @@ impl WorkerManager {
                 MISSING_INHERITED_SANDBOX_STATE_MESSAGE.to_string(),
             ));
         }
+        self.maybe_emit_pending_server_notice();
+        let pre_shutdown_end_offset = self
+            .process
+            .is_some()
+            .then(|| self.output.end_offset().unwrap_or(0));
         if let Some(process) = self.process.take() {
             let _ = process.shutdown_graceful(timeout);
         }
+        let post_shutdown_end_offset = self.output.end_offset();
         self.guardrail.busy.store(false, Ordering::Relaxed);
-        self.maybe_emit_pending_server_notice();
 
         let page_bytes = pager::resolve_page_bytes(None);
-        let reply = self.build_session_reset_reply_pager(page_bytes, "new session started");
+        let reply = match pre_shutdown_end_offset {
+            Some(end_offset) => {
+                let reply = self.build_session_reset_reply_pager_to_offset(
+                    page_bytes,
+                    "new session started",
+                    end_offset,
+                );
+                if let Some(end_offset) = post_shutdown_end_offset {
+                    self.output.advance_offset_to(end_offset);
+                }
+                reply
+            }
+            None => self.build_session_reset_reply_pager(page_bytes, "new session started"),
+        };
         self.clear_preserved_prefixes();
         self.reset_output_state_pager(true, false);
         self.note_respawn_during_write();
@@ -4586,10 +4604,19 @@ impl WorkerManager {
     }
 
     fn build_session_reset_reply_files(&mut self, meta: &str) -> ReplyWithOffset {
+        let formatted = self.drain_sealed_formatted_output();
+        self.build_session_reset_reply_files_from_formatted(meta, formatted)
+    }
+
+    fn build_session_reset_reply_files_from_formatted(
+        &mut self,
+        meta: &str,
+        formatted: FormattedPendingOutput,
+    ) -> ReplyWithOffset {
         let FormattedPendingOutput {
             mut contents,
             saw_stderr,
-        } = self.drain_sealed_formatted_output();
+        } = formatted;
         contents.retain(|content| match content {
             WorkerContent::ContentText { text, .. } => !text.trim().is_empty(),
             _ => true,
@@ -4613,6 +4640,15 @@ impl WorkerManager {
 
     fn build_session_reset_reply_pager(&mut self, page_bytes: u64, meta: &str) -> ReplyWithOffset {
         let end_offset = self.output.end_offset().unwrap_or(0);
+        self.build_session_reset_reply_pager_to_offset(page_bytes, meta, end_offset)
+    }
+
+    fn build_session_reset_reply_pager_to_offset(
+        &mut self,
+        page_bytes: u64,
+        meta: &str,
+        end_offset: u64,
+    ) -> ReplyWithOffset {
         let mut is_error = false;
 
         let SnapshotWithImages {
@@ -7603,11 +7639,11 @@ mod tests {
     }
 
     #[test]
-    fn control_prefix_strips_single_separator_newline() {
+    fn control_prefix_preserves_immediate_newline_tail() {
         let (action, remaining) =
             split_write_stdin_control_prefix("\u{4}\nprint(1)").expect("expected control prefix");
         assert!(matches!(action, WriteStdinControlAction::Restart));
-        assert_eq!(remaining, "print(1)");
+        assert_eq!(remaining, "\nprint(1)");
     }
 
     #[test]

tests/README.md

@@ -25,6 +25,6 @@ snapshot.session("default", mcp_script! {
     write_stdin("x <- 1");
     write_stdin("x <- x + 2");
     write_stdin("x", timeout = 0.2);
-    write_stdin("\u{4}");
+    write_stdin_raw_unterminated("\u{4}");
 }).await?;
 ```

tests/common/mod.rs

@@ -279,6 +279,21 @@ macro_rules! mcp_calls_inner {
         $crate::mcp_calls_inner!($session, $($rest)*);
     }};
 
+    ($session:ident, write_stdin_raw_unterminated($input:expr $(, timeout = $timeout:expr)? ); $($rest:tt)*) => {{
+        let mut args = serde_json::Map::new();
+        args.insert("input".to_string(), serde_json::Value::String($input.to_string()));
+        if let Some(timeout) = $crate::common::normalized_test_timeout($crate::mcp_timeout_opt!($($timeout)?)) {
+            args.insert(
+                "timeout_ms".to_string(),
+                serde_json::json!((timeout * 1000.0).round() as i64),
+            );
+        }
+        $session
+            .call_tool($session.repl_tool_name(), serde_json::Value::Object(args))
+            .await;
+        $crate::mcp_calls_inner!($session, $($rest)*);
+    }};
+
 }
 
 #[macro_export]
@@ -440,7 +455,7 @@ fn compact_json(value: &Value) -> String {
     serde_json::to_string(value).unwrap_or_else(|_| value.to_string())
 }
 
-fn normalized_test_timeout(timeout: Option<f64>) -> Option<f64> {
+pub(crate) fn normalized_test_timeout(timeout: Option<f64>) -> Option<f64> {
     #[cfg(windows)]
     {
         timeout.map(|value| value.min(WINDOWS_TEST_TIMEOUT_CAP_SECS))