From 95136e31c26c351be4e675078855da8257ba6ad4 Mon Sep 17 00:00:00 2001 From: schickling-assistant <261620128+schickling-assistant@users.noreply.github.com> Date: Tue, 10 Nov 2026 23:35:05 +0100 Subject: [PATCH] Fix durable event waits and idle output disconnects agent-identity: agent/dotfiles/fractal-product-lead agent-persona: generalist agent-supervisor: unavailable agent-tool: OMP agent-tool-version: 18.12.5 agent-runtime: OMP 18.12.5 agent-model: openai-codex/gpt-6.1-sol --- README.md | 51 +++-- crates/pty/src/cli/events.rs | 225 +++++++++++++--------- crates/pty/src/cli/help.rs | 14 +- crates/pty/tests/cli_events.rs | 223 ++++++++++++++++++++- crates/pty/tests/fixtures/help/events.txt | 17 +- 5 files changed, 416 insertions(+), 114 deletions(-) diff --git a/README.md b/README.md index b99a221b..f1fa5582 100644 --- a/README.md +++ b/README.md @@ -100,19 +100,44 @@ lose the current buffered window. One-shot writers (`pty emit`, metadata, tags, display name, `pty exec`, and gc) retain their existing event-lock behavior, so their records can land before daemon records from up to one window earlier. -`pty events NAME --wait session_exit --timeout SECONDS` checks recorded exit -state before and after subscribing, so an already-exited retained session -returns immediately (exit status 0), even if its event log was removed. -Other event types wait for a new matching event. A missing session returns a -not-found error (exit status 1); a vanished target with no recorded exit also -fails promptly, including a launcher that never started. A session removed -while waiting without an observed matching event returns 1. So does any wait -whose daemon dies without recording an exit, before or after the wait starts; -a named follow ends there with 0. A record no daemon has claimed yet keeps -other waits waiting, since its launcher may still start it. The timeout is an -independent deadline and returns 1 even when no filesystem events arrive. -Options may follow the session reference, and any other argument is an error -(exit status 1) rather than silently dropping the options after it. +For durable completion checks, first read the retained exit events for the +current session generation: + +```sh +pty events --recent --type session_exit --json REF +``` + +If no matching record is returned, use a bounded wait, then perform one final +durable recheck: + +```sh +pty events --wait session_exit -t SECONDS --json REF +pty events --recent --type session_exit --json REF +``` + +An empty `--recent` result is not proof of completion. A vanished session is +not success: require a recorded exit, and inspect its exit code separately +from the observer command's status. + +`--wait TYPE` replays matching retained events from the current generation for +any event type, rather than requiring a new event. It subscribes before the +durable check, rechecks after subscription and before reporting a timeout, +and continues following for a match without a check/subscribe race. +`session_exit` also checks recorded exit metadata, so an already-exited +retained session returns immediately even if its event log was removed. +Wait exit statuses are 0 for a match, 1 for timeout or general failure +(including an initially missing session), and 2 for a vanished target without +a recorded exit, including removal while waiting. A launcher that never +started or a daemon that dies without recording an exit is not successful +completion. A record no daemon has claimed yet keeps waits waiting, since its +launcher may still start it. The timeout is an independent deadline even +when no filesystem events arrive. + +A named follow ends with 0 when its session exits, vanishes, or is removed. +Both text and JSON streaming also exit cleanly with 0 as soon as their stdout +reader closes, even while idle. Options may follow the session reference, and +any other argument is an error (exit status 1) rather than silently dropping +the options after it. Followers ignore read-only filesystem notifications and coalesce change notifications, avoiding feedback from their own registry reads. Every session diff --git a/crates/pty/src/cli/events.rs b/crates/pty/src/cli/events.rs index dfda593e..1967a66e 100644 --- a/crates/pty/src/cli/events.rs +++ b/crates/pty/src/cli/events.rs @@ -5,6 +5,7 @@ //! //! node: src/cli.ts:1219-1248 (parsing), 3965-4051 (`cmdEvents`) +use std::io::{self, Write}; use std::sync::mpsc::RecvTimeoutError; use std::time::{Duration, Instant}; @@ -43,42 +44,74 @@ fn exit_zero_on_sigint() { } } -/// Lifecycle waits also observe durable exit state: the follower deliberately -/// starts existing logs at EOF, so a completed exit is not a new event. -fn print_recorded_exit(name: &str, wait_type: &str, json: bool) -> bool { - if wait_type != pty_core::events::event_type::SESSION_EXIT { - return false; - } - let Some(metadata) = registry::read_metadata(name) else { - return false; - }; - if !metadata.has_exited() && metadata.exit_code.is_none() { - return false; - } - if let Some(event) = read_retained_events_of_type(name, wait_type).pop() { - println!("{}", render(&event, json)); - } else if let Some(code) = metadata.exit_code { - // Retention or cleanup can remove the log while exit metadata remains. - let mut event = Event::session_exit(name, code, None); - if let Some(timestamp) = metadata.exited_at { - event.ts = timestamp; - } - println!("{}", render(&event, json)); +const OUTPUT_POLL_INTERVAL: Duration = Duration::from_millis(100); +const VANISHED_EXIT_CODE: i32 = 2; + +/// Polling with no requested events still reports a pipe's lost reader. +/// Unlike a zero-byte write, this detects closure even while events are idle. +fn output_closed() -> bool { + let mut fd = libc::pollfd { fd: libc::STDOUT_FILENO, events: 0, revents: 0 }; + // SAFETY: `fd` points to one initialized poll descriptor. + (unsafe { libc::poll(&mut fd, 1, 0) }) > 0 + && fd.revents & (libc::POLLERR | libc::POLLHUP | libc::POLLNVAL) != 0 +} + +fn print_event(event: &Event, json: bool) -> Result { + let stdout = io::stdout(); + let mut output = stdout.lock(); + match writeln!(output, "{}", render(event, json)).and_then(|()| output.flush()) { + Ok(()) => Ok(true), + Err(error) if error.kind() == io::ErrorKind::BrokenPipe => Ok(false), + Err(error) => Err(CliError(format!("Could not write events: {error}"))), } - true } -fn check_wait_target(name: &str, wait_type: &str) -> Result<(), CliError> { - match registry::get_session(name).map_err(CliError)? { - None => Err(CliError(format!("Session \"{name}\" not found."))), - Some(session) - if wait_type == pty_core::events::event_type::SESSION_EXIT - && session.status == registry::SessionStatus::Vanished => - { - Err(CliError(format!("Session \"{name}\" vanished without a recorded exit."))) - } - Some(_) => Ok(()), +fn current_event(event: &Event, metadata: ®istry::SessionMetadata) -> bool { + event.ts >= metadata.created_at +} + +/// Logs are cleared on replacement publication; the creation timestamp also +/// fences retained legacy records. Pin the opaque generation across reads so a +/// replacement with the same stable name cannot satisfy an old observer. +fn recorded_event( + name: &str, + wait_type: &str, + fence: ®istry::SessionMetadata, +) -> Option { + let metadata = registry::read_metadata(name)?; + if metadata.generation != fence.generation || metadata.created_at != fence.created_at { + return None; } + let event = read_retained_events_of_type(name, wait_type) + .into_iter() + .rev() + .find(|event| current_event(event, fence)) + .or_else(|| { + if wait_type != pty_core::events::event_type::SESSION_EXIT { + return None; + } + let code = metadata.exit_code?; + let timestamp = metadata.exited_at.as_ref()?; + let event = Event::session_exit(name, code, None).with_ts(timestamp); + current_event(&event, fence).then_some(event) + }); + let after = registry::read_metadata(name)?; + (after.generation == fence.generation && after.created_at == fence.created_at) + .then_some(event) + .flatten() +} + +fn target_vanished(name: &str, fence: ®istry::SessionMetadata) -> bool { + registry::get_session_by_name(name).is_none_or(|session| { + session.metadata.as_ref().is_none_or(|metadata| { + metadata.generation != fence.generation || metadata.created_at != fence.created_at + }) || session.status == registry::SessionStatus::Vanished + }) +} + +fn vanished(name: &str) -> CliResult { + eprintln!("Session \"{name}\" vanished without a recorded exit."); + Ok(VANISHED_EXIT_CODE) } /// Parse and run. @@ -168,12 +201,18 @@ fn cmd_events( Some(expected) => read_retained_events_of_type(name, expected), None => read_recent_events(name, DEFAULT_RECENT_EVENTS), }; + let metadata = registry::read_metadata(name); + let events: Vec<_> = events.into_iter().filter(|event| { + metadata.as_ref().is_none_or(|metadata| current_event(event, metadata)) + }).collect(); if events.is_empty() && event_type.is_none() { println!("No recent events for \"{name}\"."); return Ok(0); } for event in &events { - println!("{}", render(event, json)); + if !print_event(event, json)? { + return Ok(0); + } } return Ok(0); } @@ -193,57 +232,69 @@ fn cmd_events( let Some(name) = name else { return Err("--wait requires a session name.".into()); }; - if print_recorded_exit(name, wait_type, json) { - return Ok(0); - } - check_wait_target(name, wait_type)?; + let Some(fence) = registry::read_metadata(name) else { + return vanished(name); + }; exit_zero_on_sigint(); + // The channel constructor waits for the follower to establish its EOF + // offsets. Checking durable history only after this handshake leaves no + // gap between replay and subscription. let (_follower, rx) = EventFollower::channel(FollowerOptions::names(vec![name.to_string()])); - // Close the state-read/subscribe race, including removal before the - // follower observed the registry entry. - if print_recorded_exit(name, wait_type, json) { + if let Some(event) = recorded_event(name, wait_type, &fence) { + print_event(&event, json)?; return Ok(0); } - check_wait_target(name, wait_type)?; loop { - let recv = match deadline { - Some(d) => { - let now = Instant::now(); - if now >= d { - Err(RecvTimeoutError::Timeout) - } else { - rx.recv_timeout(d - now) - } + if output_closed() { + return Ok(0); + } + if target_vanished(name, &fence) { + if let Some(event) = recorded_event(name, wait_type, &fence) { + print_event(&event, json)?; + return Ok(0); } - None => rx.recv().map_err(|_| RecvTimeoutError::Disconnected), - }; - match recv { - Ok(event) => { - if event.r#type == wait_type { - println!("{}", render(&event, json)); - return Ok(0); - } + return vanished(name); + } + if deadline.is_some_and(|deadline| Instant::now() >= deadline) { + // Recheck durable history before declaring a timeout: + // filesystem notifications can be delayed. + if let Some(event) = recorded_event(name, wait_type, &fence) { + print_event(&event, json)?; + return Ok(0); } - Err(RecvTimeoutError::Timeout) => { - eprintln!( - "Timed out after {}s waiting for \"{wait_type}\" event.", - js_number(timeout) - ); - return Ok(1); + if target_vanished(name, &fence) { + return vanished(name); } + eprintln!( + "Timed out after {}s waiting for \"{wait_type}\" event.", + js_number(timeout) + ); + return Ok(1); + } + let interval = deadline.map_or(OUTPUT_POLL_INTERVAL, |deadline| { + deadline.saturating_duration_since(Instant::now()).min(OUTPUT_POLL_INTERVAL) + }); + match rx.recv_timeout(interval) { + Ok(event) if event.r#type == wait_type && current_event(&event, &fence) => { + // A stable name may have been republished while the + // event was queued. Never satisfy the original waiter + // with a replacement incarnation. + if target_vanished(name, &fence) { + return vanished(name); + } + print_event(&event, json)?; + return Ok(0); + } + Ok(_) | Err(RecvTimeoutError::Timeout) => {} Err(RecvTimeoutError::Disconnected) => { - if print_recorded_exit(name, wait_type, json) { + if let Some(event) = recorded_event(name, wait_type, &fence) { + print_event(&event, json)?; return Ok(0); } - let vanished = registry::get_session_by_name(name) - .is_some_and(|session| session.status == registry::SessionStatus::Vanished); - if vanished && wait_type == pty_core::events::event_type::SESSION_EXIT { - eprintln!("Session \"{name}\" vanished without a recorded exit."); - } else if vanished { - eprintln!("Session \"{name}\" vanished before a \"{wait_type}\" event was observed."); - } else { - eprintln!("Session \"{name}\" ended before a \"{wait_type}\" event was observed."); + if target_vanished(name, &fence) { + return vanished(name); } + eprintln!("Session \"{name}\" ended before a \"{wait_type}\" event was observed."); return Ok(1); } } @@ -257,23 +308,23 @@ fn cmd_events( }; let (_follower, rx) = EventFollower::channel(options); loop { - let recv = match deadline { - Some(d) => { - let now = Instant::now(); - if now >= d { - Err(RecvTimeoutError::Timeout) - } else { - rx.recv_timeout(d - now) + if output_closed() { + return Ok(0); + } + if deadline.is_some_and(|deadline| Instant::now() >= deadline) { + eprintln!("Timed out after {}s following events.", js_number(timeout)); + return Ok(1); + } + let interval = deadline.map_or(OUTPUT_POLL_INTERVAL, |deadline| { + deadline.saturating_duration_since(Instant::now()).min(OUTPUT_POLL_INTERVAL) + }); + match rx.recv_timeout(interval) { + Ok(event) => { + if !print_event(&event, json)? { + return Ok(0); } } - None => rx.recv().map_err(|_| RecvTimeoutError::Disconnected), - }; - match recv { - Ok(event) => println!("{}", render(&event, json)), - Err(RecvTimeoutError::Timeout) => { - eprintln!("Timed out after {}s following events.", js_number(timeout)); - return Ok(1); - } + Err(RecvTimeoutError::Timeout) => {} Err(RecvTimeoutError::Disconnected) => return Ok(0), } } diff --git a/crates/pty/src/cli/help.rs b/crates/pty/src/cli/help.rs index 47d3ef72..74bd4263 100644 --- a/crates/pty/src/cli/help.rs +++ b/crates/pty/src/cli/help.rs @@ -1,12 +1,12 @@ //! Help text for the `pty` binary. //! -//! Every text here is vendored verbatim from the Node `pty` at `e6c1cd4`. The -//! fixtures under `crates/pty/tests/fixtures/help/` are the bytes that binary -//! printed (`pty help`, `pty --help`, ...), captured by running it under a -//! scratch `PTY_ROOT`, trailing newline included. Nothing is generated or -//! reformatted here; a text changes only when its fixture is re-captured. -//! One edit since the capture: the `--remote` examples in `attach`, `list`, -//! `peek` and `send` name an invented peer, `example-host`. +//! Help text starts from the Node `pty` at `e6c1cd4`. The fixtures under +//! `crates/pty/tests/fixtures/help/` are included directly as the help source, +//! with their trailing newline. They were captured under a scratch `PTY_ROOT` +//! and are updated where the Rust contract differs: the `--remote` examples +//! in `attach`, `list`, `peek` and `send` name an invented peer, `example-host`, +//! and `events` documents durable current-generation observation, wait statuses, +//! and clean streaming reader closure. //! //! The two deferred commands (`recover`, `test`) keep the Node help so the //! binary stays a drop-in; the commands themselves report that they are not diff --git a/crates/pty/tests/cli_events.rs b/crates/pty/tests/cli_events.rs index 61bd035c..e24f45b8 100644 --- a/crates/pty/tests/cli_events.rs +++ b/crates/pty/tests/cli_events.rs @@ -35,7 +35,7 @@ fn recent_type_filter_returns_every_retained_exact_match_in_order_without_writes const RECENT_BOUND: usize = 50; let rig = Rig::new(); - rig.write_meta("query", json!({})); + rig.write_meta("query", json!({"createdAt": "2026-01-01T00:00:00.000Z"})); let mut matching = vec![ json!({ "session": "query", @@ -121,7 +121,9 @@ fn recent_type_filter_returns_every_retained_exact_match_in_order_without_writes #[test] fn recent_and_usage() { let rig = Rig::new(); - rig.write_meta("s", json!({"displayName": "Friendly"})); + rig.write_meta("s", json!({ + "displayName": "Friendly", "createdAt": "2026-01-01T00:00:00.000Z" + })); assert_eq!(rig.ok(&["events", "--recent", "s"]).stdout, "No recent events for \"s\".\n"); append_event(&rig, "s", r#"{"session":"s","type":"session_start","ts":"2026-01-02T03:04:05.000Z","tags":{"role":"web"}}"#); append_event(&rig, "s", r#"{"session":"s","type":"user.note","ts":"2026-01-02T03:04:06.000Z","text":"hi"}"#); @@ -170,7 +172,7 @@ fn recent_and_usage() { #[test] fn wait_prints_the_first_matching_event_or_times_out() { let rig = Rig::new(); - rig.write_meta("w", json!({})); + rig.write_meta("w", json!({"createdAt": "2026-01-01T00:00:00.000Z"})); let out = rig.run(&["events", "--wait", "bell", "-t", "1", "w"]); assert_eq!(out.code, 1); assert_eq!(out.stderr, "Timed out after 1s waiting for \"bell\" event.\n"); @@ -382,11 +384,222 @@ fn waits_and_follows_end_when_their_daemon_vanishes() { let out = finish_within(child, Duration::from_secs(3)); let stderr = String::from_utf8_lossy(&out.stderr); let (code, message) = match i % 3 { - 0 => (1, "Session \"doomed\" vanished before a \"user.never\" event was observed.\n"), - 1 => (1, "Session \"doomed\" vanished without a recorded exit.\n"), + 0 | 1 => (2, "Session \"doomed\" vanished without a recorded exit.\n"), _ => (0, ""), }; assert_eq!(out.status.code(), Some(code), "{i}: {out:?}"); assert_eq!(stderr, message, "{i}"); } } + +fn spawn_wait(rig: &Rig, name: &str, event_type: &str, json: bool) -> std::process::Child { + let mut args = vec!["events", name, "--wait", event_type, "--timeout", "10"]; + if json { + args.push("--json"); + } + rig.cmd(&args).stdout(Stdio::piped()).stderr(Stdio::piped()).spawn().unwrap() +} + +#[test] +fn wait_replays_a_recorded_exit_before_subscription_in_json_and_text() { + let rig = Rig::new(); + let event = json!({ + "session": "already-exited", "type": "session_exit", + "ts": "2026-01-02T03:04:06.000Z", "exitCode": 17 + }); + rig.write_meta("already-exited", json!({ + "createdAt": "2026-01-01T00:00:00.000Z", + "exitedAt": "2026-01-02T03:04:06.000Z", "exitCode": 17 + })); + append_event(&rig, "already-exited", &event.to_string()); + for json in [true, false] { + let out = finish_within( + spawn_wait(&rig, "already-exited", "session_exit", json), + Duration::from_secs(2), + ); + assert_eq!(out.status.code(), Some(0), "{out:?}"); + assert!(out.stderr.is_empty(), "{out:?}"); + let stdout = String::from_utf8(out.stdout).unwrap(); + if json { + assert_eq!(parse_json_lines(&stdout), vec![event.clone()]); + } else { + assert_eq!(stdout.lines().count(), 1, "{stdout}"); + assert!(stdout.ends_with("] already-exited: exited (code 17)\n"), "{stdout}"); + } + } +} + +#[test] +fn wait_observes_exit_published_after_the_wait_process_starts() { + let rig = Rig::new(); + rig.write_meta("exit-during-wait", json!({"createdAt": "2026-01-01T00:00:00.000Z"})); + let child = spawn_wait(&rig, "exit-during-wait", "session_exit", true); + // No readiness event is printed by --wait. Publishing after spawn covers + // both a live notification and the subscribe/snapshot race without relying + // on an arbitrary sleep to decide when the subscriber is ready. + let event = json!({ + "session": "exit-during-wait", "type": "session_exit", + "ts": "2026-01-02T03:04:06.000Z", "exitCode": 23 + }); + append_event(&rig, "exit-during-wait", &event.to_string()); + rig.write_meta("exit-during-wait", json!({ + "createdAt": "2026-01-01T00:00:00.000Z", + "exitedAt": "2026-01-02T03:04:06.000Z", "exitCode": 23 + })); + let out = finish_within(child, Duration::from_secs(2)); + assert_eq!(out.status.code(), Some(0), "{out:?}"); + assert!(out.stderr.is_empty(), "{out:?}"); + assert_eq!(parse_json_lines(&String::from_utf8(out.stdout).unwrap()), vec![event]); +} + +#[test] +fn wait_replays_arbitrary_exact_types_beyond_the_recent_tail() { + let rig = Rig::new(); + for event_type in ["bell", "user.note", "display_name_change"] { + rig.write_meta("retained", json!({"createdAt": "2026-01-01T00:00:00.000Z"})); + std::fs::write(rig.path("retained.events.jsonl"), "").unwrap(); + append_event(&rig, "retained", &json!({ + "session": "retained", "type": format!("{event_type}.suffix"), + "ts": "2026-01-02T03:04:05.000Z" + }).to_string()); + let first = json!({ + "session": "retained", "type": event_type, + "ts": "2026-01-02T03:04:06.000Z", "data": {"sequence": 1} + }); + append_event(&rig, "retained", &first.to_string()); + let latest = json!({ + "session": "retained", "type": event_type, + "ts": "2026-01-02T03:04:06.500Z", "data": {"sequence": 2} + }); + append_event(&rig, "retained", &latest.to_string()); + for sequence in 0..60 { + append_event(&rig, "retained", &json!({ + "session": "retained", "type": "user.distractor", + "ts": "2026-01-02T03:04:07.000Z", "data": {"sequence": sequence} + }).to_string()); + } + let before = std::fs::read(rig.path("retained.events.jsonl")).unwrap(); + let out = finish_within( + spawn_wait(&rig, "retained", event_type, true), + Duration::from_secs(2), + ); + assert_eq!(out.status.code(), Some(0), "{event_type}: {out:?}"); + assert!(out.stderr.is_empty(), "{out:?}"); + assert_eq!(parse_json_lines(&String::from_utf8(out.stdout).unwrap()), vec![latest]); + assert_eq!(std::fs::read(rig.path("retained.events.jsonl")).unwrap(), before); + } +} + +#[test] +fn wait_ignores_previous_generation_events_and_accepts_the_creation_boundary() { + let rig = Rig::new(); + for event_type in ["user.note", "session_exit"] { + rig.write_meta("reused", json!({ + "createdAt": "2026-01-02T03:04:06.000Z", "generation": "current" + })); + std::fs::write(rig.path("reused.events.jsonl"), "").unwrap(); + append_event(&rig, "reused", &json!({ + "session": "reused", "type": event_type, + "ts": "2026-01-02T03:04:05.999Z", "exitCode": 99 + }).to_string()); + let out = finish_within( + rig.cmd(&["events", "reused", "--wait", event_type, "-t", "0.2", "--json"]) + .stdout(Stdio::piped()).stderr(Stdio::piped()).spawn().unwrap(), + Duration::from_secs(2), + ); + assert_eq!(out.status.code(), Some(1), "{event_type}: {out:?}"); + assert!(out.stdout.is_empty(), "{out:?}"); + assert_eq!( + String::from_utf8_lossy(&out.stderr), + format!("Timed out after 0.2s waiting for \"{event_type}\" event.\n"), + ); + let current = json!({ + "session": "reused", "type": event_type, + "ts": "2026-01-02T03:04:06.000Z", "exitCode": 0 + }); + append_event(&rig, "reused", ¤t.to_string()); + let out = finish_within( + spawn_wait(&rig, "reused", event_type, true), + Duration::from_secs(2), + ); + assert_eq!(out.status.code(), Some(0), "{event_type}: {out:?}"); + assert!(out.stderr.is_empty(), "{out:?}"); + assert_eq!(parse_json_lines(&String::from_utf8(out.stdout).unwrap()), vec![current]); + } +} + +#[test] +fn wait_reports_registry_removal_as_vanished_not_timeout() { + let rig = Rig::new(); + for event_type in ["user.never", "session_exit"] { + rig.write_meta("removed-wait", json!({})); + let mut child = spawn_wait(&rig, "removed-wait", event_type, true); + // Unlike followers, waits have no stdout readiness marker. Keep the + // process alive through several follower polling intervals before + // removing the registry entry; all completion waits remain bounded. + let deadline = Instant::now() + Duration::from_millis(300); + while Instant::now() < deadline { + assert!(child.try_wait().unwrap().is_none(), "wait ended before removal"); + std::thread::sleep(Duration::from_millis(10)); + } + std::fs::remove_file(rig.path("removed-wait.json")).unwrap(); + let out = finish_within(child, Duration::from_secs(2)); + assert_eq!(out.status.code(), Some(2), "{event_type}: {out:?}"); + assert!(out.stdout.is_empty(), "{out:?}"); + assert_eq!( + String::from_utf8_lossy(&out.stderr), + "Session \"removed-wait\" vanished without a recorded exit.\n", + ); + } +} + +#[test] +fn idle_follow_exits_zero_when_its_stdout_reader_closes_in_json_and_text() { + let rig = Rig::new(); + for json in [true, false] { + rig.write_meta("idle-pipe", json!({})); + let mut args = vec!["events", "idle-pipe"]; + if json { + args.push("--json"); + } + let mut child = rig.cmd(&args) + .stdout(Stdio::piped()).stderr(Stdio::piped()).spawn().unwrap(); + let stdout = child.stdout.take().unwrap(); + let (tx, rx) = std::sync::mpsc::channel(); + let reader = std::thread::spawn(move || { + let mut reader = std::io::BufReader::new(stdout); + let mut line = String::new(); + reader.read_line(&mut line).unwrap(); + tx.send((reader, line)).ok(); + }); + let deadline = Instant::now() + Duration::from_secs(3); + let (pipe, line) = loop { + append_event(&rig, "idle-pipe", &json!({ + "session": "idle-pipe", "type": "user.ready", + "ts": "2026-01-02T03:04:05.000Z" + }).to_string()); + match rx.recv_timeout(Duration::from_millis(100)) { + Ok(value) => break value, + Err(std::sync::mpsc::RecvTimeoutError::Timeout) if Instant::now() < deadline => {} + result => { + child.kill().unwrap(); + child.wait().unwrap(); + reader.join().unwrap(); + panic!("idle follower never became ready: {result:?}"); + } + } + }; + reader.join().unwrap(); + if json { + assert_eq!(parse_json_lines(&line)[0]["type"], "user.ready"); + } else { + assert!(line.ends_with("] idle-pipe: user.ready\n"), "{line}"); + } + // No event is appended after readiness: a broken-pipe-on-write-only + // implementation would hang here instead of observing reader closure. + drop(pipe); + let out = finish_within(child, Duration::from_secs(2)); + assert_eq!(out.status.code(), Some(0), "json={json}: {out:?}"); + assert!(out.stderr.is_empty(), "{out:?}"); + } +} diff --git a/crates/pty/tests/fixtures/help/events.txt b/crates/pty/tests/fixtures/help/events.txt index e109f558..9db27453 100644 --- a/crates/pty/tests/fixtures/help/events.txt +++ b/crates/pty/tests/fixtures/help/events.txt @@ -7,16 +7,29 @@ Flags: --recent Print recent events and exit (don't follow) --json Emit raw JSONL --type With --recent, print every retained event of this exact type - --wait Block until an event of appears + --wait Replay current-generation retained matches, then wait for a match -t, --timeout Timeout (seconds) for --wait or follow; exit 1 on timeout Options may appear before or after ; any other argument is an error. A named follow exits when the session exits, its daemon vanishes without recording an exit, or its registry entry is removed outside a locked respawn; --all continues watching new sessions. Terminal metadata is followed by a final event-log drain. ---wait exits 1 if the session ends before a matching event, or on timeout. +Waits subscribe before checking durable current-generation events, recheck after +subscription and before timeout, and follow safely for any event type. session_exit +also checks recorded exit metadata if the event log is gone. +--wait exits 0 on a match, 1 on timeout/general failure (including an initially +missing session), and 2 if the target vanishes without a recorded exit, including +removal while waiting. Vanished is not successful completion. +Text and JSON streaming exit cleanly with 0 when the stdout reader closes, even idle. Examples: pty events myserver pty events --recent --json myserver pty events --recent --json --type user.tests-passed myserver + +Durable completion observation (inspect the recorded exit code separately): + pty events --recent --type session_exit --json REF + pty events --wait session_exit -t SECONDS --json REF + pty events --recent --type session_exit --json REF +Check retained current-generation events first; if none match, use a bounded wait +and a final durable recheck. An empty --recent result is not proof of completion.