fix(itmux): address PR #252 review (timeout validation + validator soundness)

Fix 1 (codex must-fix): --timeout took an unvalidated f64, so --timeout=-1 /
=0 / =NaN / =inf reached Duration::from_secs_f64 and panicked before docker.
Add a clap value_parser (parse_positive_timeout) rejecting non-finite and
non-positive values with a clean CLI error (exit 2, no panic). Unit tests
cover -1/0/NaN/inf/garbage -> error and 2.5 -> Some(2.5).

Fix 2 (codex must-fix): the event-contract validator dropped blank stdout
lines at collection, so a blank line in --json mode slipped past the purity
check. parse_lines no longer filters blank lines; a blank/whitespace-only line
surfaces as NonJson and the purity rule rejects it. Negative test added.

Fix 3 (codex must-fix): the validator counted exactly one session_end but did
not enforce ordering, so session_end -> tool_start -> result (OnStream) and
lifecycle events after session_end (ResultFile) wrongly passed. Enforce
session_end as the terminal lifecycle event: only the OnStream result line may
follow it, nothing else, in both modes. Negative tests added for both modes.

Nits: removed the dead StreamedRun.exit_ok field (+ its allow(dead_code));
documented the E3 lexicographic-ts assumption (emitter always uses fixed-width
second-precision UTC Z timestamps) rather than adding a datetime dependency.

Claude-Session: https://claude.ai/code/session_019TtekhJft4s2qBGRGgxToG

Author: NeuralEmpowerment

providers/workspaces/interactive-tmux/driver-rs/src/main.rs

@@ -127,12 +127,32 @@ enum Cmd {
         /// Wall-clock timeout, in seconds, for the whole run. Maps to
         /// `AgentRunSpec.limits.timeout_s`. When omitted, the orchestrator's
         /// default await bound applies (behaviour unchanged from before this
-        /// flag existed).
-        #[arg(long)]
+        /// flag existed). Must be a finite, strictly-positive number - a
+        /// non-finite or non-positive value is rejected with a clean CLI error
+        /// (never a downstream `Duration::from_secs_f64` panic).
+        #[arg(long, value_parser = parse_positive_timeout)]
         timeout: Option<f64>,
     },
 }
 
+/// Clap value parser for `--timeout`: accept only a finite, strictly-positive
+/// number of seconds. Rejects `<= 0.0` (an instant/zero timeout is meaningless)
+/// and non-finite values (`NaN`, `inf`) with a message clap renders as a clean
+/// CLI error - this is what keeps `Duration::from_secs_f64` from ever seeing a
+/// value that would panic.
+fn parse_positive_timeout(raw: &str) -> Result<f64, String> {
+    let value: f64 = raw
+        .parse()
+        .map_err(|_| format!("'{raw}' is not a number"))?;
+    if !value.is_finite() {
+        return Err(format!("timeout must be a finite number, got '{raw}'"));
+    }
+    if value <= 0.0 {
+        return Err(format!("timeout must be greater than 0, got '{raw}'"));
+    }
+    Ok(value)
+}
+
 /// Build the `AgentRunSpec` for `itmux run` from the CLI inputs. Pure so the
 /// `--timeout -> limits.timeout_s` mapping is unit-testable without spawning a
 /// run. When `timeout` is `None`, `limits` stays `None` so the default await
@@ -647,6 +667,23 @@ mod tests {
         assert_eq!(limits.token_budget, None);
     }
 
