diff --git a/crates/daemon/src/service.rs b/crates/daemon/src/service.rs index c7e6181b..61b1de9b 100644 --- a/crates/daemon/src/service.rs +++ b/crates/daemon/src/service.rs @@ -4,6 +4,7 @@ //! HTTP remains loopback-only. Transport adapters are separate from the shared //! ingress router so adding a channel does not fork session semantics. +mod harness_error; mod http; mod ingress; mod slack; diff --git a/crates/daemon/src/service/harness_error.rs b/crates/daemon/src/service/harness_error.rs new file mode 100644 index 00000000..1083d7fe --- /dev/null +++ b/crates/daemon/src/service/harness_error.rs @@ -0,0 +1,298 @@ +//! Recovering a harness's own error text from what it drew on its PTY. +//! +//! An interactive harness that fails mid-turn does not necessarily tell the +//! daemon anything. It prints the failure into its viewport and returns to its +//! composer, which from outside is indistinguishable from a turn that finished +//! — except that no answer came out of it. The text is right there on screen, +//! and it is usually the single most useful thing to hand the person waiting +//! at the other end of a channel: "stream disconnected before completion" +//! tells them what to do next, "the turn ended without an answer" does not. +//! +//! This is best-effort by construction. It reads a bounded tail of the PTY +//! stream, so it can only ever be as good as what the harness chose to draw, +//! and it is only consulted for a turn already known to have failed — a wrong +//! guess costs a slightly-off sentence in a failure notice, never an answer. + +use construct_protocol::{SessionEvent, TimestampedEvent}; + +/// How much decoded PTY tail to consider: comfortably more than a screenful, +/// far less than a long session's scrollback. +const TAIL_BYTES: usize = 32 * 1024; + +/// The longest error text to quote into a channel message. +const MAX_DETAIL: usize = 240; + +/// Glyphs harnesses draw to mark a line as an error. Matching the marker +/// rather than the wording keeps this from being a list of every phrase every +/// harness might use. +const ERROR_MARKERS: [char; 5] = ['■', '✗', '✘', '×', '⚠']; + +/// Glyphs that begin the harness's own chrome — the composer prompt and its +/// hints. Reaching one means the error text ended and the screen has moved on +/// to furniture that would be nonsense to quote. +const CHROME_MARKERS: [char; 4] = ['›', '‣', '⏵', '>']; + +/// The harness's last drawn error, if it drew one. +pub(super) fn harness_error_detail(events: &[TimestampedEvent]) -> Option { + let lines = display_lines(&pty_tail(events)); + let start = lines.iter().rposition(|line| is_error_line(line))?; + let mut detail = strip_marker(&lines[start]).to_string(); + // Harnesses wrap a long error across several drawn lines; the URL that + // says *which* endpoint failed routinely lands on the second one. Keep + // taking lines until the screen changes subject. + for line in &lines[start + 1..] { + if line.is_empty() || starts_with_chrome(line) { + break; + } + if detail.chars().count() + 1 + line.chars().count() > MAX_DETAIL { + break; + } + detail.push(' '); + detail.push_str(line); + } + let detail = detail.trim(); + (!detail.is_empty()).then(|| truncate(detail)) +} + +fn is_error_line(line: &str) -> bool { + let trimmed = line.trim_start(); + if trimmed.starts_with(ERROR_MARKERS) { + // A bare marker with nothing after it is a box-drawing character in + // some unrelated widget, not a message. + return !strip_marker(trimmed).is_empty(); + } + // Harnesses that spell it out instead of drawing a glyph. Anchored at the + // start so an answer *discussing* an error is not mistaken for one. + let lowered = trimmed.to_ascii_lowercase(); + ["error:", "error ", "fatal:", "failed to "] + .iter() + .any(|prefix| lowered.starts_with(prefix)) +} + +fn starts_with_chrome(line: &str) -> bool { + line.trim_start().starts_with(CHROME_MARKERS) +} + +fn strip_marker(line: &str) -> &str { + line.trim_start() + .trim_start_matches(ERROR_MARKERS) + .trim_start() +} + +fn truncate(detail: &str) -> String { + if detail.chars().count() <= MAX_DETAIL { + return detail.to_string(); + } + let kept: String = detail.chars().take(MAX_DETAIL - 1).collect(); + format!("{}…", kept.trim_end()) +} + +/// The tail of the session's raw PTY byte stream. +/// +/// Chunks are concatenated as bytes before being decoded: a harness is free to +/// split a multi-byte character across two writes, and decoding each chunk on +/// its own would turn the error marker itself into replacement characters. +fn pty_tail(events: &[TimestampedEvent]) -> String { + use base64::Engine as _; + let mut chunks: Vec> = Vec::new(); + let mut total = 0usize; + for event in events.iter().rev() { + let SessionEvent::Pty { data } = &event.event else { + continue; + }; + let Ok(bytes) = base64::engine::general_purpose::STANDARD.decode(data) else { + continue; + }; + total += bytes.len(); + chunks.push(bytes); + if total >= TAIL_BYTES { + break; + } + } + chunks.reverse(); + String::from_utf8_lossy(&chunks.concat()).into_owned() +} + +/// Split a raw terminal stream into the lines it would have drawn. +/// +/// A TUI does not end its lines with newlines — it moves the cursor and erases. +/// Treating cursor motion and erasure as line breaks is what keeps a redraw +/// from collapsing an error banner, the composer, and the model footer into one +/// run-on string. Everything else in an escape sequence is presentation and is +/// dropped. +fn display_lines(raw: &str) -> Vec { + let mut lines = Vec::new(); + let mut current = String::new(); + let mut chars = raw.chars().peekable(); + while let Some(ch) = chars.next() { + match ch { + '\x1b' => match chars.next() { + Some('[') => { + let mut final_byte = None; + for next in chars.by_ref() { + if ('\u{40}'..='\u{7e}').contains(&next) { + final_byte = Some(next); + break; + } + } + if matches!( + final_byte, + Some('A'..='H' | 'J' | 'K' | 'L' | 'M' | 'S' | 'T' | 'f') + ) { + lines.push(std::mem::take(&mut current)); + } + } + // OSC (window title and friends): runs to BEL or ST. + Some(']') => { + while let Some(next) = chars.next() { + if next == '\u{7}' { + break; + } + if next == '\x1b' { + chars.next(); + break; + } + } + } + _ => {} + }, + '\r' | '\n' => lines.push(std::mem::take(&mut current)), + ch if ch.is_control() => {} + ch => current.push(ch), + } + } + lines.push(current); + lines + .into_iter() + .map(|line| line.trim().to_string()) + .collect() +} + +#[cfg(test)] +mod tests { + use super::*; + use base64::Engine as _; + + fn pty(raw: &str) -> TimestampedEvent { + TimestampedEvent { + seq: 0, + at: chrono::Utc::now(), + event: SessionEvent::Pty { + data: base64::engine::general_purpose::STANDARD.encode(raw.as_bytes()), + }, + } + } + + #[test] + fn a_codex_stream_failure_is_recovered_whole() { + // Captured verbatim from the PTY of a live codex service session whose + // turn died against a router port that had moved. The banner and the + // URL that identifies it arrive in two separate writes, and the + // composer redraw follows immediately behind — the message is only + // useful if all three are handled: join the first two, stop before the + // third. + let events = vec![ + pty("\x1b[39;49m\x1b[K\x1b[39m\x1b[49m\x1b[0m\r\n\x1b[39;49m\x1b[K\x1b[38;5;1;49m■ stream disconnected before completion: error sending request for url\x1b[39m\x1b[49m\x1b[0m"), + pty("\x1b[39;49m\x1b[K\x1b[38;5;1;49m(https://chatgpt.com/backend-api/codex/responses)\x1b[39m\x1b[49m\x1b[0m\x1b[r\x1b[60;3H\x1b[57;2H\x1b[0m\x1b[49m\x1b[K\x1b[59;1H\x1b[1m›\x1b[59;3H\x1b[22m\x1b[2mExplain this codebase\x1b[61;3H\x1b[22mgpt-5.6-luna high fast\x1b[0m"), + ]; + + assert_eq!( + harness_error_detail(&events).as_deref(), + Some( + "stream disconnected before completion: error sending request for url \ + (https://chatgpt.com/backend-api/codex/responses)" + ) + ); + } + + #[test] + fn the_composer_is_never_quoted_as_part_of_the_error() { + let events = vec![pty( + "\x1b[K■ something broke\x1b[K› ask me anything\x1b[Kmodel · ~/somewhere", + )]; + assert_eq!( + harness_error_detail(&events).as_deref(), + Some("something broke") + ); + } + + #[test] + fn the_last_error_wins_when_a_session_has_failed_before() { + // The tail carries every error this session ever drew. Only the one + // belonging to the turn that just ended is worth reporting. + let events = vec![pty("■ an old failure\r\n\r\n"), pty("■ the current one\r\n")]; + assert_eq!( + harness_error_detail(&events).as_deref(), + Some("the current one") + ); + } + + #[test] + fn a_session_that_drew_no_error_yields_nothing() { + let events = vec![pty("\x1b[Kall good here\x1b[K› ask me anything")]; + assert_eq!(harness_error_detail(&events), None); + } + + #[test] + fn spelled_out_errors_are_recognized_without_a_glyph() { + let events = vec![pty("\x1b[KError: the model refused the request\x1b[K")]; + assert_eq!( + harness_error_detail(&events).as_deref(), + Some("Error: the model refused the request") + ); + } + + #[test] + fn an_answer_that_merely_discusses_an_error_is_not_one() { + // Anchoring at the start of a line is what keeps a genuine reply about + // error handling from being reported as a failure. + let events = vec![pty( + "\x1b[KThe function returns an error when the file is missing.\x1b[K", + )]; + assert_eq!(harness_error_detail(&events), None); + } + + #[test] + fn a_multi_byte_glyph_split_across_two_writes_still_matches() { + // The marker is three bytes and a harness may flush mid-character. + // Decoding chunk-by-chunk would replace it with U+FFFD and lose the + // line entirely. + let marker = "■".as_bytes(); + let head = [&b"\x1b[K"[..], &marker[..1]].concat(); + let tail = [&marker[1..], &b" split banner\r\n"[..]].concat(); + let events = vec![ + TimestampedEvent { + seq: 0, + at: chrono::Utc::now(), + event: SessionEvent::Pty { + data: base64::engine::general_purpose::STANDARD.encode(&head), + }, + }, + TimestampedEvent { + seq: 1, + at: chrono::Utc::now(), + event: SessionEvent::Pty { + data: base64::engine::general_purpose::STANDARD.encode(&tail), + }, + }, + ]; + assert_eq!( + harness_error_detail(&events).as_deref(), + Some("split banner") + ); + } + + #[test] + fn a_runaway_error_is_capped_rather_than_pasted_whole() { + let long = "x".repeat(1000); + let events = vec![pty(&format!("\x1b[K■ {long}\r\n"))]; + let detail = harness_error_detail(&events).unwrap(); + assert!(detail.chars().count() <= MAX_DETAIL); + assert!(detail.ends_with('…')); + } + + #[test] + fn sessions_with_no_pty_at_all_are_handled() { + assert_eq!(harness_error_detail(&[]), None); + } +} diff --git a/crates/daemon/src/service/ingress.rs b/crates/daemon/src/service/ingress.rs index e56230de..7699f6df 100644 --- a/crates/daemon/src/service/ingress.rs +++ b/crates/daemon/src/service/ingress.rs @@ -22,6 +22,20 @@ use uuid::Uuid; const REQUEST_DEDUP_CAP: usize = 4096; const PENDING_DELIVERY_TTL: std::time::Duration = std::time::Duration::from_secs(30 * 60); +/// How long a session must sit idle and quiet, with no approval pending, +/// before a turn that produced no answer is called finished-and-failed. +/// +/// Long enough that the gap between two tool calls cannot be mistaken for the +/// end of a turn, short enough that the person waiting learns something in +/// seconds rather than at the delivery TTL. +const IDLE_SETTLE: std::time::Duration = std::time::Duration::from_secs(10); + +/// The same, for a delivery whose turn was never observed to start at all. +/// More generous, because "has not begun yet" is the normal state of a session +/// for the first moments after input reaches it, and a harness that is slow to +/// pick input up must not be declared broken. +const NEVER_STARTED_GRACE: std::time::Duration = std::time::Duration::from_secs(60); + struct PendingDelivery { session_id: String, created_at: tokio::time::Instant, @@ -364,12 +378,14 @@ impl ServiceIngress { .await; } let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(30 * 60); + let mut watch = TurnWatch::default(); loop { tokio::select! { _ = cancel.cancelled() => return Err(anyhow!("channel stopped")), _ = tokio::time::sleep(std::time::Duration::from_millis(250)) => {} } - if tokio::time::Instant::now() >= deadline { + let now = tokio::time::Instant::now(); + if now >= deadline { return Err(anyhow!("service turn timed out")); } let Ok(detail) = self.shared.manager.detail(&receipt.session).await else { @@ -396,9 +412,15 @@ impl ServiceIngress { return Ok(reply); } if detail.summary.state == SessionState::Errored { - return Err(anyhow!("service session errored without a final reply")); + return Err(turn_failed_error(&detail)); } } + // The turn is over and left no answer behind. Without this the + // loop would keep polling a session that has already gone back to + // its composer, and say nothing until the deadline above. + if watch.settled_without_answer(&detail, now) { + return Err(turn_failed_error(&detail)); + } } } @@ -410,6 +432,7 @@ impl ServiceIngress { progress: &watch::Sender, ) -> Result { let deadline = tokio::time::Instant::now() + PENDING_DELIVERY_TTL; + let mut watch = TurnWatch::default(); loop { tokio::select! { _ = cancel.cancelled() => { @@ -418,7 +441,8 @@ impl ServiceIngress { }, _ = tokio::time::sleep(std::time::Duration::from_millis(250)) => {} } - if tokio::time::Instant::now() >= deadline { + let now = tokio::time::Instant::now(); + if now >= deadline { self.shared.cancel_delivery(delivery_id).await; return Err(anyhow!("service turn timed out")); } @@ -434,9 +458,14 @@ impl ServiceIngress { } if detail.summary.state == SessionState::Errored { self.shared.cancel_delivery(delivery_id).await; - return Err(anyhow!( - "interactive service session errored before replying" - )); + return Err(turn_failed_error(&detail)); + } + // A live agent TUI that fails mid-turn returns to its composer + // without ever erroring its session, so the reply this loop is + // waiting for is never coming. Nothing above would notice. + if watch.settled_without_answer(&detail, now) { + self.shared.cancel_delivery(delivery_id).await; + return Err(turn_failed_error(&detail)); } } } @@ -756,6 +785,96 @@ pub(super) enum IngressProgress { }, } +/// Watches one delivery's turn for the moment it stops without answering. +/// +/// A harness that fails mid-turn does not necessarily error its session. An +/// interactive TUI prints the failure into its viewport and returns to its +/// composer, which from outside looks exactly like a turn that finished — the +/// session is idle, nothing is pending, and the only thing distinguishing the +/// two is that no answer was produced. Left undetected, that difference costs +/// the person waiting the entire delivery TTL of silence. +/// +/// Idleness alone is not enough to conclude anything: a session is briefly +/// idle before it picks up delivered input, and a coarse state can lag the +/// transcript. So the turn must be idle, unqueued, not stopped at an approval, +/// and appending nothing, all continuously, before it is called over. +#[derive(Default)] +struct TurnWatch { + /// Whether this turn was ever seen running. Until it has been, the session + /// being idle means "not started yet", not "finished". + saw_running: bool, + /// When the current continuous stretch of quiet idleness began. + idle_since: Option, + /// Count of transcript events that represent progress. PTY frames are + /// excluded deliberately: an interactive harness repaints its viewport + /// constantly — including while idle — so counting those would mean a + /// turn never looks quiet and this never fires. + progress_events: usize, +} + +impl TurnWatch { + /// Fold in one poll of the session. True once the turn has demonstrably + /// stopped without producing an answer. + fn settled_without_answer( + &mut self, + detail: &construct_protocol::SessionDetail, + now: tokio::time::Instant, + ) -> bool { + let idle = matches!( + detail.summary.state, + SessionState::AwaitingInput | SessionState::Done + ) && !detail.summary.pending_input + && pending_approval(&detail.events).is_none(); + let progress_events = detail + .events + .iter() + .filter(|event| is_progress_event(&event.event)) + .count(); + + if !idle { + self.saw_running = true; + self.idle_since = None; + self.progress_events = progress_events; + return false; + } + // Still appending is still working, whatever the coarse state says. + if progress_events != self.progress_events { + self.progress_events = progress_events; + self.idle_since = Some(now); + return false; + } + let since = *self.idle_since.get_or_insert(now); + let grace = if self.saw_running { + IDLE_SETTLE + } else { + NEVER_STARTED_GRACE + }; + now.duration_since(since) >= grace + } +} + +/// Whether an event represents the turn getting somewhere, as opposed to the +/// harness redrawing what is already on screen. +fn is_progress_event(event: &SessionEvent) -> bool { + matches!( + event, + SessionEvent::Message { .. } + | SessionEvent::Reasoning { .. } + | SessionEvent::ToolUse { .. } + | SessionEvent::ToolResult { .. } + | SessionEvent::ToolApprovalRequest { .. } + ) +} + +/// Why a turn that produced no answer ended, in the harness's own words where +/// it left any. Phrased to read as a cause, since channels present it as one. +fn turn_failed_error(detail: &construct_protocol::SessionDetail) -> anyhow::Error { + match super::harness_error::harness_error_detail(&detail.events) { + Some(reported) => anyhow!("{reported}"), + None => anyhow!("the session stopped without replying"), + } +} + /// Publish the turn's current phase, if it changed. Only the waiting task /// writes here, so a plain compare-then-send cannot race itself. fn publish_progress( @@ -1250,4 +1369,237 @@ mod tests { "general MCP grants keep plugin tools, including an installed Slack MCP, available" ); } + + /// A session detail with only the fields turn-watching reads. + fn watched( + state: SessionState, + pending_input: bool, + events: Vec, + ) -> construct_protocol::SessionDetail { + let at = chrono::Utc::now(); + construct_protocol::SessionDetail { + summary: construct_protocol::SessionSummary { + id: "s1".into(), + harness: "codex".into(), + cwd: "/tmp".into(), + title: None, + auto_title_pending: false, + state, + created_at: at, + last_event_at: Some(at), + last_message_at: None, + cost_usd: None, + model: None, + effort: None, + route: None, + route_capable: false, + worktree: None, + pending_input, + last_prompt: None, + event_count: events.len() as u64, + has_pty: true, + mode: Some("interactive".into()), + pinned: false, + position: 0, + group_id: None, + parent_session_id: None, + native_subagent: None, + last_pty_at_ms: None, + busy_ms: 0, + busy_running_since_ms: None, + message_count: 0, + tokens: Default::default(), + context_used: None, + context_window: None, + context_segments: Vec::new(), + approval_mode: construct_protocol::ApprovalMode::Manual, + kind: SessionKind::User, + archived: false, + forked_from: None, + merge: None, + operator_loop_disabled: false, + needs_attention: false, + }, + events: events + .into_iter() + .enumerate() + .map(|(seq, event)| construct_protocol::TimestampedEvent { + at, + seq: seq as u64, + event, + }) + .collect(), + ui_panels: Vec::new(), + } + } + + #[test] + fn a_turn_that_returns_to_its_composer_without_answering_is_over() { + // The failure this exists for: an interactive harness loses its + // upstream mid-turn, draws the error in its viewport, and goes back to + // awaiting input. The session never errors, so nothing else in either + // wait loop notices, and the caller would otherwise poll until the + // delivery TTL — half an hour of a channel saying "working on it". + let mut watch = TurnWatch::default(); + let start = tokio::time::Instant::now(); + + let running = watched(SessionState::Running, false, vec![]); + assert!(!watch.settled_without_answer(&running, start)); + + let idle = watched(SessionState::AwaitingInput, false, vec![]); + assert!(!watch.settled_without_answer(&idle, start)); + // Idle, but not for long enough to be sure yet. + assert!(!watch.settled_without_answer(&idle, start + IDLE_SETTLE / 2)); + assert!(watch.settled_without_answer(&idle, start + IDLE_SETTLE)); + } + + #[test] + fn a_turn_still_appending_is_never_called_finished() { + // Coarse state can lag the transcript, and some harnesses sit at + // "awaiting input" between steps. Anything new arriving restarts the + // clock, so only genuine quiet concludes a turn. + let mut watch = TurnWatch::default(); + let start = tokio::time::Instant::now(); + watch.settled_without_answer(&watched(SessionState::Running, false, vec![]), start); + + let mut events = vec![SessionEvent::Message { + role: MessageRole::User, + text: "go".into(), + }]; + let mut now = start; + for step in 0..4 { + events.push(SessionEvent::ToolUse { + tool: "bash".into(), + args: serde_json::Value::Null, + call_id: Some(format!("c{step}")), + }); + now += IDLE_SETTLE - std::time::Duration::from_secs(1); + assert!( + !watch.settled_without_answer( + &watched(SessionState::AwaitingInput, false, events.clone()), + now + ), + "a turn that just appended a tool call has not stopped" + ); + } + // Once it genuinely stops appending, the clock runs out. + now += IDLE_SETTLE; + assert!(watch.settled_without_answer( + &watched(SessionState::AwaitingInput, false, events), + now + )); + } + + #[test] + fn a_repainting_idle_harness_does_not_look_busy() { + // An interactive TUI writes to its PTY constantly, including while it + // sits at an empty composer. If PTY frames counted as progress this + // would never fire for exactly the sessions it exists to serve. + let mut watch = TurnWatch::default(); + let start = tokio::time::Instant::now(); + watch.settled_without_answer(&watched(SessionState::Running, false, vec![]), start); + + let repaint = |events: &mut Vec| { + events.push(SessionEvent::Pty { + data: "cmVwYWludA==".into(), + }) + }; + let mut events = Vec::new(); + repaint(&mut events); + + // The first idle poll starts the clock. + let idle_began = start; + assert!(!watch.settled_without_answer( + &watched(SessionState::AwaitingInput, false, events.clone()), + idle_began + )); + // Repaints keep arriving and must not push the conclusion back. + for step in 1..=3 { + repaint(&mut events); + assert!( + !watch.settled_without_answer( + &watched(SessionState::AwaitingInput, false, events.clone()), + idle_began + IDLE_SETTLE / 4 * step, + ), + "not settled yet, but only because too little time has passed" + ); + } + repaint(&mut events); + assert!(watch.settled_without_answer( + &watched(SessionState::AwaitingInput, false, events), + idle_began + IDLE_SETTLE + )); + } + + #[test] + fn a_turn_stopped_at_an_approval_is_waiting_not_finished() { + // An approval is the one kind of stop that is *supposed* to last: it + // resolves when a human acts. Calling it a failure would replace a + // truthful "waiting for an operator" with a wrong "it broke". + let mut watch = TurnWatch::default(); + let start = tokio::time::Instant::now(); + watch.settled_without_answer(&watched(SessionState::Running, false, vec![]), start); + + let parked = watched( + SessionState::AwaitingInput, + false, + vec![SessionEvent::ToolApprovalRequest { + call_id: "c1".into(), + tool: "bash".into(), + args_summary: "cargo test".into(), + risk: construct_protocol::ToolRisk::Risky, + allow_auto_review: true, + }], + ); + assert!(!watch.settled_without_answer(&parked, start + IDLE_SETTLE * 10)); + } + + #[test] + fn input_still_queued_means_the_turn_has_not_started() { + // Between delivery and pickup the session is idle with work waiting. + // That is the normal opening moment of every turn, not a failure. + let mut watch = TurnWatch::default(); + let start = tokio::time::Instant::now(); + let queued = watched(SessionState::AwaitingInput, true, vec![]); + assert!(!watch.settled_without_answer(&queued, start)); + assert!(!watch.settled_without_answer(&queued, start + NEVER_STARTED_GRACE * 2)); + } + + #[test] + fn a_turn_never_seen_running_gets_the_longer_grace() { + // A harness slow to pick input up must not be declared broken at the + // ten-second mark — but it must not hold the caller forever either. + let mut watch = TurnWatch::default(); + let start = tokio::time::Instant::now(); + let idle = watched(SessionState::AwaitingInput, false, vec![]); + assert!(!watch.settled_without_answer(&idle, start)); + assert!(!watch.settled_without_answer(&idle, start + IDLE_SETTLE * 2)); + assert!(watch.settled_without_answer(&idle, start + NEVER_STARTED_GRACE)); + } + + #[test] + fn a_failed_turn_is_reported_in_the_harnesss_own_words() { + use base64::Engine as _; + let detail = watched( + SessionState::AwaitingInput, + false, + vec![SessionEvent::Pty { + data: base64::engine::general_purpose::STANDARD.encode( + "\x1b[K■ stream disconnected before completion\x1b[K› ask me anything", + ), + }], + ); + assert_eq!( + turn_failed_error(&detail).to_string(), + "stream disconnected before completion" + ); + + // A harness that left nothing quotable still gets a straight answer, + // not silence and not a lie about what happened. + let bare = watched(SessionState::AwaitingInput, false, vec![]); + assert_eq!( + turn_failed_error(&bare).to_string(), + "the session stopped without replying" + ); + } } diff --git a/crates/daemon/src/service/slack.rs b/crates/daemon/src/service/slack.rs index b7fd2e69..24667c13 100644 --- a/crates/daemon/src/service/slack.rs +++ b/crates/daemon/src/service/slack.rs @@ -340,20 +340,40 @@ where /// that has already become long enough to look like a dropped request. const PROGRESS_AFTER: std::time::Duration = std::time::Duration::from_secs(8); +/// How often the placeholder is rewritten so its elapsed time stays current. +/// +/// A minute is slow enough to be nowhere near Slack's edit rate limits and +/// fast enough that the number on screen is never meaningfully stale. +const PROGRESS_REFRESH: std::time::Duration = std::time::Duration::from_secs(60); + const WORKING_EMOJI: &str = "eyes"; const ANSWERED_EMOJI: &str = "white_check_mark"; const FAILED_EMOJI: &str = "warning"; -fn progress_text(progress: &IngressProgress) -> String { +/// How long this has been going, for the placeholder to admit to. +/// +/// Below a minute there is nothing worth saying — the affordance has barely +/// appeared. Past that, a number that visibly moves is the difference between +/// a wait someone can sit through and one that reads as a dropped request. +fn elapsed_suffix(elapsed: std::time::Duration) -> String { + let minutes = elapsed.as_secs() / 60; + if minutes == 0 { + return String::new(); + } + format!(" ({minutes}m)") +} + +fn progress_text(progress: &IngressProgress, elapsed: std::time::Duration) -> String { + let waited = elapsed_suffix(elapsed); match progress { - IngressProgress::Working => "_Working on it…_".to_string(), + IngressProgress::Working => format!("_Working on it…{waited}_"), // Say who has to act. A turn stopped here will not move on its own, // and the person waiting in Slack cannot see the approval prompt. IngressProgress::AwaitingApproval { tool, summary } if summary.is_empty() => { - format!("_Waiting for an operator to approve `{tool}`._") + format!("_Waiting for an operator to approve `{tool}`.{waited}_") } IngressProgress::AwaitingApproval { tool, summary } => { - format!("_Waiting for an operator to approve `{tool}`: {summary}_") + format!("_Waiting for an operator to approve `{tool}`: {summary}{waited}_") } } } @@ -383,6 +403,10 @@ async fn run_affordance( if config.progress == SlackProgress::Off { return state; } + // Measured from when the turn was submitted, not from when the placeholder + // appeared: the elapsed time this reports is the wait the person in Slack + // has actually had, which started when they hit send. + let started = tokio::time::Instant::now(); tokio::select! { _ = cancel.cancelled() => return state, _ = tokio::time::sleep(after) => {} @@ -407,7 +431,7 @@ async fn run_affordance( } } if config.progress.posts_placeholder() { - let text = progress_text(&progress.borrow_and_update().clone()); + let text = progress_text(&progress.borrow_and_update().clone(), started.elapsed()); match api .post_message(&config.bot_token, &channel, &thread_ts, &text) .await @@ -416,25 +440,32 @@ async fn run_affordance( Err(error) => tracing::warn!(%error, "Slack progress placeholder failed"), } } - // Keep the placeholder honest: a turn that stops at an approval must stop - // claiming it is working. + // Keep the placeholder honest on both counts: a turn that stops at an + // approval must stop claiming it is working, and a placeholder that has + // said the same words for twenty minutes is indistinguishable from one + // left behind by a turn nobody is waiting on any more. + let mut refresh = tokio::time::interval(PROGRESS_REFRESH); + refresh.tick().await; loop { - tokio::select! { + let phase = tokio::select! { _ = cancel.cancelled() => return state, changed = progress.changed() => { if changed.is_err() { return state; } - let text = progress_text(&progress.borrow_and_update().clone()); - let Some(ts) = state.placeholder_ts.as_deref() else { - continue; - }; - if let Err(error) = - api.update_message(&config.bot_token, &channel, ts, &text).await - { - tracing::warn!(%error, "Slack progress update failed"); - } + progress.borrow_and_update().clone() } + _ = refresh.tick() => progress.borrow().clone(), + }; + let Some(ts) = state.placeholder_ts.as_deref() else { + continue; + }; + let text = progress_text(&phase, started.elapsed()); + if let Err(error) = api + .update_message(&config.bot_token, &channel, ts, &text) + .await + { + tracing::warn!(%error, "Slack progress update failed"); } } } @@ -1035,26 +1066,69 @@ mod tests { // "Working on it" is a lie once the turn is parked on an approval: // nothing moves until a human at the TUI acts, and the person waiting // in Slack cannot see that prompt. + let fresh = std::time::Duration::ZERO; assert_eq!( - progress_text(&IngressProgress::Working), + progress_text(&IngressProgress::Working, fresh), "_Working on it…_" ); assert_eq!( - progress_text(&IngressProgress::AwaitingApproval { - tool: "bash".into(), - summary: "cargo test".into(), - }), + progress_text( + &IngressProgress::AwaitingApproval { + tool: "bash".into(), + summary: "cargo test".into(), + }, + fresh + ), "_Waiting for an operator to approve `bash`: cargo test_" ); assert_eq!( - progress_text(&IngressProgress::AwaitingApproval { - tool: "bash".into(), - summary: String::new(), - }), + progress_text( + &IngressProgress::AwaitingApproval { + tool: "bash".into(), + summary: String::new(), + }, + fresh + ), "_Waiting for an operator to approve `bash`._" ); } + #[test] + fn a_long_wait_says_how_long_it_has_been() { + // The affordance exists for waits long enough to look dropped. One + // that never changes reads as abandoned no matter what it says, so + // past the first minute the placeholder carries a number that moves. + let minutes = |n: u64| std::time::Duration::from_secs(n * 60); + assert_eq!( + progress_text(&IngressProgress::Working, minutes(0)), + "_Working on it…_" + ); + assert_eq!( + progress_text(&IngressProgress::Working, std::time::Duration::from_secs(59)), + "_Working on it…_" + ); + assert_eq!( + progress_text(&IngressProgress::Working, minutes(1)), + "_Working on it… (1m)_" + ); + assert_eq!( + progress_text(&IngressProgress::Working, minutes(17)), + "_Working on it… (17m)_" + ); + // An approval that nobody has answered needs the elapsed time most of + // all: it is the case that will never resolve on its own. + assert_eq!( + progress_text( + &IngressProgress::AwaitingApproval { + tool: "bash".into(), + summary: String::new(), + }, + minutes(9) + ), + "_Waiting for an operator to approve `bash`. (9m)_" + ); + } + #[test] fn progress_modes_select_their_affordances() { assert!(!SlackProgress::Off.posts_placeholder() && !SlackProgress::Off.reacts()); diff --git a/specs/0178-a-channel-shows-that-a-turn-is-still-running.md b/specs/0178-a-channel-shows-that-a-turn-is-still-running.md index a2449e05..88274af7 100644 --- a/specs/0178-a-channel-shows-that-a-turn-is-still-running.md +++ b/specs/0178-a-channel-shows-that-a-turn-is-still-running.md @@ -12,12 +12,17 @@ from one that dropped it. A channel may therefore show that a turn is still running, and the operator chooses how visible that is per channel — including turning it off. -Three rules constrain it. +Four rules constrain it. **Silence is correct for a turn that answers promptly.** The affordance exists for a wait that has already become long enough to look like a failure, so it appears only after such a wait. A quick turn leaves no trace of one. +**It must keep looking alive.** An affordance whose words have not changed in +twenty minutes reads as abandoned, which is the impression it exists to +prevent. Past the point where the wait is worth remarking on, it carries +something that visibly advances — how long this has been going. + **It must not claim to be working when it is not.** A turn stopped at a tool approval is not making progress and will not resume until a human acts at another surface entirely. Whatever the channel is showing has to change to say diff --git a/specs/0181-a-turn-that-stops-without-answering-is-a-failure.md b/specs/0181-a-turn-that-stops-without-answering-is-a-failure.md new file mode 100644 index 00000000..c05a3c10 --- /dev/null +++ b/specs/0181-a-turn-that-stops-without-answering-is-a-failure.md @@ -0,0 +1,89 @@ +# 0181-a-turn-that-stops-without-answering-is-a-failure + +Status: accepted +Date: 2026-08-02 +Area: architecture +Scope: How a service delivery learns that the turn behind it has ended without producing an answer, and what it reports when that happens. + +## Decision + +A turn that has stopped is finished, whether or not it answered. A delivery +waiting on one must reach that conclusion by **observing the session**, not by +waiting for the harness to declare a failure. + +A session is treated as having stopped without answering when, continuously +and for long enough to be sure, it is idle, has no input queued, is not parked +at a tool approval, and is appending nothing to its transcript. Reaching that +state without the delivery's answer ends the wait as a failure. + +Two constraints on the observation: + +**Only progress counts as activity.** A harness with a live terminal repaints +constantly, including while idle. Redraws must not read as work, or a session +that draws its own cursor will never look stopped. + +**Not-yet-started is not stopped.** A session is briefly idle between receiving +input and picking it up. A turn never observed running gets a longer grace than +one that ran and came back. + +When a turn does fail, what the channel reports is **the harness's own words +where it left any** — including text it only drew on its terminal. Failing +that, it says plainly that the session stopped without replying. It never +reports a timeout for a turn that did not time out. + +## Reason + +The daemon does not run the model. It supervises a harness, and a harness that +loses its upstream mid-turn is under no obligation to say so in a way the +daemon can read: an interactive one prints the failure into its viewport and +returns to its composer. From outside, that is a session sitting at "awaiting +input" — the same shape as a turn that finished normally, differing only in +having produced nothing. + +Every failure signal that waits to be *told* therefore misses this case, and +what is left is the delivery's own expiry. That is a wait measured in tens of +minutes, ended by a message that names the timeout rather than the cause, while +the actual explanation sat legible on a screen the whole time. Observing the +session instead collapses that to seconds and lets the report say what broke. + +Quoting the harness is what makes the report worth reading. "The turn ended +without an answer" tells the person waiting only that they are still stuck; +"stream disconnected before completion" tells them it was the network and that +sending the message again is a reasonable thing to do. + +## Consequences + +- Idle, quiet, unqueued, and unparked must **all** hold, continuously, before a + turn is called over. Any one of them alone will misfire on a normal turn. +- Approval parking is load-bearing here: it is the one stop that is supposed to + last, and it must keep reading as waiting rather than as failure. +- Recovering error text from a terminal is best-effort and is only ever + consulted for a turn already known to have failed. A wrong guess costs a + slightly-off sentence in a failure notice; it can never cost an answer or + turn a success into a failure. +- The delivery expiry stops being the mechanism that ends a failed wait and + goes back to being a backstop. +- Failure reports carry harness text, which is written by the harness and may + be long or oddly formatted. It is bounded before it is quoted. + +## Non-Goals + +- Distinguishing *kinds* of failure, or retrying one. The delivery reports what + happened and stops. +- Parsing harness terminals for anything other than a failure notice. +- Making a harness that fails silently start reporting properly. This observes + from outside precisely because that cannot be relied upon. + +## Examples + +An interactive session loses its connection to the model provider ten minutes +into a turn, prints a two-line error, and returns to its prompt. Within seconds +the waiting channel stops saying it is working and posts the harness's error +instead, naming the endpoint that failed. + +A turn runs for six minutes across a dozen tool calls with pauses between them. +None of those pauses ends the wait, because each one is interrupted by +something new appearing in the transcript. + +A turn stops at an approval and stays there for an hour. It is never reported +as failed; the channel keeps saying an operator has to act.