From 63f3b79beed148a0dd415cedef004ddc2c174e7c Mon Sep 17 00:00:00 2001 From: kuddev Date: Wed, 19 Aug 2026 21:08:25 +0800 Subject: [PATCH] feat(runtime): run commands with real exit status --- nebula_app/src/cli.rs | 12 + nebula_app/src/display/state.rs | 2 + nebula_app/src/event.rs | 23 + nebula_app/src/gpui_shell/terminal/view.rs | 61 ++- .../src/gpui_shell/workspace/residency.rs | 33 +- nebula_app/src/runtime_api.rs | 432 +++++++++++++++++- nebula_app/src/window_context/runtime.rs | 34 ++ 7 files changed, 593 insertions(+), 4 deletions(-) diff --git a/nebula_app/src/cli.rs b/nebula_app/src/cli.rs index a6f0a5e..d2428bf 100644 --- a/nebula_app/src/cli.rs +++ b/nebula_app/src/cli.rs @@ -477,6 +477,18 @@ pub enum ControlCommand { #[clap(long, default_value_t = 1)] repeat: u16, }, + /// Run one shell command and return its real OSC 133 exit status. + Run { + #[clap(long)] + window: Option, + #[clap(long)] + pane: u64, + #[clap(long)] + command: String, + /// Submit the command and return its run id without waiting for completion. + #[clap(long)] + no_wait: bool, + }, /// Wait until a pane reaches a semantic task state. Wait { #[clap(long)] diff --git a/nebula_app/src/display/state.rs b/nebula_app/src/display/state.rs index 1473fd7..822aaaf 100644 --- a/nebula_app/src/display/state.rs +++ b/nebula_app/src/display/state.rs @@ -219,6 +219,8 @@ pub struct NebulaPaneState { pub touched: bool, pub inline_images: Vec, pub command_started: Option, + pub active_run: Option, + pub last_run: Option, pub running_program: Option, /// hook 直报的 AI CLI 会话身份（claude `session_id` / codex `thread-id`）。 /// 与 `running_program` 同生命周期：133;D 命令收尾时一起清除，快照据此 diff --git a/nebula_app/src/event.rs b/nebula_app/src/event.rs index a6df286..2e164d4 100644 --- a/nebula_app/src/event.rs +++ b/nebula_app/src/event.rs @@ -611,6 +611,19 @@ impl Processor { "bytes_sent": bytes_sent })) }, + RuntimeCommand::Run { window_id, pane_id, command, .. } => { + let id = self.runtime_target_window(*window_id, Some(*pane_id))?; + let run_id = self + .windows + .get_mut(&id) + .expect("resolved runtime window exists") + .runtime_run(*pane_id, command.clone())?; + self.runtime_result(serde_json::json!({ + "window_id": u64::from(id), + "pane_id": pane_id, + "run_id": run_id + })) + }, } } @@ -3247,6 +3260,11 @@ impl input::Processor> { self.ctx.nebula_state.pending_ssh_host = crate::ssh::ssh_destination(&self.ctx.nebula_state.last_committed); self.ctx.nebula_state.awaiting_input = false; + if let Some(run) = &mut self.ctx.nebula_state.active_run + && run.phase == crate::runtime_api::RuntimeRunPhase::Submitted + { + run.phase = crate::runtime_api::RuntimeRunPhase::Started; + } }, TerminalEvent::CommandDone { exit_code } => { // Take (not just clear) the program: the toast below @@ -3263,6 +3281,11 @@ impl input::Processor> { self.ctx.nebula_state.agent_status_rule = None; let pending_ssh = self.ctx.nebula_state.pending_ssh_host.take(); self.ctx.nebula_state.awaiting_input = false; + if let Some(run) = self.ctx.nebula_state.active_run.take() { + self.ctx.nebula_state.last_run = Some( + crate::runtime_api::RuntimeRunOutcome::command_done(run, exit_code), + ); + } // 助手错误恢复（spec 001）：Nebula 集成上报的退出码 // 走触发判定；裸 133;D（第三方集成）码为 None，静默。 if let Some(code) = exit_code { diff --git a/nebula_app/src/gpui_shell/terminal/view.rs b/nebula_app/src/gpui_shell/terminal/view.rs index 3e43410..28c4eed 100644 --- a/nebula_app/src/gpui_shell/terminal/view.rs +++ b/nebula_app/src/gpui_shell/terminal/view.rs @@ -173,6 +173,8 @@ pub struct TerminalView { /// `NEBULA|` 标题报 `running_program`，装了 OSC133 集成的 bash/zsh/nu 只 /// 发这一对语义标记——两条路都要能点亮侧栏的「运行中」。 command_running: bool, + active_run: Option, + last_run: Option, /// Hook 事件驱动的 agent 回合状态（旧壳 `nebula_state.agent_status` 的 /// 边沿触发版）。`Unknown` = 本 pane 没有 agent 参与，spinner 完全由 /// `command_running`/`running_program` 决定。 @@ -617,6 +619,8 @@ impl TerminalView { branch: String::new(), running_program: None, command_running: false, + active_run: None, + last_run: None, agent_status: crate::ai_agents::AgentStatus::Unknown, agent_status_source: crate::ai_agents::AgentStatusSource::Unknown, agent_status_rule: None, @@ -773,17 +777,26 @@ impl TerminalView { self.agent_status_rule = None; } self.command_running = true; + if let Some(run) = &mut self.active_run + && run.phase == crate::runtime_api::RuntimeRunPhase::Submitted + { + run.phase = crate::runtime_api::RuntimeRunPhase::Started; + } cx.notify(); }, // 退出码先只用来结束「运行中」；失败命令的未读标记要等 // `SidebarActivity` 扩出未读态再接，现在记下来也没有呈现位置。 - TermEvent::CommandDone { exit_code: _ } => { + TermEvent::CommandDone { exit_code } => { // 旧壳同款收尾：CLI 退回提示符后，它不再是这个 pane 的前台 // 事实——hook 稍后若仍在跑会重新点亮（handle_ai_hook 覆写）。 if self.running_program.take().is_some() || self.ai_session.take().is_some() { cx.emit(TerminalViewEvent::TitleChanged); } self.command_running = false; + if let Some(run) = self.active_run.take() { + self.last_run = + Some(crate::runtime_api::RuntimeRunOutcome::command_done(run, exit_code)); + } self.agent_status = crate::ai_agents::AgentStatus::Unknown; self.agent_status_source = crate::ai_agents::AgentStatusSource::Unknown; self.agent_status_rule = None; @@ -1192,6 +1205,52 @@ impl TerminalView { Ok(bytes_sent) } + pub fn runtime_run( + &mut self, + command: String, + cx: &mut Context, + ) -> Result { + crate::runtime_api::validate_command_line(&command)?; + if self.ssh_destination.is_some() { + return Err(crate::runtime_api::ApiError::new( + "exit_code_unavailable", + "pane.run is unavailable for native SSH panes because the remote integration does not report exit codes", + )); + } + if let Some(reason) = &self.exited { + return Err(crate::runtime_api::ApiError::new( + "invalid_state", + format!("pane has exited: {reason}"), + )); + } + self.ensure_runtime_readable()?; + if self.command_running || self.active_run.is_some() { + return Err(crate::runtime_api::ApiError::new( + "run_in_progress", + "the pane is already running a command", + )); + } + let run = crate::runtime_api::begin_runtime_run(); + let run_id = run.run_id; + self.active_run = Some(run); + self.last_run = None; + self.command_running = true; + self.suggest.last_committed.clone_from(&command); + let mut bytes = command.into_bytes(); + bytes.push(b'\r'); + self.write_input(bytes, cx); + cx.emit(TerminalViewEvent::TitleChanged); + Ok(run_id) + } + + pub fn runtime_active_run(&self) -> Option { + self.active_run + } + + pub fn runtime_last_run(&self) -> Option { + self.last_run.clone() + } + pub fn runtime_prompt( &mut self, text: String, diff --git a/nebula_app/src/gpui_shell/workspace/residency.rs b/nebula_app/src/gpui_shell/workspace/residency.rs index 329ac44..9030e75 100644 --- a/nebula_app/src/gpui_shell/workspace/residency.rs +++ b/nebula_app/src/gpui_shell/workspace/residency.rs @@ -101,7 +101,8 @@ impl NebulaWorkspace { | RuntimeCommand::NewTab { .. } | RuntimeCommand::Split { .. } | RuntimeCommand::Prompt { .. } - | RuntimeCommand::SendKey { .. } => { + | RuntimeCommand::SendKey { .. } + | RuntimeCommand::Run { .. } => { self.reveal_session(cx); self.runtime_pending.push(dispatch); }, @@ -281,6 +282,34 @@ impl NebulaWorkspace { cx, ) }, + RuntimeCommand::Run { window_id, pane_id, command, .. } => { + self.runtime_window_requested(*window_id)?; + let Some(tab_ix) = self.tab_of_pane(*pane_id) else { + return Err(ApiError::new( + "target_not_found", + format!("pane {pane_id} does not exist"), + )); + }; + let view = match self.tabs.get(tab_ix) { + Some(WorkspaceTab::Terminal { panes, .. }) => panes + .iter() + .find(|pane| pane.id == *pane_id) + .expect("tab_of_pane resolved a terminal pane") + .view + .clone(), + _ => unreachable!("tab_of_pane only resolves terminal tabs"), + }; + let run_id = view.update(cx, |view, cx| view.runtime_run(command.clone(), cx))?; + self.runtime_result( + json!({ + "window_id": self.runtime_window_id, + "pane_id": pane_id, + "run_id": run_id + }), + window, + cx, + ) + }, } } @@ -328,6 +357,8 @@ impl NebulaWorkspace { agent: view.runtime_agent(), task_state: view.runtime_task_state(), state_change_seq: 0, + active_run: view.runtime_active_run(), + last_run: view.runtime_last_run(), } }) .collect(); diff --git a/nebula_app/src/runtime_api.rs b/nebula_app/src/runtime_api.rs index 8fa3e17..db4e4aa 100644 --- a/nebula_app/src/runtime_api.rs +++ b/nebula_app/src/runtime_api.rs @@ -48,6 +48,7 @@ const MAX_CLIENTS: usize = 64; const MAX_WAIT: Duration = Duration::from_secs(24 * 60 * 60); static NEXT_REQUEST_ID: AtomicU64 = AtomicU64::new(1); +static NEXT_RUN_ID: AtomicU64 = AtomicU64::new(1); #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] #[serde(deny_unknown_fields)] @@ -298,6 +299,10 @@ pub struct RuntimePane { /// time. Stamped by [`RuntimeHub::publish`]; projection callers leave it 0. #[serde(default)] pub state_change_seq: u64, + #[serde(skip_serializing_if = "Option::is_none")] + pub active_run: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub last_run: Option, } #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] @@ -350,6 +355,94 @@ pub struct RuntimePaneProcesses { pub processes: Vec, } +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum RuntimeRunPhase { + Submitted, + Started, +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +pub struct RuntimePaneRun { + pub run_id: u64, + pub phase: RuntimeRunPhase, +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum RuntimeRunState { + Finished, + Failed, + Unavailable, +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum ExitCodeCapability { + Supported, + Unavailable, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct RuntimeRunOutcome { + pub run_id: u64, + pub state: RuntimeRunState, + pub exit_code: Option, + pub exit_code_capability: ExitCodeCapability, + #[serde(skip_serializing_if = "Option::is_none")] + pub unavailable_reason: Option, +} + +impl RuntimeRunOutcome { + pub(crate) fn command_done(run: RuntimePaneRun, exit_code: Option) -> Self { + if run.phase != RuntimeRunPhase::Started { + return Self::unavailable(run.run_id, "command_start_not_observed"); + } + match exit_code { + Some(0) => Self { + run_id: run.run_id, + state: RuntimeRunState::Finished, + exit_code, + exit_code_capability: ExitCodeCapability::Supported, + unavailable_reason: None, + }, + Some(_) => Self { + run_id: run.run_id, + state: RuntimeRunState::Failed, + exit_code, + exit_code_capability: ExitCodeCapability::Supported, + unavailable_reason: None, + }, + None => Self::unavailable(run.run_id, "exit_code_not_reported"), + } + } + + pub(crate) fn unavailable(run_id: u64, reason: impl Into) -> Self { + Self { + run_id, + state: RuntimeRunState::Unavailable, + exit_code: None, + exit_code_capability: ExitCodeCapability::Unavailable, + unavailable_reason: Some(reason.into()), + } + } +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct RuntimeRunResult { + pub window_id: u64, + pub pane_id: u64, + #[serde(flatten)] + pub outcome: RuntimeRunOutcome, +} + +pub(crate) fn begin_runtime_run() -> RuntimePaneRun { + RuntimePaneRun { + run_id: NEXT_RUN_ID.fetch_add(1, Ordering::Relaxed), + phase: RuntimeRunPhase::Submitted, + } +} + /// Named control keys accepted by `pane.send_key`. Printable text is /// intentionally absent; callers must use `pane.prompt` for text input. #[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] @@ -522,6 +615,13 @@ pub enum RuntimeCommand { modifiers: RuntimeKeyModifiers, repeat: u16, }, + Run { + window_id: Option, + pane_id: u64, + command: String, + wait: bool, + timeout_ms: u64, + }, } #[derive(Debug)] @@ -621,6 +721,10 @@ struct HubState { current: Option, next_subscription: u64, subscribers: Vec<(u64, SyncSender)>, + next_run_waiter: u64, + run_waiters: + std::collections::HashMap<(u64, u64, u64), Vec<(u64, SyncSender)>>, + completed_runs: std::collections::VecDeque, } impl RuntimeHub { @@ -640,6 +744,7 @@ impl RuntimeHub { // whose state is unchanged keep their previous counter, so an // otherwise-identical projection still compares equal here. stamp_state_change_seq(state.current.as_ref(), &mut snapshot); + observe_run_lifecycle(&mut state, &snapshot); if let Some(current) = &state.current { snapshot.revision = current.revision; if current == &snapshot { @@ -673,6 +778,164 @@ impl RuntimeHub { fn current(&self) -> Option { self.lock().current.clone() } + + fn wait_run( + &self, + window_id: u64, + pane_id: u64, + run_id: u64, + timeout: Duration, + ) -> Result { + let key = (window_id, pane_id, run_id); + let (waiter_id, receiver) = { + let mut state = self.lock(); + if let Some(result) = state.completed_runs.iter().find(|result| { + result.window_id == window_id + && result.pane_id == pane_id + && result.outcome.run_id == run_id + }) { + return run_result_or_capability_error(result.clone()); + } + let (sender, receiver) = mpsc::sync_channel(1); + state.next_run_waiter = state.next_run_waiter.saturating_add(1); + let waiter_id = state.next_run_waiter; + state.run_waiters.entry(key).or_default().push((waiter_id, sender)); + (waiter_id, receiver) + }; + + match receiver.recv_timeout(timeout) { + Ok(result) => run_result_or_capability_error(result), + Err(RecvTimeoutError::Disconnected) => Err(ApiError::new( + "runtime_unavailable", + "runtime run completion channel disconnected", + )), + Err(RecvTimeoutError::Timeout) => { + let mut state = self.lock(); + if let Some(waiters) = state.run_waiters.get_mut(&key) { + waiters.retain(|(id, _)| *id != waiter_id); + if waiters.is_empty() { + state.run_waiters.remove(&key); + } + } + let phase = state.current.as_ref().and_then(|snapshot| { + snapshot + .pane(Some(window_id), pane_id) + .ok() + .and_then(|pane| pane.active_run) + .filter(|run| run.run_id == run_id) + .map(|run| run.phase) + }); + let (code, message) = match phase { + Some(RuntimeRunPhase::Submitted) => ( + "run_start_timeout", + "the shell did not report CommandStart before timeout", + ), + Some(RuntimeRunPhase::Started) => { + ("timeout", "the command did not report CommandDone before timeout") + }, + None => ("run_aborted", "the pane or run disappeared before completion"), + }; + Err(ApiError::new(code, message).details(json!({ + "window_id": window_id, + "pane_id": pane_id, + "run_id": run_id, + "phase": phase + }))) + }, + } + } +} + +const COMPLETED_RUN_CACHE: usize = 256; + +fn observe_run_lifecycle(state: &mut HubState, snapshot: &RuntimeSnapshot) { + let mut observed = std::collections::HashMap::new(); + for window in &snapshot.windows { + for tab in &window.tabs { + for pane in &tab.panes { + if let Some(active) = pane.active_run { + observed.insert((window.id, pane.id, active.run_id), active.phase); + } + let Some(outcome) = pane.last_run.clone() else { + continue; + }; + let already_cached = state.completed_runs.iter().any(|result| { + result.window_id == window.id + && result.pane_id == pane.id + && result.outcome.run_id == outcome.run_id + }); + if !already_cached { + complete_run( + state, + RuntimeRunResult { window_id: window.id, pane_id: pane.id, outcome }, + ); + } + } + } + } + + let previous_active: Vec<_> = state + .current + .iter() + .flat_map(|previous| previous.windows.iter()) + .flat_map(|window| { + window.tabs.iter().flat_map(move |tab| { + tab.panes.iter().filter_map(move |pane| { + pane.active_run.map(|run| (window.id, pane.id, run.run_id)) + }) + }) + }) + .collect(); + for key @ (window_id, pane_id, run_id) in previous_active { + if observed.contains_key(&key) + || state.completed_runs.iter().any(|result| { + result.window_id == window_id + && result.pane_id == pane_id + && result.outcome.run_id == run_id + }) + { + continue; + } + complete_run( + state, + RuntimeRunResult { + window_id, + pane_id, + outcome: RuntimeRunOutcome::unavailable(run_id, "pane_or_run_disappeared"), + }, + ); + } +} + +fn complete_run(state: &mut HubState, result: RuntimeRunResult) { + let key = (result.window_id, result.pane_id, result.outcome.run_id); + if let Some(waiters) = state.run_waiters.remove(&key) { + for (_, sender) in waiters { + let _ = sender.try_send(result.clone()); + } + } + state.completed_runs.push_back(result); + while state.completed_runs.len() > COMPLETED_RUN_CACHE { + state.completed_runs.pop_front(); + } +} + +fn run_result_or_capability_error(result: RuntimeRunResult) -> Result { + if result.outcome.state != RuntimeRunState::Unavailable { + return Ok(result); + } + let reason = result + .outcome + .unavailable_reason + .clone() + .unwrap_or_else(|| "exit_code_unavailable".to_owned()); + let code = match reason.as_str() { + "command_start_not_observed" => "shell_integration_unavailable", + "pane_or_run_disappeared" => "run_aborted", + _ => "exit_code_unavailable", + }; + Err(ApiError::new(code, "the command completed without a reliable exit-code result") + .details(serde_json::to_value(result).unwrap_or_else(|_| json!({ "reason": reason })))) } #[derive(Debug, Deserialize)] @@ -751,6 +1014,19 @@ struct SendKeyParams { repeat: u16, } +#[derive(Debug, Deserialize)] +#[serde(deny_unknown_fields)] +struct RunParams { + #[serde(default)] + window_id: Option, + pane_id: u64, + command: String, + #[serde(default = "default_true")] + wait: bool, + #[serde(default = "default_run_timeout_ms")] + timeout_ms: u64, +} + #[derive(Debug, Deserialize)] #[serde(deny_unknown_fields)] struct SubscribeParams { @@ -798,6 +1074,10 @@ fn default_key_repeat() -> u16 { 1 } +fn default_run_timeout_ms() -> u64 { + COMMAND_TIMEOUT.as_millis() as u64 +} + impl RuntimeCommand { fn from_request(request: &ApiRequest) -> Result { match request.method.as_str() { @@ -862,6 +1142,22 @@ impl RuntimeCommand { repeat: params.repeat, }) }, + "pane.run" => { + let params: RunParams = parse_params(&request.params)?; + validate_command_line(¶ms.command)?; + if params.timeout_ms == 0 || Duration::from_millis(params.timeout_ms) > MAX_WAIT { + return Err(ApiError::invalid_params( + "timeout_ms must be between 1 and 86400000", + )); + } + Ok(Self::Run { + window_id: params.window_id, + pane_id: params.pane_id, + command: params.command, + wait: params.wait, + timeout_ms: params.timeout_ms, + }) + }, method => Err(ApiError::new( "method_not_found", format!("runtime API method {method:?} does not exist"), @@ -991,6 +1287,23 @@ pub(crate) fn validate_prompt(text: &str) -> Result<(), ApiError> { Ok(()) } +pub(crate) fn validate_command_line(command: &str) -> Result<(), ApiError> { + if command.trim().is_empty() { + return Err(ApiError::invalid_params("command must not be empty")); + } + if command.len() > MAX_PROMPT_BYTES { + return Err(ApiError::invalid_params(format!( + "command exceeds the {MAX_PROMPT_BYTES}-byte limit" + ))); + } + if command.chars().any(char::is_control) { + return Err(ApiError::invalid_params( + "command contains control characters; pane.run accepts one plain-text shell line", + )); + } + Ok(()) +} + /// A wait is satisfied only when the pane both reads as the requested state and /// has moved past the caller's baseline. Without the counter check, waiting on /// an already-idle pane returns immediately and the caller concludes its work @@ -1274,7 +1587,7 @@ fn handle_connection( "events.subscribe" => subscribe_connection(&mut stream, request, hub), "agents.list" => agents_connection(&mut stream, request, hub), "pane.wait" => wait_connection(&mut stream, request, hub), - _ => dispatch_connection(&mut stream, request, sink), + _ => dispatch_connection(&mut stream, request, sink, hub), } } @@ -1319,6 +1632,7 @@ fn runtime_description() -> Value { "pane.read", "pane.procs", "pane.send_key", + "pane.run", "pane.wait" ], // Additive params cannot be detected from `capabilities`: an older @@ -1521,6 +1835,7 @@ fn dispatch_connection( stream: &mut TcpStream, request: ApiRequest, sink: &EventSink, + hub: &RuntimeHub, ) -> Result<(), IoError> { let command = match RuntimeCommand::from_request(&request) { Ok(command) => command, @@ -1536,6 +1851,12 @@ fn dispatch_connection( modifiers.control, ); } + let run_wait = match &command { + RuntimeCommand::Run { wait: true, timeout_ms, .. } => { + Some(Duration::from_millis(*timeout_ms)) + }, + _ => None, + }; let (dispatch, receiver) = RuntimeDispatch::new(command); if !sink.emit_control(dispatch) { return write_response( @@ -1547,7 +1868,36 @@ fn dispatch_connection( ); } let response = match receiver.recv_timeout(COMMAND_TIMEOUT) { - Ok(Ok(result)) => ApiResponse::success(request.id, result), + Ok(Ok(result)) => { + if let Some(timeout) = run_wait { + let action = result.get("action").unwrap_or(&result); + let target = ( + action.get("window_id").and_then(Value::as_u64), + action.get("pane_id").and_then(Value::as_u64), + action.get("run_id").and_then(Value::as_u64), + ); + match target { + (Some(window_id), Some(pane_id), Some(run_id)) => { + match hub.wait_run(window_id, pane_id, run_id, timeout) { + Ok(run) => ApiResponse::success( + request.id, + json!({ "run": run, "snapshot": hub.current() }), + ), + Err(error) => ApiResponse::failure(request.id, error), + } + }, + _ => ApiResponse::failure( + request.id, + ApiError::new( + "invalid_runtime_response", + "pane.run did not return its window, pane, and run identity", + ), + ), + } + } else { + ApiResponse::success(request.id, result) + } + }, Ok(Err(error)) => ApiResponse::failure(request.id, error), Err(_) => ApiResponse::failure( request.id, @@ -1717,6 +2067,20 @@ pub fn run_cli(options: ControlOptions) -> Result<(), Box> { )?; print_response(&response, options.pretty) }, + CliCommand::Run { window, pane, command, no_wait } => { + let response = request_once( + "pane.run", + json!({ + "window_id": window, + "pane_id": pane, + "command": command, + "wait": !no_wait, + "timeout_ms": options.timeout_ms + }), + timeout.saturating_add(Duration::from_secs(1)), + )?; + print_response(&response, options.pretty) + }, CliCommand::Wait { window, pane, state, after_seq } => { wait_cli(window, pane, state, after_seq, timeout, options.pretty) }, @@ -1839,6 +2203,8 @@ mod tests { agent: None, task_state: state, state_change_seq: 0, + active_run: None, + last_run: None, }], }], }], @@ -1905,6 +2271,68 @@ mod tests { ); } + #[test] + fn run_requires_one_plain_shell_line() { + let valid = ApiRequest::new( + "token".into(), + "pane.run", + json!({ "pane_id": 3, "command": "cargo test", "wait": true }), + ); + assert!(matches!( + RuntimeCommand::from_request(&valid), + Ok(RuntimeCommand::Run { wait: true, .. }) + )); + + let multiline = ApiRequest::new( + "token".into(), + "pane.run", + json!({ "pane_id": 3, "command": "echo one\necho two" }), + ); + assert_eq!(RuntimeCommand::from_request(&multiline).unwrap_err().code, "invalid_params"); + } + + #[test] + fn run_outcome_requires_a_real_start_and_exit_code() { + let submitted = RuntimePaneRun { run_id: 41, phase: RuntimeRunPhase::Submitted }; + let no_start = RuntimeRunOutcome::command_done(submitted, Some(0)); + assert_eq!(no_start.state, RuntimeRunState::Unavailable); + assert_eq!(no_start.unavailable_reason.as_deref(), Some("command_start_not_observed")); + + let started = RuntimePaneRun { run_id: 42, phase: RuntimeRunPhase::Started }; + assert_eq!( + RuntimeRunOutcome::command_done(started, Some(0)).state, + RuntimeRunState::Finished + ); + assert_eq!( + RuntimeRunOutcome::command_done(started, Some(7)).state, + RuntimeRunState::Failed + ); + let missing_code = RuntimeRunOutcome::command_done(started, None); + assert_eq!(missing_code.state, RuntimeRunState::Unavailable); + assert_eq!(missing_code.exit_code_capability, ExitCodeCapability::Unavailable); + } + + #[test] + fn completed_run_cache_closes_the_waiter_registration_race() { + let hub = RuntimeHub::new(); + let mut running = snapshot(RuntimeTaskState::Running); + running.windows[0].tabs[0].panes[0].active_run = + Some(RuntimePaneRun { run_id: 51, phase: RuntimeRunPhase::Started }); + hub.publish(running); + + let mut done = snapshot(RuntimeTaskState::Finished); + done.windows[0].tabs[0].panes[0].last_run = Some(RuntimeRunOutcome::command_done( + RuntimePaneRun { run_id: 51, phase: RuntimeRunPhase::Started }, + Some(0), + )); + hub.publish(done); + + // The result was published before this waiter existed. The bounded + // cache must still return the exact run rather than timing out. + let result = hub.wait_run(7, 3, 51, Duration::from_millis(10)).unwrap(); + assert_eq!(result.outcome.exit_code, Some(0)); + } + #[test] fn settled_wait_excludes_only_running() { assert!(!wait_state_matches(RuntimeTaskState::Running, RuntimeWaitState::Settled)); diff --git a/nebula_app/src/window_context/runtime.rs b/nebula_app/src/window_context/runtime.rs index bddc7bf..bf2ff93 100644 --- a/nebula_app/src/window_context/runtime.rs +++ b/nebula_app/src/window_context/runtime.rs @@ -45,6 +45,8 @@ impl WindowContext { // counter at publish time, where the previous snapshot // is available to compare against. state_change_seq: 0, + active_run: pane.nebula_state.active_run, + last_run: pane.nebula_state.last_run.clone(), }) .collect(); let label = tab.custom_name.clone().unwrap_or_else(|| { @@ -253,6 +255,38 @@ impl WindowContext { self.display.window.request_redraw(); Ok(bytes_sent) } + + pub(crate) fn runtime_run(&mut self, pane_id: u64, command: String) -> Result { + crate::runtime_api::validate_command_line(&command)?; + let Some(index) = self.pane_index(pane_id) else { + return Err(ApiError::new( + "target_not_found", + format!("pane {pane_id} does not belong to the target window"), + )); + }; + let pane = &mut self.panes[index]; + if pane.ssh_destination.is_some() { + return Err(ApiError::new( + "exit_code_unavailable", + "pane.run is unavailable for native SSH panes because the remote integration does not report exit codes", + )); + } + if pane.nebula_state.command_started.is_some() || pane.nebula_state.active_run.is_some() { + return Err(ApiError::new("run_in_progress", "the pane is already running a command")); + } + let run = crate::runtime_api::begin_runtime_run(); + let run_id = run.run_id; + pane.nebula_state.active_run = Some(run); + pane.nebula_state.last_run = None; + pane.nebula_state.last_committed.clone_from(&command); + pane.nebula_state.touched = true; + pane.nebula_state.awaiting_input = false; + pane.notifier.notify(command.into_bytes()); + pane.notifier.notify(vec![b'\r']); + self.dirty = true; + self.display.window.request_redraw(); + Ok(run_id) + } } fn runtime_agent(pane: &super::Pane) -> Option {