diff --git a/src/cosh-ng/crates/cosh-shell/src/runtime/controller.rs b/src/cosh-ng/crates/cosh-shell/src/runtime/controller.rs index 376e0d403b..d0a63a2421 100644 --- a/src/cosh-ng/crates/cosh-shell/src/runtime/controller.rs +++ b/src/cosh-ng/crates/cosh-shell/src/runtime/controller.rs @@ -69,7 +69,7 @@ fn render_raw_inline_events( ghost_route: std::mem::take(&mut inline_state.pending_input_ghost_route), }); } - let shell_busy = shell_has_active_foreground_command(snapshot.events()); + let shell_busy = inline_state.control.shell_busy(); if let Some(action) = shell_handoff_timeout_recovery_action(inline_state, shell_busy, &mut terminal_output)? { diff --git a/src/cosh-ng/crates/cosh-shell/src/runtime/dispatcher.rs b/src/cosh-ng/crates/cosh-shell/src/runtime/dispatcher.rs index 001122d019..7e5947cb04 100644 --- a/src/cosh-ng/crates/cosh-shell/src/runtime/dispatcher.rs +++ b/src/cosh-ng/crates/cosh-shell/src/runtime/dispatcher.rs @@ -42,7 +42,7 @@ use crate::runtime::state::InlineState; use crate::slash::runtime::render_slash_actions; use crate::slash::session::poll_background_compaction; -use super::controller::{pending_card_capture, shell_has_active_foreground_command}; +use super::controller::pending_card_capture; use super::events::{ShellEventBatch, ShellEventCursor, ShellEventSnapshot}; use super::startup::{ render_pending_recommendation_notice, render_startup_banner, render_startup_health_banner, @@ -108,11 +108,18 @@ fn render_inline_guidance_from_batch( ) -> std::io::Result<()> { state.personalization.poll_ready(); let events = snapshot.events(); - let action_events = batch.events.as_slice(); + let action_events = batch.events; let event_index_base = batch.global_index(0); - state.shell_exited = events + if action_events.is_empty() { + return poll_inline_runtime_without_shell_events(adapter, state, output); + } + + state.control.observe_shell_command_activity(action_events); + state.shell_exited |= action_events .iter() .any(|event| event.kind == ShellEventKind::ShellExited); + #[cfg(test)] + state.control.record_ledger_rebuild(); let ledger = build_command_blocks(events); record_completed_command_blocks(state, &ledger.blocks); state.session_blocks = ledger.blocks.clone(); @@ -122,9 +129,9 @@ fn render_inline_guidance_from_batch( // (and cd'd inside) a command. `events` is the session's cumulative // stream (the raw relay always passes the parser's full event vec, // which is append-only), so once a command marker is present it is - // present in every later dispatch and this assignment never - // regresses to `false`. - state.shell_command_activity_observed = events.iter().any(|event| { + // present in an incremental batch at most once, so this flag is + // monotonic without rescanning the cumulative stream. + state.shell_command_activity_observed |= action_events.iter().any(|event| { matches!( event.kind, ShellEventKind::CommandStarted @@ -144,7 +151,7 @@ fn render_inline_guidance_from_batch( // follows it. Scanned newest first: the most recent decisive // event wins, and an event-free dispatch never erases the last // known state. - for event in events.iter().rev() { + for event in action_events.iter().rev() { if event.kind == ShellEventKind::UserInputIntercepted && event.component.as_deref() == Some("shell_pty_input") && event.message.as_deref() == Some("write") @@ -198,7 +205,7 @@ fn render_inline_guidance_from_batch( event_index_base, )?; RuntimeDispatcher::apply_actions(approval_actions, state); - let shell_busy = shell_has_active_foreground_command(events); + let shell_busy = state.control.shell_busy(); if shell_busy { if let Some(cancellation) = state.personalization.analyzer_cancellation.as_ref() { cancellation.set_foreground_idle(false); @@ -387,6 +394,47 @@ fn render_inline_guidance_from_batch( Ok(()) } +fn poll_inline_runtime_without_shell_events( + adapter: &AdapterInstance, + state: &mut InlineState, + output: &mut W, +) -> std::io::Result<()> { + if state.shell_exited { + stop_active_agent_run_without_rendering(state, output)?; + return Ok(()); + } + + if state.control.shell_busy() { + if let Some(cancellation) = state.personalization.analyzer_cancellation.as_ref() { + cancellation.set_foreground_idle(false); + } + state.personalization.idle_since = None; + poll_active_agent_run_deferred(state, output, adapter)?; + poll_background_compaction(state, output, adapter, true)?; + return Ok(()); + } + + render_startup_health_banner(state, output)?; + render_pending_recommendation_notice(state, output)?; + let personal_idle = + state.agent_run.active.is_none() && !state.personalization.shell_input_active; + crate::recommendation::personal_session::poll_personal_session(state, adapter, personal_idle); + render_queued_hook_consultation(state, output)?; + if pending_card_capture(state).is_none() { + render_pending_command_insight(state, output)?; + } else { + state.pending_command_insight = None; + } + flush_held_agent_events(state, output)?; + if !state.control.shell_handoff().has_active_handoff() { + poll_active_agent_run(state, output, adapter)?; + } + flush_held_agent_events(state, output)?; + start_pending_shell_handoff_continuations(adapter, state, output)?; + poll_background_compaction(state, output, adapter, false)?; + render_owned_shell_prompt(state, output) +} + /// Starts the shell-evidence continuation whose delivery to the owning provider /// run failed. Reuses the existing `PendingRecovery` claim, so a given approval /// recovers at most once. diff --git a/src/cosh-ng/crates/cosh-shell/src/runtime/dispatcher_tests.rs b/src/cosh-ng/crates/cosh-shell/src/runtime/dispatcher_tests.rs index e946cbcd5d..76858878b1 100644 --- a/src/cosh-ng/crates/cosh-shell/src/runtime/dispatcher_tests.rs +++ b/src/cosh-ng/crates/cosh-shell/src/runtime/dispatcher_tests.rs @@ -8,10 +8,11 @@ fn dispatcher_advances_cursor_to_snapshot_end() { let adapter = AdapterInstance::Fake(FakeAgentAdapter); let mut state = InlineState::default(); let mut output = Vec::new(); - let snapshot = ShellEventSnapshot::new(&[ + let events = [ ShellEvent::user_input_intercepted("s", "/help"), ShellEvent::user_input_intercepted("s", "/help"), - ]); + ]; + let snapshot = ShellEventSnapshot::new(&events); let actions = RuntimeDispatcher::dispatch_inline_batch( &snapshot, @@ -136,6 +137,65 @@ fn dispatcher_records_the_latest_shell_prompt_cwd_report() { ); } +#[test] +fn idle_dispatch_reuses_ledger_until_new_events_arrive() { + let adapter = AdapterInstance::Fake(FakeAgentAdapter); + let mut state = InlineState::default(); + let mut output = Vec::new(); + let mut events = vec![ + ShellEvent::command_started("s", "cmd-1", "echo one", "/tmp", 1), + ShellEvent::command_finished( + ShellEventKind::CommandCompleted, + "s", + "cmd-1", + 0, + 2, + "/tmp/cmd-1", + ), + ]; + + dispatch_and_apply(&events, &adapter, &mut state, &mut output); + assert_eq!(state.session_blocks.len(), 1); + assert_eq!(state.control.ledger_rebuild_count(), 1); + + dispatch_and_apply(&events, &adapter, &mut state, &mut output); + assert_eq!(state.session_blocks.len(), 1); + assert_eq!( + state.control.ledger_rebuild_count(), + 1, + "an end cursor must bypass the cumulative ledger" + ); + + events.extend([ + ShellEvent::command_started("s", "cmd-2", "echo two", "/tmp", 3), + ShellEvent::command_finished( + ShellEventKind::CommandCompleted, + "s", + "cmd-2", + 0, + 4, + "/tmp/cmd-2", + ), + ]); + dispatch_and_apply(&events, &adapter, &mut state, &mut output); + + assert_eq!(state.session_blocks.len(), 2); + assert_eq!(state.control.ledger_rebuild_count(), 2); +} + +fn dispatch_and_apply( + events: &[ShellEvent], + adapter: &AdapterInstance, + state: &mut InlineState, + output: &mut Vec, +) { + let snapshot = ShellEventSnapshot::new(events); + let actions = + RuntimeDispatcher::dispatch_inline_batch(&snapshot, adapter, "bash", state, output) + .expect("dispatch should render"); + RuntimeDispatcher::apply_actions(actions, state); +} + #[test] fn pty_input_invalidates_the_prompt_cwd_report() { // Any PTY input may submit a `cd` through a binding the @@ -237,9 +297,10 @@ fn busy_shell_updates_the_analyzer_foreground_gate() { ..InlineState::default() }; let mut output = Vec::new(); - let snapshot = ShellEventSnapshot::new(&[ShellEvent::command_started( + let events = [ShellEvent::command_started( "session", "command", "sleep 1", "/tmp", 1, - )]); + )]; + let snapshot = ShellEventSnapshot::new(&events); RuntimeDispatcher::dispatch_inline_batch(&snapshot, &adapter, "bash", &mut state, &mut output) .expect("dispatch should render"); diff --git a/src/cosh-ng/crates/cosh-shell/src/runtime/events.rs b/src/cosh-ng/crates/cosh-shell/src/runtime/events.rs index df7cb52c5d..26c5ddf8fc 100644 --- a/src/cosh-ng/crates/cosh-shell/src/runtime/events.rs +++ b/src/cosh-ng/crates/cosh-shell/src/runtime/events.rs @@ -9,45 +9,43 @@ impl ShellEventCursor { } } -#[derive(Debug, Clone)] -pub(crate) struct ShellEventBatch { +#[derive(Debug, Clone, Copy)] +pub(crate) struct ShellEventBatch<'a> { pub(crate) from: ShellEventCursor, pub(crate) to: ShellEventCursor, - pub(crate) events: Vec, + pub(crate) events: &'a [ShellEvent], } -impl ShellEventBatch { +impl ShellEventBatch<'_> { pub(crate) fn global_index(&self, local_index: usize) -> usize { self.from.position() + local_index } } -#[derive(Debug, Clone)] -pub(crate) struct ShellEventSnapshot { - events: Vec, +#[derive(Debug, Clone, Copy)] +pub(crate) struct ShellEventSnapshot<'a> { + events: &'a [ShellEvent], } -impl ShellEventSnapshot { - pub(crate) fn new(events: &[ShellEvent]) -> Self { - Self { - events: events.to_vec(), - } +impl<'a> ShellEventSnapshot<'a> { + pub(crate) fn new(events: &'a [ShellEvent]) -> Self { + Self { events } } pub(crate) fn events(&self) -> &[ShellEvent] { - &self.events + self.events } pub(crate) fn cursor(&self) -> ShellEventCursor { ShellEventCursor(self.events.len()) } - pub(crate) fn batch_since(&self, cursor: ShellEventCursor) -> ShellEventBatch { + pub(crate) fn batch_since(&self, cursor: ShellEventCursor) -> ShellEventBatch<'a> { let from = cursor.position().min(self.events.len()); ShellEventBatch { from: ShellEventCursor(from), to: self.cursor(), - events: self.events[from..].to_vec(), + events: &self.events[from..], } } } @@ -76,12 +74,22 @@ mod tests { } #[test] - fn batch_maps_local_to_global_event_index() { + fn snapshot_and_batch_borrow_the_event_history() { let events = vec![ShellEvent::user_input_intercepted("s", "one")]; + let snapshot = ShellEventSnapshot::new(&events); + let batch = snapshot.batch_since(ShellEventCursor::default()); + + assert!(std::ptr::eq(snapshot.events().as_ptr(), events.as_ptr())); + assert!(std::ptr::eq(batch.events.as_ptr(), events.as_ptr())); + } + + #[test] + fn batch_maps_local_to_global_event_index() { + let events = [ShellEvent::user_input_intercepted("s", "one")]; let batch = ShellEventBatch { from: ShellEventCursor(7), to: ShellEventCursor(8), - events, + events: &events, }; assert_eq!(batch.global_index(0), 7); diff --git a/src/cosh-ng/crates/cosh-shell/src/runtime/state.rs b/src/cosh-ng/crates/cosh-shell/src/runtime/state.rs index 956ecaac68..10069c3dd4 100644 --- a/src/cosh-ng/crates/cosh-shell/src/runtime/state.rs +++ b/src/cosh-ng/crates/cosh-shell/src/runtime/state.rs @@ -30,7 +30,7 @@ use crate::runtime::state_prelude::{ }; use crate::runtime::trust_state::ApprovalTrustState; use crate::slash::session::SessionControlState; -use crate::types::AgentContextBinding; +use crate::types::{AgentContextBinding, ShellEvent, ShellEventKind}; pub(crate) struct AnalysisThrottle { recent: HashMap, @@ -507,6 +507,9 @@ pub(crate) struct ControlState { selectable_after_event_index: Option, pub(crate) trust: ApprovalTrustState, event_cursor: ShellEventCursor, + active_shell_command_ids: HashSet, + #[cfg(test)] + ledger_rebuild_count: usize, } impl ControlState { @@ -821,6 +824,35 @@ impl ControlState { pub(crate) fn set_event_cursor(&mut self, cursor: ShellEventCursor) { self.event_cursor = cursor; } + pub(crate) fn observe_shell_command_activity(&mut self, events: &[ShellEvent]) { + for event in events { + let Some(command_id) = event.command_id.as_ref() else { + continue; + }; + match event.kind { + ShellEventKind::CommandStarted => { + self.active_shell_command_ids.insert(command_id.clone()); + } + ShellEventKind::CommandCompleted + | ShellEventKind::CommandFailed + | ShellEventKind::UserInputIntercepted => { + self.active_shell_command_ids.remove(command_id); + } + _ => {} + } + } + } + pub(crate) fn shell_busy(&self) -> bool { + !self.active_shell_command_ids.is_empty() + } + #[cfg(test)] + pub(crate) fn record_ledger_rebuild(&mut self) { + self.ledger_rebuild_count += 1; + } + #[cfg(test)] + pub(crate) fn ledger_rebuild_count(&self) -> usize { + self.ledger_rebuild_count + } } #[derive(Default)]