+    #[test]
+    fn parse_positive_timeout_accepts_a_finite_positive_value() {
+        assert_eq!(parse_positive_timeout("2.5"), Ok(2.5));
+    }
+
+    #[test]
+    fn parse_positive_timeout_rejects_non_positive_and_non_finite() {
+        // Each of these would reach Duration::from_secs_f64 and panic if it
+        // slipped through - the parser must reject them cleanly instead.
+        for bad in ["-1", "0", "NaN", "inf", "-inf", "not-a-number"] {
+            assert!(
+                parse_positive_timeout(bad).is_err(),
+                "expected {bad:?} to be rejected"
+            );
+        }
+    }
+
     #[test]
     fn build_run_spec_without_timeout_leaves_limits_none() {
         // Omitting --timeout must not change existing behaviour: no limits, so

providers/workspaces/interactive-tmux/driver-rs/tests/live_eval.rs

@@ -96,12 +96,18 @@ enum ParsedLine {
     NonJson(String),
 }
 
-/// Parse a stdout blob into per-line results. Blank lines are ignored (they are
-/// not stream content); every other line must parse as an [`AgentRunEvent`].
+/// Parse a stdout blob into per-line results, ONE per line, dropping nothing.
+///
+/// In `--json` mode stdout is pure event JSONL: every line must parse as an
+/// [`AgentRunEvent`], and a blank/whitespace-only line is itself a purity
+/// violation (it is not valid event JSONL), so it is surfaced as a
+/// [`ParsedLine::NonJson`] for [`validate_event_contract`] to reject rather than
+/// silently swallowed. (`str::lines()` already omits a single trailing newline,
+/// so a normal `println!`-terminated stream does not produce a spurious blank
+/// final line.)
 fn parse_lines(stdout: &str) -> Vec<ParsedLine> {
     stdout
         .lines()
-        .filter(|line| !line.trim().is_empty())
         .map(|line| match serde_json::from_str::<AgentRunEvent>(line) {
             Ok(event) => ParsedLine::Event(Box::new(event)),
             Err(_) => ParsedLine::NonJson(line.to_string()),
@@ -152,7 +158,11 @@ enum ResultDelivery {
 /// * zero non-JSON lines on stdout (`--json` mode is pure event JSONL);
 /// * `seq` monotonic from 0 with no gaps;
 /// * `run_id` identical across every line;
-/// * exactly one `session_end`;
+/// * exactly one `session_end`, and it is the LAST lifecycle event - no
+///   `tool_start`/`tool_end`/`token_usage`/further `session_end` may follow it
+///   (in BOTH modes). In `OnStream` mode the terminal `type:"result"` line is
+///   the only event allowed after `session_end`; in `ResultFile` mode nothing
+///   may follow it;
 /// * the final result is delivered via the expected channel: exactly one
 ///   `type:"result"` line for [`ResultDelivery::OnStream`], and ZERO for
 ///   [`ResultDelivery::ResultFile`] (stdout stays pure lifecycle events).
@@ -203,6 +213,25 @@ fn validate_event_contract(parsed: &[ParsedLine], delivery: ResultDelivery) -> R
         ));
     }
 
+    // session_end is the TERMINAL lifecycle event: the orchestrator emits it
+    // last, immediately before the (optional) terminal result. So nothing but a
+    // `result` line may appear after it - any lifecycle event
+    // (tool_start/tool_end/token_usage/another session_end) following
+    // session_end is a contract violation in BOTH modes. (`ResultFile` also
+    // forbids result lines entirely, below, so there it means NOTHING follows.)
+    let session_end_index = events
+        .iter()
+        .position(|event| matches!(event.payload, AgentRunEventPayload::SessionEnd { .. }))
+        .expect("exactly one session_end confirmed above");
+    for (offset, event) in events.iter().enumerate().skip(session_end_index + 1) {
+        if !matches!(event.payload, AgentRunEventPayload::Result { .. }) {
+            return Err(format!(
+                "session_end is not terminal: a non-result event (seq {}, line {offset}) follows it",
+                event.seq
+            ));
+        }
+    }
+
     // Final-result delivery channel.
     let result_lines = events
         .iter()
@@ -329,13 +358,12 @@ impl Signaler {
     }
 }
 
-/// A streamed run: every stdout line tagged with its arrival [`Instant`], plus
-/// the terminal exit status. Used by E3 (arrival spread) and E4 (signal at a
-/// stream boundary).
+/// A streamed run: every stdout line tagged with its arrival [`Instant`]. Used
+/// by E3 (arrival spread) and E4 (signal at a stream boundary). The terminal
+/// verdict is read from the parsed result, not the process exit code, so no
+/// exit status is retained here.
 struct StreamedRun {
     lines: Vec<(Instant, String)>,
-    #[allow(dead_code)]
-    exit_ok: bool,
 }
 
 impl StreamedRun {
@@ -387,11 +415,10 @@ fn stream_run(
         lines.push((at, line));
     }
     let _ = reader.join();
-    let status = child.wait()?;
-    Ok(StreamedRun {
-        lines,
-        exit_ok: status.success(),
-    })
+    // Reap the child so it does not linger as a zombie; the terminal verdict is
+    // taken from the parsed result stream, not this exit status.
+    let _ = child.wait()?;
+    Ok(StreamedRun { lines })
 }
 
 /// Is a line a `tool_end submit` or `tool_start await` event? That is the
@@ -578,6 +605,43 @@ mod plumbing_tests {
         assert!(err.contains("non-JSON"), "err: {err}");
     }
 
+    #[test]
+    fn validator_rejects_an_embedded_blank_line() {
+        // A blank line on stdout in --json mode is a purity violation: it is
+        // not valid event JSONL and must NOT be silently dropped. Inject one
+        // between two otherwise-valid events.
+        let sample = good_sample();
+        let (head, tail) = sample.split_once('\n').expect("multi-line sample");
+        let corrupted = format!("{head}\n\n{tail}");
+        let parsed = parse_lines(&corrupted);
+        let err = validate_event_contract(&parsed, ResultDelivery::OnStream).unwrap_err();
+        assert!(err.contains("non-JSON"), "err: {err}");
+    }
+
+    #[test]
+    fn validator_rejects_a_lifecycle_event_after_session_end_on_stream() {
+        // session_end must be terminal: a tool_start AFTER it (before the
+        // result line) is rejected even though counts/seq are otherwise valid.
+        let stream = "\
+{\"run_id\":\"r\",\"seq\":0,\"ts\":\"t\",\"type\":\"session_end\",\"outcome\":{\"success\":true,\"summary\":\"ok\"}}
+{\"run_id\":\"r\",\"seq\":1,\"ts\":\"t\",\"type\":\"tool_start\",\"tool_name\":\"provision\",\"tool_input\":null}
+{\"run_id\":\"r\",\"seq\":2,\"ts\":\"t\",\"type\":\"result\",\"result\":{\"result\":{\"success\":true,\"summary\":\"ok\"},\"output_artifacts\":[],\"session_log\":\"\",\"observability\":null}}";
+        let parsed = parse_lines(stream);
+        let err = validate_event_contract(&parsed, ResultDelivery::OnStream).unwrap_err();
+        assert!(err.contains("session_end is not terminal"), "err: {err}");
+    }
+
+    #[test]
+    fn validator_rejects_a_lifecycle_event_after_session_end_with_result_file() {
+        // Same rule in --result-file mode: nothing may follow session_end.
+        let stream = "\
+{\"run_id\":\"r\",\"seq\":0,\"ts\":\"t\",\"type\":\"session_end\",\"outcome\":{\"success\":true,\"summary\":\"ok\"}}
+{\"run_id\":\"r\",\"seq\":1,\"ts\":\"t\",\"type\":\"tool_end\",\"tool_name\":\"capture\",\"success\":true,\"output_summary\":null}";
+        let parsed = parse_lines(stream);
+        let err = validate_event_contract(&parsed, ResultDelivery::ResultFile).unwrap_err();
+        assert!(err.contains("session_end is not terminal"), "err: {err}");
+    }
+
     #[test]
     fn validator_rejects_a_seq_gap() {
         // Drop the seq=1 line so the sequence jumps 0 -> 2.
@@ -831,6 +895,12 @@ fn e3_live_events_stream_incrementally() {
         first_tool_start_ts.expect("a tool_start"),
         session_end_ts.expect("a session_end"),
     );
+    // Lexicographic compare is sound here because the emitter
+    // (`workspace_executor::now_rfc3339`) always produces fixed-width,
+    // second-precision UTC timestamps with a literal `Z` offset
+    // (`YYYY-MM-DDThh:mm:ssZ`), so string order equals chronological order. The
+    // crate carries no datetime parser (it hand-rolls RFC3339), and adding one
+    // solely for this assertion is not worth the dependency weight.
     assert!(
         ts_start <= ts_end,
         "tool_start {ts_start} must be <= session_end {ts_end}"