diff --git a/src-tauri/src/router/mod.rs b/src-tauri/src/router/mod.rs index ddb9ef6..cf19c12 100644 --- a/src-tauri/src/router/mod.rs +++ b/src-tauri/src/router/mod.rs @@ -139,7 +139,7 @@ impl StdinInjector for SessionManager { #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum DeliveryReservation { Ready(u64), - PendingInput, + PendingInput(Duration), RecentlyTyping(Duration), InFlight, } @@ -698,7 +698,7 @@ impl Router { } (session_id, None) } - DeliveryReservation::PendingInput => { + DeliveryReservation::PendingInput(delay) => { if reconciliation.is_some() { return Ok(false); } @@ -710,6 +710,7 @@ impl Router { outbox.handle = handle.to_string(); outbox.pending_input_blocked = true; outbox.enqueue(delivery); + retry = Some((session_id.clone(), delay)); self.blocked_transition(&mut state, &session_id); } (session_id, None) @@ -1035,13 +1036,16 @@ impl Router { } self.schedule_outbox_retry(session_id.to_string(), delay); } - DeliveryReservation::PendingInput => { - let mut state = self.state.lock().unwrap(); - let Some(outbox) = state.outbox_by_session.get_mut(session_id) else { - return; - }; - outbox.pending_input_blocked = true; - self.blocked_transition(&mut state, session_id); + DeliveryReservation::PendingInput(delay) => { + { + let mut state = self.state.lock().unwrap(); + let Some(outbox) = state.outbox_by_session.get_mut(session_id) else { + return; + }; + outbox.pending_input_blocked = true; + self.blocked_transition(&mut state, session_id); + } + self.schedule_outbox_retry(session_id.to_string(), delay); } DeliveryReservation::InFlight => { let mut state = self.state.lock().unwrap(); diff --git a/src-tauri/src/router/tests.rs b/src-tauri/src/router/tests.rs index 3ead2b8..87688b2 100644 --- a/src-tauri/src/router/tests.rs +++ b/src-tauri/src/router/tests.rs @@ -48,6 +48,7 @@ struct RecordingInjector { #[derive(Default)] struct RecordingInputState { pending: bool, + pending_until: Option, last_input_at: Option, in_flight: bool, generation: u64, @@ -122,9 +123,14 @@ impl RecordingInjector { } fn set_pending(&self, session_id: &str) { + self.set_pending_for(session_id, Duration::from_secs(10 * 60)); + } + + fn set_pending_for(&self, session_id: &str, duration: Duration) { let mut input = self.input.lock().unwrap(); let input = input.entry(session_id.to_string()).or_default(); input.pending = true; + input.pending_until = Some(Instant::now() + duration); input.last_input_at = Some(Instant::now()); } @@ -132,6 +138,7 @@ impl RecordingInjector { let mut input = self.input.lock().unwrap(); let input = input.entry(session_id.to_string()).or_default(); input.pending = false; + input.pending_until = None; input.last_input_at = Some(Instant::now() - Duration::from_millis(1950)); } @@ -147,6 +154,7 @@ impl RecordingInjector { fn clear_pending(&self, session_id: &str) { if let Some(input) = self.input.lock().unwrap().get_mut(session_id) { input.pending = false; + input.pending_until = None; input.last_input_at = None; } self.notify(session_id, SessionDeliveryEvent::InputCleared); @@ -160,6 +168,7 @@ impl RecordingInjector { input.generation = input.generation.wrapping_add(1); input.in_flight = false; input.pending = false; + input.pending_until = None; input.last_input_at = None; } self.notify(session_id, SessionDeliveryEvent::Respawned); @@ -171,6 +180,7 @@ impl RecordingInjector { input.generation = input.generation.wrapping_add(1); input.in_flight = false; input.pending = false; + input.pending_until = None; input.last_input_at = None; } self.notify(session_id, SessionDeliveryEvent::Exited); @@ -276,7 +286,16 @@ impl StdinInjector for RecordingInjector { return Ok(DeliveryReservation::InFlight); } if input.pending { - return Ok(DeliveryReservation::PendingInput); + let remaining = input + .pending_until + .expect("pending test input must have an abandonment deadline") + .saturating_duration_since(Instant::now()); + if !remaining.is_zero() { + return Ok(DeliveryReservation::PendingInput(remaining)); + } + input.pending = false; + input.pending_until = None; + input.last_input_at = None; } if let Some(last) = input.last_input_at { let elapsed = last.elapsed(); @@ -1124,6 +1143,36 @@ fn recent_typing_retries_after_quiet_window() { }); } +#[test] +fn pending_input_retries_after_abandonment_deadline() { + let (router, injector, _log, _dir) = fixture( + vec![ + slot_with_runner("lead", true), + slot_with_runner("impl", false), + ], + &[("lead", "S-LEAD"), ("impl", "S-IMPL")], + ); + injector.set_pending_for("S-IMPL", Duration::from_millis(25)); + router + .inject_and_submit("impl", b"after abandonment") + .unwrap(); + assert!(injector.pushes_for("S-IMPL").is_empty()); + assert!( + router + .state + .lock() + .unwrap() + .outbox_by_session + .get("S-IMPL") + .is_some_and(|outbox| outbox.retry_scheduled), + "a draft-gated outbox must always have a scheduled retry" + ); + + wait_until(Duration::from_millis(300), || { + injector.submitted_bodies_for("S-IMPL") == ["after abandonment"] + }); +} + #[test] fn deferred_nudges_coalesce_while_relays_preserve_order() { let (router, injector, log, _dir) = fixture( diff --git a/src-tauri/src/session/manager/lifecycle.rs b/src-tauri/src/session/manager/lifecycle.rs index 949003e..29206da 100644 --- a/src-tauri/src/session/manager/lifecycle.rs +++ b/src-tauri/src/session/manager/lifecycle.rs @@ -73,7 +73,7 @@ impl SessionManager { gate.ready.notify_all(); state.activity = None; state.suppress_local_input_busy = false; - state.local_input_pending = false; + state.draft.clear(); state.last_local_input_at = None; state.mission_status_sink = None; state.completion_armed = false; @@ -312,7 +312,7 @@ impl SessionManager { state.handle = None; state.activity = None; state.suppress_local_input_busy = false; - state.local_input_pending = false; + state.draft.clear(); state.last_local_input_at = None; state.mission_status_sink = None; state.completion_armed = false; diff --git a/src-tauri/src/session/manager/mod.rs b/src-tauri/src/session/manager/mod.rs index 36525ce..fd357f6 100644 --- a/src-tauri/src/session/manager/mod.rs +++ b/src-tauri/src/session/manager/mod.rs @@ -49,6 +49,8 @@ mod tests; const MAX_OUTPUT_BUFFER_CHUNKS: usize = 4096; const RECENT_LOCAL_INPUT_WINDOW: Duration = Duration::from_secs(2); +const DRAFT_MAX_BYTES: usize = 4 * 1024; +const DRAFT_ABANDON_WINDOW: Duration = Duration::from_secs(10 * 60); pub(crate) const DEFAULT_PTY_SIZE: (u16, u16) = (80, 24); /// How long a cols-change resize storm must stay quiet before the @@ -557,12 +559,130 @@ struct PendingResize { events: Arc, } +#[derive(Clone, Debug, Default, PartialEq, Eq)] +struct DraftState { + bytes: Vec, + /// Once the cap is reached, deletion cannot prove what remains beyond it. + saturated: bool, + /// History recall materializes unknown text, so only an explicit clear can reopen the gate. + opaque: bool, +} + +impl DraftState { + fn is_empty(&self) -> bool { + self.bytes.is_empty() + } + + fn append(&mut self, bytes: &[u8]) -> bool { + if self.saturated || bytes.is_empty() { + return false; + } + let available = DRAFT_MAX_BYTES - self.bytes.len(); + self.bytes + .extend_from_slice(&bytes[..bytes.len().min(available)]); + if self.bytes.len() == DRAFT_MAX_BYTES { + self.saturated = true; + return true; + } + false + } + + fn pop_char(&mut self) { + if self.saturated || self.opaque || self.bytes.is_empty() { + return; + } + let mut new_len = self.bytes.len() - 1; + while new_len > 0 && self.bytes[new_len] & 0b1100_0000 == 0b1000_0000 { + new_len -= 1; + } + self.bytes.truncate(new_len); + } + + fn pop_word(&mut self) { + if self.saturated || self.opaque { + return; + } + while self + .bytes + .last() + .is_some_and(|byte| byte.is_ascii_whitespace()) + { + self.pop_char(); + } + while self + .bytes + .last() + .is_some_and(|byte| !byte.is_ascii_whitespace()) + { + self.pop_char(); + } + } + + fn clear(&mut self) { + self.bytes.clear(); + self.saturated = false; + self.opaque = false; + } + + fn mark_opaque(&mut self, marker: &[u8]) { + if self.bytes.is_empty() { + self.append(marker); + } + self.opaque = true; + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum DraftGateCause { + Printable, + Paste, + HistoryRecall, + BackspaceEmptied, + Esc, + Submit, + Interrupt, + Abandoned, + ManualClear, + Saturated, +} + +impl DraftGateCause { + fn as_str(self) -> &'static str { + match self { + Self::Printable => "printable", + Self::Paste => "paste", + Self::HistoryRecall => "history-recall", + Self::BackspaceEmptied => "backspace-emptied", + Self::Esc => "esc", + Self::Submit => "submit", + Self::Interrupt => "interrupt", + Self::Abandoned => "abandoned", + Self::ManualClear => "manual-clear", + Self::Saturated => "saturated", + } + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +struct DraftGateEvent { + blocked: bool, + cause: DraftGateCause, +} + +fn log_draft_gate_event(session_id: &str, event: DraftGateEvent) { + let state = if event.blocked { "blocked" } else { "open" }; + log::info!( + "[draft-gate] session={session_id} state={state} cause={}", + event.cause.as_str() + ); +} + #[derive(Default)] struct SessionState { handle: Option, activity: Option, suppress_local_input_busy: bool, - local_input_pending: bool, + draft: DraftState, last_local_input_at: Option, delivery_gate: Arc, mission_status_sink: Option, @@ -605,7 +725,7 @@ impl SessionState { self.handle.is_none() && self.activity.is_none() && !self.suppress_local_input_busy - && !self.local_input_pending + && self.draft.is_empty() && self.last_local_input_at.is_none() && self.mission_status_sink.is_none() && !self.completion_armed @@ -669,6 +789,10 @@ pub struct SessionManager { /// in milliseconds. A field rather than a const so tests can pin /// it high (deterministic manual settle) or low (thread wiring). resize_settle_ms: AtomicU64, + /// How long an untouched draft blocks delivery. Production uses + /// `DRAFT_ABANDON_WINDOW`; tests pin this low to exercise the lazy + /// reserve-time backstop without waiting ten minutes. + draft_abandon_ms: AtomicU64, } /// RAII guard that releases a session state's `resuming` flag on drop. The @@ -772,6 +896,7 @@ impl SessionManager { pending_mission_cancels: Mutex::new(HashMap::new()), runtime, resize_settle_ms: AtomicU64::new(RESIZE_SETTLE_MS), + draft_abandon_ms: AtomicU64::new(DRAFT_ABANDON_WINDOW.as_millis() as u64), }) } @@ -780,6 +905,11 @@ impl SessionManager { self.resize_settle_ms.store(ms, Ordering::Relaxed); } + #[cfg(test)] + pub(crate) fn set_draft_abandon_ms(&self, ms: u64) { + self.draft_abandon_ms.store(ms, Ordering::Relaxed); + } + fn session_state(&self, session_id: &str) -> Option>> { self.sessions.lock().unwrap().get(session_id).cloned() } @@ -843,7 +973,7 @@ impl SessionManager { session.handle.is_some() && !delivery.in_flight && delivery.next_ticket == delivery.next_served - && !session.local_input_pending + && session.draft.is_empty() && session .last_local_input_at .is_none_or(|last| last.elapsed() >= RECENT_LOCAL_INPUT_WINDOW) @@ -860,7 +990,7 @@ impl SessionManager { .ok_or_else(|| Error::msg(format!("session not found: {session_id}")))?; let gate = session.lock().unwrap().delivery_gate.clone(); let mut delivery = gate.state.lock().unwrap(); - let session = session.lock().unwrap(); + let mut session = session.lock().unwrap(); if session.handle.is_none() { return Err(Error::msg(format!("session not found: {session_id}"))); } @@ -870,8 +1000,28 @@ impl SessionManager { if delivery.next_ticket != delivery.next_served { return Ok(router::DeliveryReservation::InFlight); } - if session.local_input_pending { - return Ok(router::DeliveryReservation::PendingInput); + if !session.draft.is_empty() { + let abandon_window = + Duration::from_millis(self.draft_abandon_ms.load(Ordering::Relaxed)); + let Some(last) = session.last_local_input_at else { + session.last_local_input_at = Some(Instant::now()); + return Ok(router::DeliveryReservation::PendingInput(abandon_window)); + }; + let elapsed = last.elapsed(); + if elapsed < abandon_window { + return Ok(router::DeliveryReservation::PendingInput( + abandon_window - elapsed, + )); + } + session.draft.clear(); + session.last_local_input_at = None; + log_draft_gate_event( + session_id, + DraftGateEvent { + blocked: false, + cause: DraftGateCause::Abandoned, + }, + ); } if let Some(last) = session.last_local_input_at { let elapsed = last.elapsed(); @@ -933,7 +1083,7 @@ impl SessionManager { delivery.next_ticket = 0; delivery.next_served = 0; gate.ready.notify_all(); - state.local_input_pending = false; + state.draft.clear(); state.last_local_input_at = None; state.handle = Some(handle); state.mission_status_sink = mission_status_sink; diff --git a/src-tauri/src/session/manager/output.rs b/src-tauri/src/session/manager/output.rs index a355df3..32f90b4 100644 --- a/src-tauri/src/session/manager/output.rs +++ b/src-tauri/src/session/manager/output.rs @@ -7,57 +7,224 @@ pub(super) const KEEP_RESUME_SEAM: &[u8] = b"\x1b[0m\x1b[?2004l\x1b[?1000l\x1b[?1002l\x1b[?1003l\x1b[?1006l\r\n"; #[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub(super) enum LocalInputClass { - SetPending, - ClearPending, +pub(super) enum DraftOperation<'a> { + Append { + bytes: &'a [u8], + cause: DraftGateCause, + }, + Backspace, + DeleteWord, + Clear(DraftGateCause), + MarkNonEmpty(DraftGateCause), ActivityOnly, } -pub(super) fn classify_local_input(bytes: &[u8]) -> Option { - if bytes.is_empty() { - return None; - } - if bytes == b"\r" || bytes == b"\x03" { - return Some(LocalInputClass::ClearPending); +#[derive(Debug, Default, PartialEq, Eq)] +pub(super) struct LocalInputUpdate { + pub(super) input_cleared: bool, + pub(super) events: Vec, +} + +const BRACKETED_PASTE_START: &[u8] = b"\x1b[200~"; +const BRACKETED_PASTE_END: &[u8] = b"\x1b[201~"; +const DRAFT_CONTENT_MARKER: &[u8] = b"?"; +const DRAFT_NEWLINE_MARKER: &[u8] = b"\n"; + +fn find_sequence(bytes: &[u8], needle: &[u8]) -> Option { + bytes + .windows(needle.len()) + .position(|window| window == needle) +} + +fn csi_sequence_end(bytes: &[u8], start: usize) -> usize { + let mut cursor = start + 2; + while cursor < bytes.len() { + if matches!(bytes[cursor], 0x40..=0x7e) { + return cursor + 1; + } + cursor += 1; } - if bytes == b"\x16" || bytes.starts_with(b"\x1b[200~") { - return Some(LocalInputClass::SetPending); + bytes.len() +} + +pub(super) fn classify_local_input(bytes: &[u8]) -> Vec> { + if bytes.is_empty() { + return Vec::new(); } - if bytes.starts_with(b"\x1b") { - return Some(LocalInputClass::ActivityOnly); + if bytes == b"\x1b" { + return vec![DraftOperation::Clear(DraftGateCause::Esc)]; } - if bytes - .iter() - .any(|byte| matches!(byte, 0x20..=0x7e | 0x80..=0xff)) - { - Some(LocalInputClass::SetPending) - } else { - Some(LocalInputClass::ActivityOnly) + + let mut operations = Vec::new(); + let mut index = 0; + while index < bytes.len() { + let remaining = &bytes[index..]; + if remaining.starts_with(BRACKETED_PASTE_START) { + let payload_start = index + BRACKETED_PASTE_START.len(); + let payload_and_rest = &bytes[payload_start..]; + if let Some(end_offset) = find_sequence(payload_and_rest, BRACKETED_PASTE_END) { + let payload_end = payload_start + end_offset; + operations.push(DraftOperation::Append { + bytes: if payload_end == payload_start { + DRAFT_CONTENT_MARKER + } else { + &bytes[payload_start..payload_end] + }, + cause: DraftGateCause::Paste, + }); + index = payload_end + BRACKETED_PASTE_END.len(); + } else { + operations.push(DraftOperation::Append { + bytes: if payload_start == bytes.len() { + DRAFT_CONTENT_MARKER + } else { + &bytes[payload_start..] + }, + cause: DraftGateCause::Paste, + }); + index = bytes.len(); + } + continue; + } + if remaining.starts_with(b"\x1b\r") { + operations.push(DraftOperation::Append { + bytes: DRAFT_NEWLINE_MARKER, + cause: DraftGateCause::Printable, + }); + index += 2; + continue; + } + if remaining.starts_with(b"\x1b[") { + let end = csi_sequence_end(bytes, index); + let sequence = &bytes[index..end]; + if sequence == b"\x1b[A" || sequence == b"\x1b[B" { + operations.push(DraftOperation::MarkNonEmpty(DraftGateCause::HistoryRecall)); + } else { + operations.push(DraftOperation::ActivityOnly); + } + index = end; + continue; + } + if remaining.starts_with(b"\x1bO") { + let end = (index + 3).min(bytes.len()); + let sequence = &bytes[index..end]; + if sequence == b"\x1bOA" || sequence == b"\x1bOB" { + operations.push(DraftOperation::MarkNonEmpty(DraftGateCause::HistoryRecall)); + } else { + operations.push(DraftOperation::ActivityOnly); + } + index = end; + continue; + } + if bytes[index] == 0x1b { + operations.push(DraftOperation::ActivityOnly); + index = (index + 2).min(bytes.len()); + continue; + } + match bytes[index] { + b'\r' => { + operations.push(DraftOperation::Clear(DraftGateCause::Submit)); + index += 1; + } + 0x03 => { + operations.push(DraftOperation::Clear(DraftGateCause::Interrupt)); + index += 1; + } + 0x15 => { + operations.push(DraftOperation::Clear(DraftGateCause::ManualClear)); + index += 1; + } + 0x7f | 0x08 => { + operations.push(DraftOperation::Backspace); + index += 1; + } + 0x17 => { + operations.push(DraftOperation::DeleteWord); + index += 1; + } + 0x16 => { + operations.push(DraftOperation::Append { + bytes: DRAFT_CONTENT_MARKER, + cause: DraftGateCause::Paste, + }); + index += 1; + } + 0x00..=0x1f => { + operations.push(DraftOperation::ActivityOnly); + index += 1; + } + _ => { + let start = index; + index += 1; + while index < bytes.len() && !matches!(bytes[index], 0x00..=0x1f | 0x7f) { + index += 1; + } + operations.push(DraftOperation::Append { + bytes: &bytes[start..index], + cause: DraftGateCause::Printable, + }); + } + } } + operations } pub(super) fn update_local_input_state( state: &mut SessionState, - input_class: Option, + operations: &[DraftOperation<'_>], now: Instant, -) -> bool { - match input_class { - Some(LocalInputClass::SetPending) => { - state.local_input_pending = true; - state.last_local_input_at = Some(now); - false - } - Some(LocalInputClass::ClearPending) => { - state.local_input_pending = false; - state.last_local_input_at = None; - true +) -> LocalInputUpdate { + let mut update = LocalInputUpdate::default(); + for operation in operations { + let was_empty = state.draft.is_empty(); + let mut saturated = false; + let cause = match operation { + DraftOperation::Append { bytes, cause } => { + saturated = state.draft.append(bytes); + state.last_local_input_at = Some(now); + *cause + } + DraftOperation::Backspace => { + state.draft.pop_char(); + state.last_local_input_at = Some(now); + DraftGateCause::BackspaceEmptied + } + DraftOperation::DeleteWord => { + state.draft.pop_word(); + state.last_local_input_at = Some(now); + DraftGateCause::BackspaceEmptied + } + DraftOperation::Clear(cause) => { + state.draft.clear(); + state.last_local_input_at = None; + *cause + } + DraftOperation::MarkNonEmpty(cause) => { + state.draft.mark_opaque(DRAFT_CONTENT_MARKER); + state.last_local_input_at = Some(now); + *cause + } + DraftOperation::ActivityOnly => { + state.last_local_input_at = Some(now); + continue; + } + }; + let is_empty = state.draft.is_empty(); + if was_empty != is_empty { + update.events.push(DraftGateEvent { + blocked: !is_empty, + cause, + }); + update.input_cleared |= is_empty; } - Some(LocalInputClass::ActivityOnly) => { - state.last_local_input_at = Some(now); - false + if saturated { + update.events.push(DraftGateEvent { + blocked: true, + cause: DraftGateCause::Saturated, + }); } - None => false, } + update } impl SessionManager { @@ -297,7 +464,7 @@ impl SessionManager { events: &dyn SessionEvents, ) -> Result<()> { let submitted = bytes == b"\r"; - let input_class = classify_local_input(bytes); + let draft_operations = classify_local_input(bytes); let session = self .session_state(session_id) .ok_or_else(|| Error::msg(format!("session not found: {session_id}")))?; @@ -326,7 +493,7 @@ impl SessionManager { .ok_or_else(|| Error::msg(format!("session not found: {session_id}")))?; let previous_activity = session.activity; let previous_suppression = session.suppress_local_input_busy; - let previous_input_pending = session.local_input_pending; + let previous_draft = session.draft.clone(); let previous_input_at = session.last_local_input_at; let mission_status_sink = session.mission_status_sink.clone(); let mission_scoped = session @@ -351,11 +518,12 @@ impl SessionManager { } None }; - let input_cleared = update_local_input_state(&mut session, input_class, Instant::now()); + let draft_update = + update_local_input_state(&mut session, &draft_operations, Instant::now()); if let Err(error) = self.write_stdin_bytes(&rt_session, bytes) { session.activity = previous_activity; session.suppress_local_input_busy = previous_suppression; - session.local_input_pending = previous_input_pending; + session.draft = previous_draft; session.last_local_input_at = previous_input_at; return Err(error); } @@ -366,7 +534,7 @@ impl SessionManager { transition, mission_status_sink, mission_scoped, - input_cleared, + draft_update, )) })(); @@ -374,17 +542,20 @@ impl SessionManager { let input_queue_drained = delivery.next_served == delivery.next_ticket; let successful_clear = outcome .as_ref() - .is_ok_and(|(_, _, _, input_cleared)| *input_cleared); + .is_ok_and(|(_, _, _, update)| update.input_cleared); gate.ready.notify_all(); drop(delivery); if input_queue_drained && !successful_clear { self.notify_delivery_event(session_id, router::SessionDeliveryEvent::InputQueueDrained); } - let (transition, mission_status_sink, mission_scoped, input_cleared) = outcome?; + let (transition, mission_status_sink, mission_scoped, draft_update) = outcome?; if submitted { self.capture_codex_session_key(session_id); } - if input_cleared { + for event in draft_update.events { + log_draft_gate_event(session_id, event); + } + if draft_update.input_cleared { self.notify_delivery_event(session_id, router::SessionDeliveryEvent::InputCleared); } if let Some(transition) = transition.as_ref() { @@ -857,6 +1028,8 @@ impl SessionManager { state.mouse_1002_on = false; state.mouse_1003_on = false; state.mouse_1006_on = false; + state.draft.clear(); + state.last_local_input_at = None; state.last_pty_cols = None; state.pending_resize = None; } diff --git a/src-tauri/src/session/manager/tests.rs b/src-tauri/src/session/manager/tests.rs index ad80a93..06a1c17 100644 --- a/src-tauri/src/session/manager/tests.rs +++ b/src-tauri/src/session/manager/tests.rs @@ -62,6 +62,7 @@ fn inert_runtime() -> Arc { struct FakeRuntime { spawns: std::sync::Mutex>, inputs: std::sync::Mutex>, + input_failures: std::sync::Mutex>, stops: std::sync::Mutex>, stop_failures: std::sync::Mutex>, resizes: std::sync::Mutex>, @@ -162,6 +163,13 @@ impl FakeRuntime { .insert(session_id.to_string()); } + fn fail_input_for(&self, session_id: &str) { + self.input_failures + .lock() + .unwrap() + .insert(session_id.to_string()); + } + fn allow_stop_for(&self, session_id: &str) { self.stop_failures.lock().unwrap().remove(session_id); } @@ -265,6 +273,17 @@ impl SessionRuntime for FakeRuntime { } fn send_bytes(&self, session: &RuntimeSession, bytes: &[u8]) -> RuntimeResult<()> { + if self + .input_failures + .lock() + .unwrap() + .contains(&session.session_id) + { + return Err(RuntimeError::Msg(format!( + "injected input failure for {}", + session.session_id + ))); + } self.inputs.lock().unwrap().push(FakeInput::Bytes { session_id: session.session_id.clone(), bytes: bytes.to_vec(), @@ -273,6 +292,17 @@ impl SessionRuntime for FakeRuntime { } fn send_key(&self, session: &RuntimeSession, key: &str) -> RuntimeResult<()> { + if self + .input_failures + .lock() + .unwrap() + .contains(&session.session_id) + { + return Err(RuntimeError::Msg(format!( + "injected input failure for {}", + session.session_id + ))); + } self.inputs.lock().unwrap().push(FakeInput::Key { session_id: session.session_id.clone(), key: key.to_string(), @@ -335,6 +365,26 @@ fn mgr_with_fake(shell: Option, fake: Arc) -> Arc LocalInputUpdate { + let operations = classify_local_input(bytes); + update_local_input_state(state, &operations, now) + } + + let now = Instant::now(); + let mut state = SessionState::default(); + let typed = apply(&mut state, b"a", now); + assert!(!state.draft.is_empty()); + assert_eq!(state.last_local_input_at, Some(now)); assert_eq!( - classify_local_input(b"x"), - Some(LocalInputClass::SetPending) - ); - assert_eq!( - classify_local_input("界".as_bytes()), - Some(LocalInputClass::SetPending) - ); - assert_eq!( - classify_local_input(b"\x1b[A"), - Some(LocalInputClass::ActivityOnly) - ); - assert_eq!( - classify_local_input(b"\x1b[200~pasted text\x1b[201~"), - Some(LocalInputClass::SetPending) - ); - assert_eq!( - classify_local_input(b"\x16"), - Some(LocalInputClass::SetPending) - ); - assert_eq!( - classify_local_input(b"\r"), - Some(LocalInputClass::ClearPending) + typed.events, + [DraftGateEvent { + blocked: true, + cause: DraftGateCause::Printable, + }] ); + + let erased = apply(&mut state, b"\x7f", now); + assert!(state.draft.is_empty()); + assert_eq!(state.last_local_input_at, Some(now)); + assert!(erased.input_cleared); assert_eq!( - classify_local_input(b"\x03"), - Some(LocalInputClass::ClearPending) + erased.events, + [DraftGateEvent { + blocked: false, + cause: DraftGateCause::BackspaceEmptied, + }] + ); + + apply(&mut state, "界".as_bytes(), now); + apply(&mut state, b"\x08", now); + assert!(state.draft.is_empty(), "backspace removes one UTF-8 char"); + + apply(&mut state, b"one-word", now); + apply(&mut state, b"\x17", now); + assert!(state.draft.is_empty(), "Ctrl-W removes the pending word"); + + for clear in [b"\x15".as_slice(), b"\x1b", b"\x03"] { + apply(&mut state, b"draft", now); + assert!(!state.draft.is_empty()); + assert!(apply(&mut state, clear, now).input_cleared); + assert!(state.draft.is_empty()); + } + + for arrow in [b"\x1b[A".as_slice(), b"\x1b[B", b"\x1bOA", b"\x1bOB"] { + let mut arrow_state = SessionState::default(); + let update = apply(&mut arrow_state, arrow, now); + assert!(!arrow_state.draft.is_empty()); + assert_eq!( + update.events, + [DraftGateEvent { + blocked: true, + cause: DraftGateCause::HistoryRecall, + }] + ); + apply(&mut arrow_state, b"\x7f", now); + apply(&mut arrow_state, b"\x17", now); + assert!( + !arrow_state.draft.is_empty(), + "unknown recalled content must stay fail-closed under editing" + ); + apply(&mut arrow_state, b"\x15", now); + assert!(arrow_state.draft.is_empty()); + } + + for neutral in [ + b"\x1b[D".as_slice(), + b"\x1b[C", + b"\x1b[H", + b"\x1b[F", + b"\x1b[1~", + b"\x1b[4~", + b"\x1b[3~", + b"\x1bOC", + b"\x1bOD", + b"\x1bOP", + b"\x1bOQ", + b"\x1bOR", + b"\x1bOS", + b"\x1bx", + b"\x1b\x7f", + ] { + let mut empty = SessionState::default(); + apply(&mut empty, neutral, now); + assert!( + empty.draft.is_empty(), + "{neutral:?} must not invent content" + ); + assert_eq!(empty.last_local_input_at, Some(now)); + + apply(&mut empty, b"draft", now); + let before = empty.draft.clone(); + apply(&mut empty, neutral, now); + assert_eq!(empty.draft, before, "{neutral:?} must preserve a draft"); + } + + let mut multiline = SessionState::default(); + apply(&mut multiline, b"\x1b\r", now); + assert!(!multiline.draft.is_empty()); + + let mut pasted = SessionState::default(); + let paste_update = apply(&mut pasted, b"\x1b[200~first\r\nsecond\x1b[201~", now); + assert!(!pasted.draft.is_empty()); + assert_eq!(paste_update.events[0].cause, DraftGateCause::Paste); + apply(&mut pasted, b"\r", now); + assert!(pasted.draft.is_empty()); + + let mut image = SessionState::default(); + apply(&mut image, b"\x16", now); + assert!(!image.draft.is_empty()); + + let mut unknown = SessionState::default(); + apply(&mut unknown, b"draft", now); + let before = unknown.draft.clone(); + apply(&mut unknown, b"\x01", now); + assert_eq!(unknown.draft, before); + let mut unknown_empty = SessionState::default(); + apply(&mut unknown_empty, b"\x01", now); + assert!(unknown_empty.draft.is_empty()); + + let mut saturated = SessionState::default(); + apply(&mut saturated, &vec![b'x'; DRAFT_MAX_BYTES], now); + assert!(saturated.draft.saturated); + assert_eq!(saturated.draft.bytes.len(), DRAFT_MAX_BYTES); + apply(&mut saturated, b"\x7f", now); + apply(&mut saturated, b"\x17", now); + assert!( + !saturated.draft.is_empty(), + "editing cannot fail open after the bounded model saturates" ); + apply(&mut saturated, b"\x15", now); + assert!(saturated.draft.is_empty()); + assert!(!saturated.draft.saturated); - let now = Instant::now(); - let mut state = SessionState::default(); - update_local_input_state(&mut state, classify_local_input(b"draft"), now); - assert!(state.local_input_pending); - assert_eq!(state.last_local_input_at, Some(now)); + let mut interrupted = SessionState::default(); + apply(&mut interrupted, b"draft", now); + apply(&mut interrupted, b"\x03", now); + assert!(interrupted.draft.is_empty()); +} - update_local_input_state(&mut state, classify_local_input(b"\r"), now); - assert!(!state.local_input_pending); - assert!(state.last_local_input_at.is_none()); +#[test] +fn reserve_delivery_distinguishes_draft_recency_and_abandonment() { + let fake = fake_runtime(); + let mgr = mgr_with_fake(None, Arc::clone(&fake)); + let cap = capture(); - update_local_input_state(&mut state, classify_local_input(b"\x1b[D"), now); - assert!(!state.local_input_pending); - assert_eq!(state.last_local_input_at, Some(now)); + install_fake_session(&mgr, "draft-blocked"); + mgr.inject_direct_stdin("draft-blocked", b"unsent", cap.as_ref()) + .unwrap(); + match mgr.reserve_delivery("draft-blocked").unwrap() { + router::DeliveryReservation::PendingInput(remaining) => { + assert!(remaining > Duration::from_secs(9 * 60)); + assert!(remaining <= DRAFT_ABANDON_WINDOW); + } + other => panic!("a live draft must block delivery, got {other:?}"), + } + assert!(!mgr.input_quiescent("draft-blocked")); + + install_fake_session(&mgr, "backspace-cleared"); + mgr.inject_direct_stdin("backspace-cleared", b"x", cap.as_ref()) + .unwrap(); + mgr.inject_direct_stdin("backspace-cleared", b"\x7f", cap.as_ref()) + .unwrap(); + assert!(!mgr.input_quiescent("backspace-cleared")); + assert!(matches!( + mgr.reserve_delivery("backspace-cleared").unwrap(), + router::DeliveryReservation::RecentlyTyping(_) + )); + + install_fake_session(&mgr, "recent-activity"); + mgr.inject_direct_stdin("recent-activity", b"\x1b[D", cap.as_ref()) + .unwrap(); + assert!(matches!( + mgr.reserve_delivery("recent-activity").unwrap(), + router::DeliveryReservation::RecentlyTyping(_) + )); + + install_fake_session(&mgr, "abandoned-draft"); + mgr.inject_direct_stdin("abandoned-draft", b"forgotten", cap.as_ref()) + .unwrap(); + mgr.set_draft_abandon_ms(0); + let abandoned_token = match mgr.reserve_delivery("abandoned-draft").unwrap() { + router::DeliveryReservation::Ready(token) => token, + other => panic!("an abandoned draft must stop gating, got {other:?}"), + }; + assert!(mgr + .session_state("abandoned-draft") + .unwrap() + .lock() + .unwrap() + .draft + .is_empty()); + mgr.finish_delivery("abandoned-draft", abandoned_token); +} + +#[test] +fn failed_direct_stdin_write_restores_the_whole_draft() { + let fake = fake_runtime(); + let mgr = mgr_with_fake(None, Arc::clone(&fake)); + let cap = capture(); + let session_id = "draft-rollback"; + install_fake_session(&mgr, session_id); + + mgr.inject_direct_stdin(session_id, b"original", cap.as_ref()) + .unwrap(); + let (draft_before, input_at_before) = { + let state = mgr.session_state(session_id).unwrap(); + let state = state.lock().unwrap(); + (state.draft.clone(), state.last_local_input_at) + }; - state.local_input_pending = true; - update_local_input_state(&mut state, classify_local_input(b"\x03"), now); - assert!(!state.local_input_pending); - assert!(state.last_local_input_at.is_none()); + fake.fail_input_for(session_id); + assert!(mgr + .inject_direct_stdin(session_id, b"\x7f", cap.as_ref()) + .is_err()); + let state = mgr.session_state(session_id).unwrap(); + let state = state.lock().unwrap(); + assert_eq!(state.draft, draft_before); + assert_eq!(state.last_local_input_at, input_at_before); } // `await_pty_output` was deleted in the Step 9 cutover. Tests diff --git a/src/components/InboxBlockedPill.test.tsx b/src/components/InboxBlockedPill.test.tsx index bcc5477..df202d3 100644 --- a/src/components/InboxBlockedPill.test.tsx +++ b/src/components/InboxBlockedPill.test.tsx @@ -44,7 +44,6 @@ describe("InboxBlockedPill", () => { {}} {...props} @@ -53,15 +52,15 @@ describe("InboxBlockedPill", () => { }); } - it("renders the unread count, body, and Enter action", async () => { + it("renders the draft copy and clears it with Ctrl-U", async () => { await render(); expect(container.textContent).toContain("Inbox waiting (2)"); expect(container.textContent).toContain( - "— typing detected, delivery paused", + "— delivery paused: you have unsent input here", ); - expect(container.textContent).toContain("Clear input"); - expect(container.textContent).toContain("↵"); + expect(container.textContent).toContain("Clear draft"); + expect(container.textContent).toContain("⌃U"); await act(async () => { container @@ -69,7 +68,8 @@ describe("InboxBlockedPill", () => { ?.dispatchEvent(new MouseEvent("click", { bubbles: true })); }); expect(mocks.injectStdin).toHaveBeenCalledTimes(1); - expect(mocks.injectStdin).toHaveBeenCalledWith("S-IMPL", "\r"); + expect(mocks.injectStdin).toHaveBeenCalledWith("S-IMPL", "\x15"); + expect(mocks.injectStdin).not.toHaveBeenCalledWith("S-IMPL", "\r"); expect(mocks.injectStdin).not.toHaveBeenCalledWith("S-IMPL", "\x03"); }); @@ -80,20 +80,14 @@ describe("InboxBlockedPill", () => { expect(container.textContent).not.toContain("Inbox waiting (1)"); }); - it("keeps the pill visible but hides the action while busy", async () => { - await render({ idle: false }); - - expect(container.textContent).toContain("Inbox waiting (2)"); - expect(container.querySelector("button")).toBeNull(); - }); - it("hides the body in a narrow pane", async () => { await render({ narrow: true }); expect(container.textContent).toContain("Inbox waiting (2)"); expect(container.textContent).not.toContain( - "— typing detected, delivery paused", + "— delivery paused: you have unsent input here", ); - expect(container.textContent).toContain("Clear input"); + expect(container.textContent).toContain("Clear draft"); + expect(container.querySelector("button")).not.toBeNull(); }); }); diff --git a/src/components/InboxBlockedPill.tsx b/src/components/InboxBlockedPill.tsx index 8abd495..f8e0e0e 100644 --- a/src/components/InboxBlockedPill.tsx +++ b/src/components/InboxBlockedPill.tsx @@ -5,18 +5,16 @@ import { api } from "../lib/api"; export function InboxBlockedPill({ sessionId, unreadCount, - idle, narrow, onError, }: { sessionId: string; unreadCount: number; - idle: boolean; narrow: boolean; onError: (message: string) => void; }) { - const clearInput = () => { - void api.session.injectStdin(sessionId, "\r").catch((error) => { + const clearDraft = () => { + void api.session.injectStdin(sessionId, "\x15").catch((error) => { onError(String(error)); }); }; @@ -33,23 +31,21 @@ export function InboxBlockedPill({ {!narrow ? ( - — typing detected, delivery paused + — delivery paused: you have unsent input here ) : null} - {idle ? ( - - ) : null} + ); } diff --git a/src/pages/MissionWorkspace.tsx b/src/pages/MissionWorkspace.tsx index 5754a91..eb1da7e 100644 --- a/src/pages/MissionWorkspace.tsx +++ b/src/pages/MissionWorkspace.tsx @@ -1220,7 +1220,6 @@ export default function MissionWorkspace({ deliveryBlockedUnreadCount={ deliveryBlockedBySession[s.id]?.unread_count ?? null } - runnerIdle={runnerStatusMap[s.handle] === "idle"} // `visible` gate: a hidden keep-alive workspace // deactivates even its front tab's terminal so it // releases WebGL and skips geometry pushes. @@ -1377,7 +1376,6 @@ export default function MissionWorkspace({ function SlotPtyPane({ session, deliveryBlockedUnreadCount, - runnerIdle, active, forcedResuming, anySessionLive, @@ -1390,7 +1388,6 @@ function SlotPtyPane({ }: { session: SessionRow; deliveryBlockedUnreadCount: number | null; - runnerIdle: boolean; active: boolean; /** True when the parent's "Resume mission" button is iterating * through every slot. Drives the resuming overlay in this pane. */ @@ -1571,7 +1568,6 @@ function SlotPtyPane({