diff --git a/crates/openshell-cli/src/commands/common.rs b/crates/openshell-cli/src/commands/common.rs index 0f12c3df4e..412f1fcbd3 100644 --- a/crates/openshell-cli/src/commands/common.rs +++ b/crates/openshell-cli/src/commands/common.rs @@ -1246,4 +1246,29 @@ mod tests { assert!(keys.contains(&"MY_ACCESS_KEY")); assert!(!keys.contains(&"PRIMARY_KEY")); } + #[test] + fn parse_cli_setting_value_parses_bool_aliases() { + let yes_value = parse_cli_setting_value("ocsf_json_enabled", "yes").expect("parse yes"); + assert_eq!(yes_value.value, Some(setting_value::Value::BoolValue(true))); + + let zero_value = parse_cli_setting_value("ocsf_json_enabled", "0").expect("parse 0"); + assert_eq!( + zero_value.value, + Some(setting_value::Value::BoolValue(false)) + ); + } + + #[test] + fn parse_cli_setting_value_rejects_invalid_bool() { + let err = parse_cli_setting_value("ocsf_json_enabled", "maybe") + .expect_err("invalid bool should fail"); + assert!(err.to_string().contains("invalid bool value")); + } + + #[test] + fn parse_cli_setting_value_rejects_unknown_key() { + let err = + parse_cli_setting_value("unknown_key", "value").expect_err("unknown key should fail"); + assert!(err.to_string().contains("unknown setting key")); + } } diff --git a/crates/openshell-cli/src/commands/inference.rs b/crates/openshell-cli/src/commands/inference.rs new file mode 100644 index 0000000000..3b08bf608d --- /dev/null +++ b/crates/openshell-cli/src/commands/inference.rs @@ -0,0 +1,309 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +use crate::color::Colorize; +use crate::tls::{TlsOptions, grpc_inference_client}; +use indicatif::{ProgressBar, ProgressStyle}; +use miette::{IntoDiagnostic, Result}; +use openshell_core::proto::{ + DeleteInferenceRouteRequest, GetInferenceRouteRequest, SetInferenceRouteRequest, +}; +use std::io::IsTerminal; +use std::time::Duration; +use tonic::{Code, Status}; + +#[allow(clippy::too_many_arguments)] +pub async fn gateway_inference_set( + server: &str, + provider_name: &str, + model_id: &str, + route_name: &str, + no_verify: bool, + timeout_secs: u64, + workspace: &str, + tls: &TlsOptions, +) -> Result<()> { + let progress = if std::io::stdout().is_terminal() { + let spinner = ProgressBar::new_spinner(); + spinner.set_style( + ProgressStyle::with_template("{spinner:.cyan} {msg} ({elapsed})") + .unwrap_or_else(|_| ProgressStyle::default_spinner()), + ); + spinner.set_message("Configuring inference..."); + spinner.enable_steady_tick(Duration::from_millis(120)); + Some(spinner) + } else { + None + }; + + let mut client = grpc_inference_client(server, tls).await?; + let response = client + .set_inference_route(SetInferenceRouteRequest { + provider_name: provider_name.to_string(), + model_id: model_id.to_string(), + route_name: route_name.to_string(), + verify: false, + no_verify, + timeout_secs, + workspace: workspace.to_string(), + }) + .await; + + if let Some(progress) = &progress { + progress.finish_and_clear(); + } + + let response = response.map_err(format_inference_status)?; + + let configured = response.into_inner(); + let label = if configured.route_name == "sandbox-system" { + "System inference configured:" + } else { + "Inference configured:" + }; + println!("{}", label.cyan().bold()); + println!(); + println!(" {} {}", "Workspace:".dimmed(), configured.workspace); + println!(" {} {}", "Route:".dimmed(), configured.route_name); + println!(" {} {}", "Provider:".dimmed(), configured.provider_name); + println!(" {} {}", "Model:".dimmed(), configured.model_id); + println!(" {} {}", "Version:".dimmed(), configured.version); + print_timeout(configured.timeout_secs); + if configured.validation_performed { + println!(" {}", "Validated Endpoints:".dimmed()); + for endpoint in configured.validated_endpoints { + println!(" - {} ({})", endpoint.url, endpoint.protocol); + } + } + Ok(()) +} + +#[allow(clippy::too_many_arguments)] +pub async fn gateway_inference_update( + server: &str, + provider_name: Option<&str>, + model_id: Option<&str>, + route_name: &str, + no_verify: bool, + timeout_secs: Option, + workspace: &str, + tls: &TlsOptions, +) -> Result<()> { + if provider_name.is_none() && model_id.is_none() && timeout_secs.is_none() { + return Err(miette::miette!( + "at least one of --provider, --model, or --timeout must be specified" + )); + } + + let mut client = grpc_inference_client(server, tls).await?; + + // Fetch current config to use as base for the partial update. + let current = client + .get_inference_route(GetInferenceRouteRequest { + route_name: route_name.to_string(), + workspace: workspace.to_string(), + }) + .await + .into_diagnostic()? + .into_inner(); + + let provider = provider_name.unwrap_or(¤t.provider_name); + let model = model_id.unwrap_or(¤t.model_id); + let timeout = timeout_secs.unwrap_or(current.timeout_secs); + + let progress = if std::io::stdout().is_terminal() { + let spinner = ProgressBar::new_spinner(); + spinner.set_style( + ProgressStyle::with_template("{spinner:.cyan} {msg} ({elapsed})") + .unwrap_or_else(|_| ProgressStyle::default_spinner()), + ); + spinner.set_message("Configuring inference..."); + spinner.enable_steady_tick(Duration::from_millis(120)); + Some(spinner) + } else { + None + }; + + let response = client + .set_inference_route(SetInferenceRouteRequest { + provider_name: provider.to_string(), + model_id: model.to_string(), + route_name: route_name.to_string(), + verify: false, + no_verify, + timeout_secs: timeout, + workspace: workspace.to_string(), + }) + .await; + + if let Some(progress) = &progress { + progress.finish_and_clear(); + } + + let response = response.map_err(format_inference_status)?; + + let configured = response.into_inner(); + let label = if configured.route_name == "sandbox-system" { + "System inference updated:" + } else { + "Inference updated:" + }; + println!("{}", label.cyan().bold()); + println!(); + println!(" {} {}", "Workspace:".dimmed(), configured.workspace); + println!(" {} {}", "Route:".dimmed(), configured.route_name); + println!(" {} {}", "Provider:".dimmed(), configured.provider_name); + println!(" {} {}", "Model:".dimmed(), configured.model_id); + println!(" {} {}", "Version:".dimmed(), configured.version); + print_timeout(configured.timeout_secs); + if configured.validation_performed { + println!(" {}", "Validated Endpoints:".dimmed()); + for endpoint in configured.validated_endpoints { + println!(" - {} ({})", endpoint.url, endpoint.protocol); + } + } + Ok(()) +} + +pub async fn gateway_inference_get( + server: &str, + route_name: Option<&str>, + workspace: &str, + tls: &TlsOptions, +) -> Result<()> { + let mut client = grpc_inference_client(server, tls).await?; + + if let Some(name) = route_name { + // Show a single route (--system was specified). + let response = client + .get_inference_route(GetInferenceRouteRequest { + route_name: name.to_string(), + workspace: workspace.to_string(), + }) + .await + .into_diagnostic()?; + + let configured = response.into_inner(); + let label = if name == "sandbox-system" { + "System inference:" + } else { + "Inference:" + }; + println!("{}", label.cyan().bold()); + println!(); + println!(" {} {}", "Workspace:".dimmed(), configured.workspace); + println!(" {} {}", "Provider:".dimmed(), configured.provider_name); + println!(" {} {}", "Model:".dimmed(), configured.model_id); + println!(" {} {}", "Version:".dimmed(), configured.version); + print_timeout(configured.timeout_secs); + } else { + // Show both routes by default. + print_inference_route(&mut client, "Inference", "", workspace).await; + println!(); + print_inference_route(&mut client, "System inference", "sandbox-system", workspace).await; + } + Ok(()) +} + +pub async fn gateway_inference_delete( + server: &str, + route_name: &str, + workspace: &str, + tls: &TlsOptions, +) -> Result<()> { + let mut client = grpc_inference_client(server, tls).await?; + + let response = client + .delete_inference_route(DeleteInferenceRouteRequest { + route_name: route_name.to_string(), + workspace: workspace.to_string(), + }) + .await + .into_diagnostic()?; + + let label = if route_name == "sandbox-system" { + "System inference route" + } else { + "Inference route" + }; + + if response.into_inner().deleted { + println!("{label} deleted."); + } else { + println!("{label} not found (already deleted)."); + } + Ok(()) +} + +async fn print_inference_route( + client: &mut crate::tls::GrpcInferenceClient, + label: &str, + route_name: &str, + workspace: &str, +) { + match client + .get_inference_route(GetInferenceRouteRequest { + route_name: route_name.to_string(), + workspace: workspace.to_string(), + }) + .await + { + Ok(response) => { + let configured = response.into_inner(); + println!("{}", format!("{label}:").cyan().bold()); + println!(); + println!(" {} {}", "Workspace:".dimmed(), configured.workspace); + println!(" {} {}", "Provider:".dimmed(), configured.provider_name); + println!(" {} {}", "Model:".dimmed(), configured.model_id); + println!(" {} {}", "Version:".dimmed(), configured.version); + print_timeout(configured.timeout_secs); + } + Err(e) if e.code() == Code::NotFound => { + println!("{}", format!("{label}:").cyan().bold()); + println!(); + println!(" {}", "Not configured".dimmed()); + } + Err(e) => { + println!("{}", format!("{label}:").cyan().bold()); + println!(); + println!(" {} {}", "Error:".red(), e.message()); + } + } +} + +fn print_timeout(timeout_secs: u64) { + if timeout_secs == 0 { + println!(" {} {}s (default)", "Timeout:".dimmed(), 60); + } else { + println!(" {} {}s", "Timeout:".dimmed(), timeout_secs); + } +} + +fn format_inference_status(status: Status) -> miette::Report { + let message = status.message().trim(); + + if message.is_empty() { + return miette::miette!("inference configuration failed ({})", status.code()); + } + + miette::miette!("{message}") +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn inference_status_preserves_actionable_server_messages() { + let err = format_inference_status(Status::invalid_argument("provider is missing")); + assert_eq!(err.to_string(), "provider is missing"); + } + + #[test] + fn inference_status_falls_back_to_the_status_code_for_empty_messages() { + let err = format_inference_status(Status::new(Code::Unavailable, "")); + let message = err.to_string(); + assert!(message.contains("inference configuration failed")); + assert!(message.to_lowercase().contains("unavailable")); + } +} diff --git a/crates/openshell-cli/src/commands/mod.rs b/crates/openshell-cli/src/commands/mod.rs index 8d75cf5ae3..e153b9dc33 100644 --- a/crates/openshell-cli/src/commands/mod.rs +++ b/crates/openshell-cli/src/commands/mod.rs @@ -3,4 +3,7 @@ pub mod common; pub mod gateway; +pub mod inference; pub mod provider; +pub mod service; +pub mod settings; diff --git a/crates/openshell-cli/src/commands/service.rs b/crates/openshell-cli/src/commands/service.rs new file mode 100644 index 0000000000..d03a0a022b --- /dev/null +++ b/crates/openshell-cli/src/commands/service.rs @@ -0,0 +1,746 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +use crate::color::Colorize; +use crate::commands::common::phase_name; +use crate::tls::{TlsOptions, grpc_client}; +use miette::{IntoDiagnostic, Result, WrapErr, miette}; +use openshell_core::ObjectId; +use openshell_core::net::set_tcp_nodelay_best_effort; +use openshell_core::proto::{ + CreateSshSessionRequest, DeleteServiceRequest, ExposeServiceRequest, GetSandboxRequest, + GetServiceRequest, ListServicesRequest, RevokeSshSessionRequest, Sandbox, SandboxPhase, + ServiceEndpointResponse, TcpForwardFrame, TcpForwardInit, TcpRelayTarget, tcp_forward_init, +}; +use std::time::Duration; +use tonic::{Code, Status}; + +pub async fn service_forward_tcp( + server: &str, + name: &str, + local: Option<&str>, + target_host: &str, + target_port: u16, + tls: &TlsOptions, + workspace: &str, +) -> Result<()> { + let (bind_addr, bind_port) = parse_tcp_forward_spec(local, target_port)?; + let mut client = grpc_client(server, tls).await?; + + let sandbox = fetch_ready_sandbox_for_forward(&mut client, name, workspace).await?; + + let listener = tokio::net::TcpListener::bind((bind_addr.as_str(), bind_port)) + .await + .into_diagnostic() + .wrap_err_with(|| format!("failed to bind local forward on {bind_addr}:{bind_port}"))?; + let local_addr = listener + .local_addr() + .into_diagnostic() + .wrap_err("failed to read local forward address")?; + eprintln!( + "{} Forwarding {} -> {}:{} in sandbox {} via gRPC", + "✓".green().bold(), + local_addr, + target_host, + target_port, + name, + ); + + let sandbox_id = sandbox.object_id().to_string(); + let (fatal_tx, mut fatal_rx) = tokio::sync::mpsc::channel::(1); + let mut health_check = tokio::time::interval(Duration::from_secs(2)); + health_check.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + loop { + tokio::select! { + Some(reason) = fatal_rx.recv() => { + return Err(miette::miette!("service forward stopped: {reason}")); + } + + _ = health_check.tick() => { + fetch_ready_sandbox_for_forward(&mut client, name, workspace).await?; + } + + accepted = listener.accept() => { + let (socket, peer) = accepted + .into_diagnostic() + .wrap_err("failed to accept local forward connection")?; + set_tcp_nodelay_best_effort(&socket); + let mut client = client.clone(); + let sandbox_id = sandbox_id.clone(); + let target_host = target_host.to_string(); + let service_id = format!("service-forward:{name}:{target_host}:{target_port}"); + let fatal_tx = fatal_tx.clone(); + tokio::spawn(async move { + let token = match create_forward_session_token(&mut client, &sandbox_id).await { + Ok(token) => token, + Err(err) => { + tracing::warn!(peer = %peer, error = %err, "service forward session creation failed"); + if err.fatal { + let _ = fatal_tx.send(err.message).await; + } + return; + } + }; + if let Err(err) = forward_one_tcp_connection( + &mut client, + socket, + sandbox_id, + target_host, + target_port, + service_id, + token.clone(), + ) + .await + { + tracing::warn!(peer = %peer, error = %err, "service forward connection failed"); + if err.fatal { + let _ = fatal_tx.send(err.message).await; + } + } + let _ = client + .revoke_ssh_session(RevokeSshSessionRequest { token }) + .await; + }); + } + } + } +} + +async fn create_forward_session_token( + client: &mut crate::tls::GrpcClient, + sandbox_id: &str, +) -> std::result::Result { + let response = client + .create_ssh_session(CreateSshSessionRequest { + sandbox_id: sandbox_id.to_string(), + }) + .await + .map_err(ForwardTcpConnectionError::from_status)?; + Ok(response.into_inner().token) +} + +async fn fetch_ready_sandbox_for_forward( + client: &mut crate::tls::GrpcClient, + name: &str, + workspace: &str, +) -> Result { + let response = match client + .get_sandbox(GetSandboxRequest { + name: name.to_string(), + workspace: workspace.to_string(), + }) + .await + { + Ok(response) => response, + Err(status) if status.code() == Code::NotFound => { + return Err(miette::miette!( + "sandbox '{name}' no longer exists; stopping service forward" + )); + } + Err(status) => return Err(status).into_diagnostic(), + }; + + let sandbox = response + .into_inner() + .sandbox + .ok_or_else(|| miette::miette!("sandbox '{name}' not found"))?; + + if SandboxPhase::try_from(sandbox.phase()) != Ok(SandboxPhase::Ready) { + return Err(miette::miette!( + "sandbox '{}' is no longer ready (phase: {}); stopping service forward", + name, + phase_name(sandbox.phase()) + )); + } + + Ok(sandbox) +} + +#[derive(Debug)] +struct ForwardTcpConnectionError { + message: String, + fatal: bool, +} + +impl ForwardTcpConnectionError { + fn transient(message: impl Into) -> Self { + Self { + message: message.into(), + fatal: false, + } + } + + fn from_status(status: Status) -> Self { + let fatal = matches!(status.code(), Code::NotFound | Code::FailedPrecondition); + Self { + message: status.to_string(), + fatal, + } + } +} + +impl std::fmt::Display for ForwardTcpConnectionError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(&self.message) + } +} + +impl std::error::Error for ForwardTcpConnectionError {} + +fn parse_tcp_forward_spec(local: Option<&str>, default_port: u16) -> Result<(String, u16)> { + let Some(spec) = local else { + return Ok(("127.0.0.1".to_string(), default_port)); + }; + + if let Some(pos) = spec.rfind(':') { + let addr = &spec[..pos]; + let port_str = &spec[pos + 1..]; + if let Ok(port) = port_str.parse::() { + if addr.is_empty() { + return Err(miette::miette!("bind address is required before ':'")); + } + return Ok((addr.to_string(), port)); + } + } + + let port: u16 = spec.parse().map_err(|_| { + miette::miette!("invalid local forward spec '{spec}': expected [bind_address:]port") + })?; + Ok(("127.0.0.1".to_string(), port)) +} + +async fn forward_one_tcp_connection( + client: &mut crate::tls::GrpcClient, + socket: tokio::net::TcpStream, + sandbox_id: String, + target_host: String, + target_port: u16, + service_id: String, + authorization_token: String, +) -> std::result::Result<(), ForwardTcpConnectionError> { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + use tokio_stream::wrappers::ReceiverStream; + + let (tx, rx) = tokio::sync::mpsc::channel::(16); + tx.send(TcpForwardFrame { + payload: Some(openshell_core::proto::tcp_forward_frame::Payload::Init( + TcpForwardInit { + sandbox_id, + service_id, + target: Some(tcp_forward_init::Target::Tcp(TcpRelayTarget { + host: target_host, + port: u32::from(target_port), + })), + authorization_token, + }, + )), + }) + .await + .map_err(|_| ForwardTcpConnectionError::transient("failed to initialize forward stream"))?; + + let mut response = match client.forward_tcp(ReceiverStream::new(rx)).await { + Ok(response) => response.into_inner(), + Err(status) => { + let err = ForwardTcpConnectionError::from_status(status); + drain_and_shutdown_local_socket(socket).await; + return Err(err); + } + }; + + let (mut local_read, mut local_write) = socket.into_split(); + + let to_gateway = tokio::spawn(async move { + let mut buf = vec![0u8; 64 * 1024]; + loop { + let n = local_read.read(&mut buf).await?; + if n == 0 { + break; + } + if tx + .send(TcpForwardFrame { + payload: Some(openshell_core::proto::tcp_forward_frame::Payload::Data( + buf[..n].to_vec(), + )), + }) + .await + .is_err() + { + break; + } + } + Ok::<(), std::io::Error>(()) + }); + + while let Some(frame) = response + .message() + .await + .map_err(ForwardTcpConnectionError::from_status)? + { + let Some(openshell_core::proto::tcp_forward_frame::Payload::Data(data)) = frame.payload + else { + continue; + }; + if data.is_empty() { + continue; + } + local_write + .write_all(&data) + .await + .map_err(|err| ForwardTcpConnectionError::transient(err.to_string()))?; + } + + let _ = local_write.shutdown().await; + to_gateway.abort(); + Ok(()) +} + +async fn drain_and_shutdown_local_socket(mut socket: tokio::net::TcpStream) { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + let mut buf = [0u8; 4096]; + while matches!( + tokio::time::timeout(Duration::from_millis(25), socket.read(&mut buf)).await, + Ok(Ok(n)) if n != 0 + ) {} + let _ = socket.shutdown().await; +} + +pub async fn service_expose( + server: &str, + sandbox: &str, + service: &str, + target_port: u16, + workspace: &str, + tls: &TlsOptions, +) -> Result<()> { + let mut client = grpc_client(server, tls).await?; + let response = client + .expose_service(ExposeServiceRequest { + sandbox: sandbox.to_string(), + service: service.to_string(), + target_port: u32::from(target_port), + domain: true, + workspace: workspace.to_string(), + }) + .await + .map_err(service_expose_status_error)? + .into_inner(); + + if service.is_empty() { + println!( + "{} Exposed sandbox {} -> 127.0.0.1:{}", + "✓".green().bold(), + sandbox.bold(), + target_port, + ); + } else { + println!( + "{} Exposed service {} on sandbox {} -> 127.0.0.1:{}", + "✓".green().bold(), + service.bold(), + sandbox.bold(), + target_port, + ); + } + if !response.url.is_empty() { + let url = service_url_for_gateway(&response.url, server); + println!(" URL: {}", url.cyan()); + } + Ok(()) +} + +fn service_expose_status_error(status: Status) -> miette::Report { + service_status_error("expose service", "sandbox:write", status) +} + +#[allow(clippy::too_many_arguments)] // user-facing CLI command +pub async fn service_list( + server: &str, + sandbox: Option<&str>, + limit: u32, + offset: u32, + workspace: &str, + all_workspaces: bool, + output: &str, + tls: &TlsOptions, +) -> Result<()> { + let mut client = grpc_client(server, tls).await?; + let response = client + .list_services(ListServicesRequest { + sandbox: sandbox.unwrap_or_default().to_string(), + limit, + offset, + workspace: if all_workspaces { + String::new() + } else { + workspace.to_string() + }, + all_workspaces, + }) + .await + .map_err(|status| service_status_error("list services", "sandbox:read", status))? + .into_inner(); + + let services = response + .services + .iter() + .filter_map(|response| service_endpoint_to_json(response, server)) + .collect::>(); + if crate::output::print_output_collection(output, &services, Clone::clone)? { + return Ok(()); + } + + if response.services.is_empty() { + if let Some(sandbox) = sandbox { + println!("No services exposed for sandbox {sandbox}."); + } else { + println!("No services exposed."); + } + return Ok(()); + } + + print_service_endpoint_table(&response.services, server, all_workspaces); + Ok(()) +} + +pub async fn service_get( + server: &str, + sandbox: &str, + service: &str, + workspace: &str, + tls: &TlsOptions, +) -> Result<()> { + let mut client = grpc_client(server, tls).await?; + let response = client + .get_service(GetServiceRequest { + sandbox: sandbox.to_string(), + service: service.to_string(), + workspace: workspace.to_string(), + }) + .await + .map_err(|status| service_status_error("get service", "sandbox:read", status))? + .into_inner(); + + print_service_endpoint_table(&[response], server, false); + Ok(()) +} + +pub async fn service_delete( + server: &str, + sandbox: &str, + service: &str, + workspace: &str, + tls: &TlsOptions, +) -> Result<()> { + let mut client = grpc_client(server, tls).await?; + let response = client + .delete_service(DeleteServiceRequest { + sandbox: sandbox.to_string(), + service: service.to_string(), + workspace: workspace.to_string(), + }) + .await + .map_err(|status| service_status_error("delete service", "sandbox:write", status))? + .into_inner(); + + if !response.deleted { + return Err(miette!("delete service failed: service endpoint not found")); + } + + if service.is_empty() { + println!( + "{} Deleted exposed sandbox {}", + "✓".green().bold(), + sandbox.bold(), + ); + } else { + println!( + "{} Deleted service {} on sandbox {}", + "✓".green().bold(), + service.bold(), + sandbox.bold(), + ); + } + Ok(()) +} + +fn service_status_error(action: &str, required_scope: &str, status: Status) -> miette::Report { + let message = status.message(); + match status.code() { + Code::PermissionDenied => { + miette!("{action} failed: permission denied (requires {required_scope})") + } + Code::Unauthenticated => miette!("{action} failed: authentication required"), + Code::NotFound if message == "sandbox not found" => { + miette!("{action} failed: sandbox not found") + } + Code::NotFound if message == "service endpoint not found" => { + miette!("{action} failed: service endpoint not found") + } + Code::InvalidArgument if !message.is_empty() => { + miette!("{action} failed: invalid request: {message}") + } + _ => miette!("{action} failed: {status}"), + } +} + +fn print_service_endpoint_table( + services: &[ServiceEndpointResponse], + gateway_endpoint: &str, + all_workspaces: bool, +) { + let rows = services + .iter() + .filter_map(|response| { + let endpoint = response.endpoint.as_ref()?; + let workspace = endpoint + .metadata + .as_ref() + .map_or("", |m| m.workspace.as_str()); + let service = service_display_name(&endpoint.service_name).to_string(); + let target = format!("127.0.0.1:{}", endpoint.target_port); + let url = if response.url.is_empty() { + String::new() + } else { + service_url_for_gateway(&response.url, gateway_endpoint) + }; + Some(( + workspace.to_string(), + endpoint.sandbox_name.clone(), + service, + target, + url, + )) + }) + .collect::>(); + + if rows.is_empty() { + return; + } + + let ws_width = if all_workspaces { + rows.iter() + .map(|(ws, _, _, _, _)| ws.len()) + .max() + .unwrap_or(9) + .max(9) + } else { + 0 + }; + let sandbox_width = rows + .iter() + .map(|(_, sandbox, _, _, _)| sandbox.len()) + .max() + .unwrap_or(7) + .max(7); + let service_width = rows + .iter() + .map(|(_, _, service, _, _)| service.len()) + .max() + .unwrap_or(7) + .max(7); + let target_width = rows + .iter() + .map(|(_, _, _, target, _)| target.len()) + .max() + .unwrap_or(6) + .max(6); + + if all_workspaces { + println!( + "{: Option { + let endpoint = response.endpoint.as_ref()?; + let workspace = endpoint + .metadata + .as_ref() + .map_or("", |metadata| metadata.workspace.as_str()); + let url = if response.url.is_empty() { + String::new() + } else { + service_url_for_gateway(&response.url, gateway_endpoint) + }; + + Some(serde_json::json!({ + "workspace": workspace, + "sandbox": endpoint.sandbox_name, + "service": endpoint.service_name, + "target_port": endpoint.target_port, + "url": url, + })) +} + +fn service_display_name(service: &str) -> &str { + if service.is_empty() { "-" } else { service } +} + +fn service_url_for_gateway(service_url: &str, gateway_endpoint: &str) -> String { + let (Ok(mut service_url), Ok(gateway_endpoint)) = ( + url::Url::parse(service_url), + url::Url::parse(gateway_endpoint), + ) else { + return service_url.to_string(); + }; + + if service_url + .set_port(gateway_endpoint.port_or_known_default()) + .is_err() + { + return service_url.to_string(); + } + + service_url.to_string() +} + +#[cfg(test)] +mod tests { + use super::*; + use openshell_core::proto::{ServiceEndpoint, datamodel::v1::ObjectMeta}; + + #[test] + fn service_endpoint_json_has_raw_fields_and_normalized_url() { + let response = ServiceEndpointResponse { + endpoint: Some(ServiceEndpoint { + metadata: Some(ObjectMeta { + workspace: "team-a".to_string(), + ..Default::default() + }), + sandbox_name: "api".to_string(), + service_name: String::new(), + target_port: 8080, + ..Default::default() + }), + url: "https://api.openshell.localhost:3000/".to_string(), + }; + + let value = service_endpoint_to_json(&response, "https://gateway.example:17670") + .expect("service endpoint JSON"); + assert_eq!( + value, + serde_json::json!({ + "workspace": "team-a", + "sandbox": "api", + "service": "", + "target_port": 8080, + "url": "https://api.openshell.localhost:17670/", + }) + ); + assert!(service_endpoint_to_json(&ServiceEndpointResponse::default(), "unused").is_none()); + } + + #[test] + fn service_url_for_gateway_uses_external_gateway_port() { + assert_eq!( + service_url_for_gateway( + "https://quiet-flamingo--notebook.navigator.openshell.localhost:8080/", + "https://127.0.0.1:31886" + ), + "https://quiet-flamingo--notebook.navigator.openshell.localhost:31886/" + ); + } + + #[test] + fn service_url_for_gateway_omits_default_external_port() { + assert_eq!( + service_url_for_gateway( + "https://quiet-flamingo--notebook.navigator.openshell.localhost:8080/", + "https://gateway.example.com" + ), + "https://quiet-flamingo--notebook.navigator.openshell.localhost/" + ); + } + + #[test] + fn service_url_for_gateway_preserves_service_scheme() { + assert_eq!( + service_url_for_gateway( + "http://quiet-flamingo--notebook.navigator.openshell.localhost:8080/", + "https://127.0.0.1:31886" + ), + "http://quiet-flamingo--notebook.navigator.openshell.localhost:31886/" + ); + } + + #[test] + fn service_url_for_gateway_uses_gateway_default_port() { + assert_eq!( + service_url_for_gateway( + "http://quiet-flamingo--notebook.navigator.openshell.localhost:8080/", + "https://gateway.example.com" + ), + "http://quiet-flamingo--notebook.navigator.openshell.localhost:443/" + ); + } + + #[test] + fn service_expose_status_error_mentions_required_scope() { + let report = service_expose_status_error(Status::permission_denied( + "scope 'sandbox:write' required", + )); + + assert_eq!( + report.to_string(), + "expose service failed: permission denied (requires sandbox:write)" + ); + } + + #[test] + fn tcp_forward_spec_parses_defaults_and_explicit_bind_addresses() { + assert_eq!( + parse_tcp_forward_spec(None, 8080).expect("default forward"), + ("127.0.0.1".to_string(), 8080) + ); + assert_eq!( + parse_tcp_forward_spec(Some("9090"), 8080).expect("port-only forward"), + ("127.0.0.1".to_string(), 9090) + ); + assert_eq!( + parse_tcp_forward_spec(Some("0.0.0.0:7070"), 8080).expect("explicit bind"), + ("0.0.0.0".to_string(), 7070) + ); + assert!(parse_tcp_forward_spec(Some(":7070"), 8080).is_err()); + assert!(parse_tcp_forward_spec(Some("not-a-port"), 8080).is_err()); + } + + #[test] + fn forward_connection_errors_classify_terminal_sandbox_states_as_fatal() { + for code in [Code::NotFound, Code::FailedPrecondition] { + assert!(ForwardTcpConnectionError::from_status(Status::new(code, "terminal")).fatal); + } + for code in [Code::Unavailable, Code::Internal, Code::PermissionDenied] { + assert!(!ForwardTcpConnectionError::from_status(Status::new(code, "retryable")).fatal); + } + } +} diff --git a/crates/openshell-cli/src/commands/settings.rs b/crates/openshell-cli/src/commands/settings.rs new file mode 100644 index 0000000000..c53d30d700 --- /dev/null +++ b/crates/openshell-cli/src/commands/settings.rs @@ -0,0 +1,389 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +use crate::color::Colorize; +use crate::commands::common::{ + confirm_global_setting_delete, confirm_global_setting_takeover, format_setting_value, + parse_cli_setting_value, +}; +use crate::tls::{TlsOptions, grpc_client}; +use miette::{IntoDiagnostic, Result}; +use openshell_core::ObjectId; +use openshell_core::proto::{ + GetGatewayConfigRequest, GetSandboxConfigRequest, GetSandboxConfigResponse, GetSandboxRequest, + PolicySource, SettingScope, UpdateConfigRequest, +}; + +pub async fn sandbox_settings_get( + server: &str, + name: &str, + json: bool, + workspace: &str, + tls: &TlsOptions, +) -> Result<()> { + let mut client = grpc_client(server, tls).await?; + let sandbox = client + .get_sandbox(GetSandboxRequest { + name: name.to_string(), + workspace: workspace.to_string(), + }) + .await + .into_diagnostic()? + .into_inner() + .sandbox + .ok_or_else(|| miette::miette!("sandbox not found"))?; + + let response = client + .get_sandbox_config(GetSandboxConfigRequest { + sandbox_id: sandbox.object_id().to_string(), + }) + .await + .into_diagnostic()? + .into_inner(); + + if json { + let obj = settings_to_json_sandbox(name, workspace, &response); + println!("{}", serde_json::to_string_pretty(&obj).into_diagnostic()?); + return Ok(()); + } + + let policy_source = if response.policy_source == PolicySource::Global as i32 { + "global" + } else { + "sandbox" + }; + + println!("Sandbox: {name}"); + println!("Config Rev: {}", response.config_revision); + println!("Policy Source: {policy_source}"); + println!("Policy Hash: {}", response.policy_hash); + + if response.settings.is_empty() { + println!("Settings: No settings available."); + return Ok(()); + } + + println!("Settings:"); + let mut keys: Vec<_> = response.settings.keys().cloned().collect(); + keys.sort(); + for key in keys { + if let Some(setting) = response.settings.get(&key) { + let scope = match SettingScope::try_from(setting.scope) { + Ok(SettingScope::Global) => "global", + Ok(SettingScope::Sandbox) => "sandbox", + _ => "unset", + }; + println!( + " {} = {} ({})", + key, + format_setting_value(setting.value.as_ref()), + scope + ); + } + } + + Ok(()) +} + +pub async fn gateway_settings_get(server: &str, json: bool, tls: &TlsOptions) -> Result<()> { + let mut client = grpc_client(server, tls).await?; + let response = client + .get_gateway_config(GetGatewayConfigRequest {}) + .await + .into_diagnostic()? + .into_inner(); + + if json { + let obj = settings_to_json_global(&response); + println!("{}", serde_json::to_string_pretty(&obj).into_diagnostic()?); + return Ok(()); + } + + println!("Scope: global"); + println!("Settings Rev: {}", response.settings_revision); + + if response.settings.is_empty() { + println!("Settings: No settings available."); + return Ok(()); + } + + println!("Settings:"); + let mut keys: Vec<_> = response.settings.keys().cloned().collect(); + keys.sort(); + for key in keys { + if let Some(setting) = response.settings.get(&key) { + println!(" {} = {}", key, format_setting_value(Some(setting))); + } + } + Ok(()) +} + +fn settings_to_json_sandbox( + name: &str, + workspace: &str, + response: &GetSandboxConfigResponse, +) -> serde_json::Value { + let policy_source = if response.policy_source == PolicySource::Global as i32 { + "global" + } else { + "sandbox" + }; + + let mut settings = serde_json::Map::new(); + let mut keys: Vec<_> = response.settings.keys().cloned().collect(); + keys.sort(); + for key in keys { + if let Some(setting) = response.settings.get(&key) { + let scope = match SettingScope::try_from(setting.scope) { + Ok(SettingScope::Global) => "global", + Ok(SettingScope::Sandbox) => "sandbox", + _ => "unset", + }; + settings.insert( + key, + serde_json::json!({ + "value": format_setting_value(setting.value.as_ref()), + "scope": scope, + }), + ); + } + } + + serde_json::json!({ + "sandbox": name, + "workspace": workspace, + "config_revision": response.config_revision, + "policy_source": policy_source, + "policy_hash": response.policy_hash, + "settings": settings, + }) +} + +fn settings_to_json_global( + response: &openshell_core::proto::GetGatewayConfigResponse, +) -> serde_json::Value { + let mut settings = serde_json::Map::new(); + let mut keys: Vec<_> = response.settings.keys().cloned().collect(); + keys.sort(); + for key in keys { + if let Some(setting) = response.settings.get(&key) { + settings.insert(key, serde_json::json!(format_setting_value(Some(setting)))); + } + } + + serde_json::json!({ + "scope": "global", + "settings_revision": response.settings_revision, + "settings": settings, + }) +} + +pub async fn gateway_setting_set( + server: &str, + key: &str, + value: &str, + yes: bool, + workspace: &str, + tls: &TlsOptions, +) -> Result<()> { + let setting_value = parse_cli_setting_value(key, value)?; + confirm_global_setting_takeover(key, yes)?; + + let mut client = grpc_client(server, tls).await?; + let response = client + .update_config(UpdateConfigRequest { + name: String::new(), + setting_key: key.to_string(), + setting_value: Some(setting_value), + global: true, + workspace: workspace.to_string(), + ..Default::default() + }) + .await + .into_diagnostic()? + .into_inner(); + + println!( + "{} Set global setting {}={} (revision {})", + "✓".green().bold(), + key, + value, + response.settings_revision + ); + Ok(()) +} + +pub async fn sandbox_setting_set( + server: &str, + name: &str, + key: &str, + value: &str, + workspace: &str, + tls: &TlsOptions, +) -> Result<()> { + let setting_value = parse_cli_setting_value(key, value)?; + + let mut client = grpc_client(server, tls).await?; + let response = client + .update_config(UpdateConfigRequest { + name: name.to_string(), + setting_key: key.to_string(), + setting_value: Some(setting_value), + workspace: workspace.to_string(), + ..Default::default() + }) + .await + .into_diagnostic()? + .into_inner(); + + println!( + "{} Set sandbox setting {}={} for {} (revision {})", + "✓".green().bold(), + key, + value, + name, + response.settings_revision + ); + Ok(()) +} + +pub async fn gateway_setting_delete( + server: &str, + key: &str, + yes: bool, + workspace: &str, + tls: &TlsOptions, +) -> Result<()> { + confirm_global_setting_delete(key, yes)?; + + let mut client = grpc_client(server, tls).await?; + let response = client + .update_config(UpdateConfigRequest { + name: String::new(), + setting_key: key.to_string(), + delete_setting: true, + global: true, + workspace: workspace.to_string(), + ..Default::default() + }) + .await + .into_diagnostic()? + .into_inner(); + + if response.deleted { + println!( + "{} Deleted global setting {} (revision {})", + "✓".green().bold(), + key, + response.settings_revision + ); + } else { + println!("{} Global setting {} not found", "!".yellow(), key); + } + Ok(()) +} + +pub async fn sandbox_setting_delete( + server: &str, + name: &str, + key: &str, + workspace: &str, + tls: &TlsOptions, +) -> Result<()> { + let mut client = grpc_client(server, tls).await?; + let response = client + .update_config(UpdateConfigRequest { + name: name.to_string(), + setting_key: key.to_string(), + delete_setting: true, + workspace: workspace.to_string(), + ..Default::default() + }) + .await + .into_diagnostic()? + .into_inner(); + + if response.deleted { + println!( + "{} Deleted sandbox setting {} for {} (revision {})", + "✓".green().bold(), + key, + name, + response.settings_revision + ); + } else { + println!( + "{} Sandbox setting {} not found for {}", + "!".yellow(), + key, + name, + ); + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + use openshell_core::proto::{ + EffectiveSetting, GetGatewayConfigResponse, SettingValue, setting_value, + }; + use std::collections::HashMap; + + fn bool_setting(value: bool) -> SettingValue { + SettingValue { + value: Some(setting_value::Value::BoolValue(value)), + } + } + + #[test] + fn sandbox_settings_json_preserves_effective_scope_and_metadata() { + let response = GetSandboxConfigResponse { + settings: HashMap::from([( + "ocsf_json_enabled".to_string(), + EffectiveSetting { + value: Some(bool_setting(true)), + scope: SettingScope::Global as i32, + }, + )]), + config_revision: 7, + policy_source: PolicySource::Global as i32, + policy_hash: "policy-hash".to_string(), + ..Default::default() + }; + + assert_eq!( + settings_to_json_sandbox("dev", "team-a", &response), + serde_json::json!({ + "sandbox": "dev", + "workspace": "team-a", + "config_revision": 7, + "policy_source": "global", + "policy_hash": "policy-hash", + "settings": { + "ocsf_json_enabled": { + "value": "true", + "scope": "global", + } + }, + }) + ); + } + + #[test] + fn gateway_settings_json_preserves_revision_and_values() { + let response = GetGatewayConfigResponse { + settings: HashMap::from([("ocsf_json_enabled".to_string(), bool_setting(false))]), + settings_revision: 9, + }; + + assert_eq!( + settings_to_json_global(&response), + serde_json::json!({ + "scope": "global", + "settings_revision": 9, + "settings": {"ocsf_json_enabled": "false"}, + }) + ); + } +} diff --git a/crates/openshell-cli/src/run.rs b/crates/openshell-cli/src/run.rs index 78a794aa30..3dc15110ef 100644 --- a/crates/openshell-cli/src/run.rs +++ b/crates/openshell-cli/src/run.rs @@ -8,11 +8,10 @@ pub use crate::commands::common::{ parse_secret_material_env_pairs, warn_credential_env_vars, }; use crate::commands::common::{ - ProvisioningDisplay, ProvisioningStep, confirm_global_setting_delete, - confirm_global_setting_takeover, format_epoch_ms, format_setting_value, format_timestamp, - format_timestamp_ms, handle_platform_progress_event, is_provisioning_progress_event, - non_empty_or, parse_cli_setting_value, parse_duration_to_ms, phase_name, - print_policy_merge_warnings, print_sandbox_header, print_sandbox_policy, + ProvisioningDisplay, ProvisioningStep, confirm_global_setting_takeover, format_epoch_ms, + format_timestamp, format_timestamp_ms, handle_platform_progress_event, + is_provisioning_progress_event, non_empty_or, parse_cli_setting_value, parse_duration_to_ms, + phase_name, print_policy_merge_warnings, print_sandbox_header, print_sandbox_policy, provisioning_timeout_message, ready_false_condition_message, scrub_git_env, short_hash, truncate_status_field, }; @@ -33,33 +32,39 @@ pub use crate::commands::provider::{ sandbox_provider_detach, sandbox_provider_list, }; +pub use crate::commands::inference::{ + gateway_inference_delete, gateway_inference_get, gateway_inference_set, + gateway_inference_update, +}; +pub use crate::commands::service::{ + service_delete, service_expose, service_forward_tcp, service_get, service_list, +}; +pub use crate::commands::settings::{ + gateway_setting_delete, gateway_setting_set, gateway_settings_get, sandbox_setting_delete, + sandbox_setting_set, sandbox_settings_get, +}; + use crate::color::Colorize; use crate::policy_update::build_policy_update_plan; -use crate::tls::{TlsOptions, grpc_client, grpc_inference_client}; +use crate::tls::{TlsOptions, grpc_client}; use futures::StreamExt; -use indicatif::{ProgressBar, ProgressStyle}; use miette::{IntoDiagnostic, Result, WrapErr, miette}; use openshell_bootstrap::{ GatewayMetadata, clear_last_sandbox_if_matches, get_gateway_metadata, save_last_sandbox, }; -use openshell_core::net::set_tcp_nodelay_best_effort; use openshell_core::proto::{ ApproveAllDraftChunksRequest, ApproveDraftChunkRequest, ClearDraftChunksRequest, - CreateSandboxRequest, CreateSandboxTemplateRequest, CreateSshSessionRequest, - DeleteInferenceRouteRequest, DeleteSandboxRequest, DeleteSandboxTemplateRequest, - DeleteServiceRequest, ExecSandboxRequest, ExposeServiceRequest, GetCurrentUserRequest, - GetDraftHistoryRequest, GetDraftPolicyRequest, GetGatewayConfigRequest, - GetInferenceRouteRequest, GetSandboxConfigRequest, GetSandboxConfigResponse, - GetSandboxLogsRequest, GetSandboxPolicyStatusRequest, GetSandboxRequest, - GetSandboxTemplateRequest, GetServiceRequest, GpuResourceRequirements, - ListSandboxPoliciesRequest, ListSandboxTemplatesRequest, ListSandboxesRequest, - ListServicesRequest, PolicySource, PolicyStatus, RejectDraftChunkRequest, ResourceRequirements, - RevokeSshSessionRequest, Sandbox, SandboxPhase, SandboxPolicy, SandboxResources, - SandboxServiceLevel, SandboxSpec, SandboxStartup, SandboxTemplate, SandboxWorkloadConfig, - SandboxWorkloadTemplate, SandboxWorkloadTemplateSpec, ServiceEndpointResponse, - SetInferenceRouteRequest, SettingScope, StartSandboxRequest, StopSandboxRequest, - TcpForwardFrame, TcpForwardInit, TcpRelayTarget, UpdateConfigRequest, WatchSandboxRequest, - exec_sandbox_event, tcp_forward_init, + CreateSandboxRequest, CreateSandboxTemplateRequest, DeleteSandboxRequest, + DeleteSandboxTemplateRequest, ExecSandboxRequest, GetCurrentUserRequest, + GetDraftHistoryRequest, GetDraftPolicyRequest, GetSandboxConfigRequest, + GetSandboxConfigResponse, GetSandboxLogsRequest, GetSandboxPolicyStatusRequest, + GetSandboxRequest, GetSandboxTemplateRequest, GpuResourceRequirements, + ListSandboxPoliciesRequest, ListSandboxTemplatesRequest, ListSandboxesRequest, PolicySource, + PolicyStatus, RejectDraftChunkRequest, ResourceRequirements, Sandbox, SandboxPhase, + SandboxPolicy, SandboxResources, SandboxServiceLevel, SandboxSpec, SandboxStartup, + SandboxTemplate, SandboxWorkloadConfig, SandboxWorkloadTemplate, SandboxWorkloadTemplateSpec, + StartSandboxRequest, StopSandboxRequest, UpdateConfigRequest, WatchSandboxRequest, + exec_sandbox_event, }; use openshell_core::settings; use openshell_core::{ObjectId, ObjectName, ObjectWorkspace}; @@ -69,7 +74,7 @@ use std::io::{ErrorKind, IsTerminal, Read, Write}; use std::path::{Path, PathBuf}; use std::process::Command; use std::time::{Duration, Instant}; -use tonic::{Code, Status}; +use tonic::Code; // Re-export SSH functions for backward compatibility pub use crate::ssh::{Editor, print_ssh_config}; @@ -1674,296 +1679,6 @@ pub async fn sandbox_exec_grpc( Ok(exit_code) } -pub async fn service_forward_tcp( - server: &str, - name: &str, - local: Option<&str>, - target_host: &str, - target_port: u16, - tls: &TlsOptions, - workspace: &str, -) -> Result<()> { - let (bind_addr, bind_port) = parse_tcp_forward_spec(local, target_port)?; - let mut client = grpc_client(server, tls).await?; - - let sandbox = fetch_ready_sandbox_for_forward(&mut client, name, workspace).await?; - - let listener = tokio::net::TcpListener::bind((bind_addr.as_str(), bind_port)) - .await - .into_diagnostic() - .wrap_err_with(|| format!("failed to bind local forward on {bind_addr}:{bind_port}"))?; - let local_addr = listener - .local_addr() - .into_diagnostic() - .wrap_err("failed to read local forward address")?; - eprintln!( - "{} Forwarding {} -> {}:{} in sandbox {} via gRPC", - "✓".green().bold(), - local_addr, - target_host, - target_port, - name, - ); - - let sandbox_id = sandbox.object_id().to_string(); - let (fatal_tx, mut fatal_rx) = tokio::sync::mpsc::channel::(1); - let mut health_check = tokio::time::interval(Duration::from_secs(2)); - health_check.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); - loop { - tokio::select! { - Some(reason) = fatal_rx.recv() => { - return Err(miette::miette!("service forward stopped: {reason}")); - } - - _ = health_check.tick() => { - fetch_ready_sandbox_for_forward(&mut client, name, workspace).await?; - } - - accepted = listener.accept() => { - let (socket, peer) = accepted - .into_diagnostic() - .wrap_err("failed to accept local forward connection")?; - set_tcp_nodelay_best_effort(&socket); - let mut client = client.clone(); - let sandbox_id = sandbox_id.clone(); - let target_host = target_host.to_string(); - let service_id = format!("service-forward:{name}:{target_host}:{target_port}"); - let fatal_tx = fatal_tx.clone(); - tokio::spawn(async move { - let token = match create_forward_session_token(&mut client, &sandbox_id).await { - Ok(token) => token, - Err(err) => { - tracing::warn!(peer = %peer, error = %err, "service forward session creation failed"); - if err.fatal { - let _ = fatal_tx.send(err.message).await; - } - return; - } - }; - if let Err(err) = forward_one_tcp_connection( - &mut client, - socket, - sandbox_id, - target_host, - target_port, - service_id, - token.clone(), - ) - .await - { - tracing::warn!(peer = %peer, error = %err, "service forward connection failed"); - if err.fatal { - let _ = fatal_tx.send(err.message).await; - } - } - let _ = client - .revoke_ssh_session(RevokeSshSessionRequest { token }) - .await; - }); - } - } - } -} - -async fn create_forward_session_token( - client: &mut crate::tls::GrpcClient, - sandbox_id: &str, -) -> std::result::Result { - let response = client - .create_ssh_session(CreateSshSessionRequest { - sandbox_id: sandbox_id.to_string(), - }) - .await - .map_err(ForwardTcpConnectionError::from_status)?; - Ok(response.into_inner().token) -} - -async fn fetch_ready_sandbox_for_forward( - client: &mut crate::tls::GrpcClient, - name: &str, - workspace: &str, -) -> Result { - let response = match client - .get_sandbox(GetSandboxRequest { - name: name.to_string(), - workspace: workspace.to_string(), - }) - .await - { - Ok(response) => response, - Err(status) if status.code() == Code::NotFound => { - return Err(miette::miette!( - "sandbox '{name}' no longer exists; stopping service forward" - )); - } - Err(status) => return Err(status).into_diagnostic(), - }; - - let sandbox = response - .into_inner() - .sandbox - .ok_or_else(|| miette::miette!("sandbox '{name}' not found"))?; - - if SandboxPhase::try_from(sandbox.phase()) != Ok(SandboxPhase::Ready) { - return Err(miette::miette!( - "sandbox '{}' is no longer ready (phase: {}); stopping service forward", - name, - phase_name(sandbox.phase()) - )); - } - - Ok(sandbox) -} - -#[derive(Debug)] -struct ForwardTcpConnectionError { - message: String, - fatal: bool, -} - -impl ForwardTcpConnectionError { - fn transient(message: impl Into) -> Self { - Self { - message: message.into(), - fatal: false, - } - } - - fn from_status(status: Status) -> Self { - let fatal = matches!(status.code(), Code::NotFound | Code::FailedPrecondition); - Self { - message: status.to_string(), - fatal, - } - } -} - -impl std::fmt::Display for ForwardTcpConnectionError { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - f.write_str(&self.message) - } -} - -impl std::error::Error for ForwardTcpConnectionError {} - -fn parse_tcp_forward_spec(local: Option<&str>, default_port: u16) -> Result<(String, u16)> { - let Some(spec) = local else { - return Ok(("127.0.0.1".to_string(), default_port)); - }; - - if let Some(pos) = spec.rfind(':') { - let addr = &spec[..pos]; - let port_str = &spec[pos + 1..]; - if let Ok(port) = port_str.parse::() { - if addr.is_empty() { - return Err(miette::miette!("bind address is required before ':'")); - } - return Ok((addr.to_string(), port)); - } - } - - let port: u16 = spec.parse().map_err(|_| { - miette::miette!("invalid local forward spec '{spec}': expected [bind_address:]port") - })?; - Ok(("127.0.0.1".to_string(), port)) -} - -async fn forward_one_tcp_connection( - client: &mut crate::tls::GrpcClient, - socket: tokio::net::TcpStream, - sandbox_id: String, - target_host: String, - target_port: u16, - service_id: String, - authorization_token: String, -) -> std::result::Result<(), ForwardTcpConnectionError> { - use tokio::io::{AsyncReadExt, AsyncWriteExt}; - use tokio_stream::wrappers::ReceiverStream; - - let (tx, rx) = tokio::sync::mpsc::channel::(16); - tx.send(TcpForwardFrame { - payload: Some(openshell_core::proto::tcp_forward_frame::Payload::Init( - TcpForwardInit { - sandbox_id, - service_id, - target: Some(tcp_forward_init::Target::Tcp(TcpRelayTarget { - host: target_host, - port: u32::from(target_port), - })), - authorization_token, - }, - )), - }) - .await - .map_err(|_| ForwardTcpConnectionError::transient("failed to initialize forward stream"))?; - - let mut response = match client.forward_tcp(ReceiverStream::new(rx)).await { - Ok(response) => response.into_inner(), - Err(status) => { - let err = ForwardTcpConnectionError::from_status(status); - drain_and_shutdown_local_socket(socket).await; - return Err(err); - } - }; - - let (mut local_read, mut local_write) = socket.into_split(); - - let to_gateway = tokio::spawn(async move { - let mut buf = vec![0u8; 64 * 1024]; - loop { - let n = local_read.read(&mut buf).await?; - if n == 0 { - break; - } - if tx - .send(TcpForwardFrame { - payload: Some(openshell_core::proto::tcp_forward_frame::Payload::Data( - buf[..n].to_vec(), - )), - }) - .await - .is_err() - { - break; - } - } - Ok::<(), std::io::Error>(()) - }); - - while let Some(frame) = response - .message() - .await - .map_err(ForwardTcpConnectionError::from_status)? - { - let Some(openshell_core::proto::tcp_forward_frame::Payload::Data(data)) = frame.payload - else { - continue; - }; - if data.is_empty() { - continue; - } - local_write - .write_all(&data) - .await - .map_err(|err| ForwardTcpConnectionError::transient(err.to_string()))?; - } - - let _ = local_write.shutdown().await; - to_gateway.abort(); - Ok(()) -} - -async fn drain_and_shutdown_local_socket(mut socket: tokio::net::TcpStream) { - use tokio::io::{AsyncReadExt, AsyncWriteExt}; - - let mut buf = [0u8; 4096]; - while matches!( - tokio::time::timeout(Duration::from_millis(25), socket.read(&mut buf)).await, - Ok(Ok(n)) if n != 0 - ) {} - let _ = socket.shutdown().await; -} - struct RawModeGuard; impl Drop for RawModeGuard { @@ -3125,460 +2840,134 @@ async fn wait_for_lifecycle_phase( } } -pub async fn service_expose( +// --------------------------------------------------------------------------- +// Workspace commands +// --------------------------------------------------------------------------- + +pub async fn workspace_create( server: &str, - sandbox: &str, - service: &str, - target_port: u16, - workspace: &str, + name: &str, + label_args: &[String], tls: &TlsOptions, ) -> Result<()> { + use openshell_core::proto::CreateWorkspaceRequest; + + let labels = label_args + .iter() + .filter_map(|arg| { + let (k, v) = arg.split_once('=')?; + Some((k.to_string(), v.to_string())) + }) + .collect::>(); + let mut client = grpc_client(server, tls).await?; let response = client - .expose_service(ExposeServiceRequest { - sandbox: sandbox.to_string(), - service: service.to_string(), - target_port: u32::from(target_port), - domain: true, - workspace: workspace.to_string(), + .create_workspace(CreateWorkspaceRequest { + name: name.to_string(), + labels, }) .await - .map_err(service_expose_status_error)? - .into_inner(); + .into_diagnostic()?; - if service.is_empty() { - println!( - "{} Exposed sandbox {} -> 127.0.0.1:{}", - "✓".green().bold(), - sandbox.bold(), - target_port, - ); - } else { + let workspace = response + .into_inner() + .workspace + .ok_or_else(|| miette!("workspace missing from response"))?; + + println!( + "{} Created workspace {}", + "✓".green().bold(), + workspace.object_name().bold() + ); + + Ok(()) +} + +pub async fn workspace_get(server: &str, name: &str, tls: &TlsOptions) -> Result<()> { + use openshell_core::proto::GetWorkspaceRequest; + + let mut client = grpc_client(server, tls).await?; + let response = client + .get_workspace(GetWorkspaceRequest { + name: name.to_string(), + }) + .await + .into_diagnostic()?; + + let workspace = response + .into_inner() + .workspace + .ok_or_else(|| miette!("workspace missing from response"))?; + + println!("{}", "Workspace:".cyan().bold()); + println!(); + println!(" {} {}", "Name:".dimmed(), workspace.object_name()); + if let Some(meta) = &workspace.metadata { + println!(" {} {}", "Id:".dimmed(), meta.id); println!( - "{} Exposed service {} on sandbox {} -> 127.0.0.1:{}", - "✓".green().bold(), - service.bold(), - sandbox.bold(), - target_port, + " {} {}", + "Resource version:".dimmed(), + meta.resource_version ); + if meta.created_at_ms != 0 { + println!( + " {} {}", + "Created:".dimmed(), + format_epoch_ms(meta.created_at_ms) + ); + } + if !meta.labels.is_empty() { + println!( + " {} {}", + "Labels:".dimmed(), + meta.labels + .iter() + .map(|(k, v)| format!("{k}={v}")) + .collect::>() + .join(", ") + ); + } } - if !response.url.is_empty() { - let url = service_url_for_gateway(&response.url, server); - println!(" URL: {}", url.cyan()); - } - Ok(()) -} -fn service_expose_status_error(status: Status) -> miette::Report { - service_status_error("expose service", "sandbox:write", status) + Ok(()) } -#[allow(clippy::too_many_arguments)] // user-facing CLI command -pub async fn service_list( +pub async fn workspace_list( server: &str, - sandbox: Option<&str>, limit: u32, offset: u32, - workspace: &str, - all_workspaces: bool, + label_selector: &str, output: &str, tls: &TlsOptions, ) -> Result<()> { + use openshell_core::proto::ListWorkspacesRequest; + let mut client = grpc_client(server, tls).await?; let response = client - .list_services(ListServicesRequest { - sandbox: sandbox.unwrap_or_default().to_string(), + .list_workspaces(ListWorkspacesRequest { limit, offset, - workspace: if all_workspaces { - String::new() - } else { - workspace.to_string() - }, - all_workspaces, + label_selector: label_selector.to_string(), }) .await - .map_err(|status| service_status_error("list services", "sandbox:read", status))? - .into_inner(); + .into_diagnostic()?; + let workspaces = response.into_inner().workspaces; - let services = response - .services - .iter() - .filter_map(|response| service_endpoint_to_json(response, server)) - .collect::>(); - if crate::output::print_output_collection(output, &services, Clone::clone)? { + if crate::output::print_output_collection(output, &workspaces, workspace_to_json)? { return Ok(()); } - if response.services.is_empty() { - if let Some(sandbox) = sandbox { - println!("No services exposed for sandbox {sandbox}."); - } else { - println!("No services exposed."); - } + if workspaces.is_empty() { + println!("No workspaces found."); return Ok(()); } - print_service_endpoint_table(&response.services, server, all_workspaces); - Ok(()) -} - -pub async fn service_get( - server: &str, - sandbox: &str, - service: &str, - workspace: &str, - tls: &TlsOptions, -) -> Result<()> { - let mut client = grpc_client(server, tls).await?; - let response = client - .get_service(GetServiceRequest { - sandbox: sandbox.to_string(), - service: service.to_string(), - workspace: workspace.to_string(), - }) - .await - .map_err(|status| service_status_error("get service", "sandbox:read", status))? - .into_inner(); - - print_service_endpoint_table(&[response], server, false); - Ok(()) -} - -pub async fn service_delete( - server: &str, - sandbox: &str, - service: &str, - workspace: &str, - tls: &TlsOptions, -) -> Result<()> { - let mut client = grpc_client(server, tls).await?; - let response = client - .delete_service(DeleteServiceRequest { - sandbox: sandbox.to_string(), - service: service.to_string(), - workspace: workspace.to_string(), - }) - .await - .map_err(|status| service_status_error("delete service", "sandbox:write", status))? - .into_inner(); - - if !response.deleted { - return Err(miette!("delete service failed: service endpoint not found")); - } - - if service.is_empty() { - println!( - "{} Deleted exposed sandbox {}", - "✓".green().bold(), - sandbox.bold(), - ); - } else { - println!( - "{} Deleted service {} on sandbox {}", - "✓".green().bold(), - service.bold(), - sandbox.bold(), - ); - } - Ok(()) -} - -fn service_status_error(action: &str, required_scope: &str, status: Status) -> miette::Report { - let message = status.message(); - match status.code() { - Code::PermissionDenied => { - miette!("{action} failed: permission denied (requires {required_scope})") - } - Code::Unauthenticated => miette!("{action} failed: authentication required"), - Code::NotFound if message == "sandbox not found" => { - miette!("{action} failed: sandbox not found") - } - Code::NotFound if message == "service endpoint not found" => { - miette!("{action} failed: service endpoint not found") - } - Code::InvalidArgument if !message.is_empty() => { - miette!("{action} failed: invalid request: {message}") - } - _ => miette!("{action} failed: {status}"), - } -} - -fn print_service_endpoint_table( - services: &[ServiceEndpointResponse], - gateway_endpoint: &str, - all_workspaces: bool, -) { - let rows = services - .iter() - .filter_map(|response| { - let endpoint = response.endpoint.as_ref()?; - let workspace = endpoint - .metadata - .as_ref() - .map_or("", |m| m.workspace.as_str()); - let service = service_display_name(&endpoint.service_name).to_string(); - let target = format!("127.0.0.1:{}", endpoint.target_port); - let url = if response.url.is_empty() { - String::new() - } else { - service_url_for_gateway(&response.url, gateway_endpoint) - }; - Some(( - workspace.to_string(), - endpoint.sandbox_name.clone(), - service, - target, - url, - )) - }) - .collect::>(); - - if rows.is_empty() { - return; - } - - let ws_width = if all_workspaces { - rows.iter() - .map(|(ws, _, _, _, _)| ws.len()) - .max() - .unwrap_or(9) - .max(9) - } else { - 0 - }; - let sandbox_width = rows - .iter() - .map(|(_, sandbox, _, _, _)| sandbox.len()) - .max() - .unwrap_or(7) - .max(7); - let service_width = rows - .iter() - .map(|(_, _, service, _, _)| service.len()) - .max() - .unwrap_or(7) - .max(7); - let target_width = rows - .iter() - .map(|(_, _, _, target, _)| target.len()) - .max() - .unwrap_or(6) - .max(6); - - if all_workspaces { - println!( - "{: Option { - let endpoint = response.endpoint.as_ref()?; - let workspace = endpoint - .metadata - .as_ref() - .map_or("", |metadata| metadata.workspace.as_str()); - let url = if response.url.is_empty() { - String::new() - } else { - service_url_for_gateway(&response.url, gateway_endpoint) - }; - - Some(serde_json::json!({ - "workspace": workspace, - "sandbox": endpoint.sandbox_name, - "service": endpoint.service_name, - "target_port": endpoint.target_port, - "url": url, - })) -} - -fn service_display_name(service: &str) -> &str { - if service.is_empty() { "-" } else { service } -} - -/// Read gcloud Application Default Credentials from disk. -/// -/// Returns `(client_id, client_secret, refresh_token)`. -/// -/// Checks `GOOGLE_APPLICATION_CREDENTIALS` first; falls back to -/// `$CLOUDSDK_CONFIG/application_default_credentials.json` when set, then to -/// `~/.config/gcloud/application_default_credentials.json`. -fn service_url_for_gateway(service_url: &str, gateway_endpoint: &str) -> String { - let (Ok(mut service_url), Ok(gateway_endpoint)) = ( - url::Url::parse(service_url), - url::Url::parse(gateway_endpoint), - ) else { - return service_url.to_string(); - }; - - if service_url - .set_port(gateway_endpoint.port_or_known_default()) - .is_err() - { - return service_url.to_string(); - } - - service_url.to_string() -} - -// --------------------------------------------------------------------------- -// Workspace commands -// --------------------------------------------------------------------------- - -pub async fn workspace_create( - server: &str, - name: &str, - label_args: &[String], - tls: &TlsOptions, -) -> Result<()> { - use openshell_core::proto::CreateWorkspaceRequest; - - let labels = label_args - .iter() - .filter_map(|arg| { - let (k, v) = arg.split_once('=')?; - Some((k.to_string(), v.to_string())) - }) - .collect::>(); - - let mut client = grpc_client(server, tls).await?; - let response = client - .create_workspace(CreateWorkspaceRequest { - name: name.to_string(), - labels, - }) - .await - .into_diagnostic()?; - - let workspace = response - .into_inner() - .workspace - .ok_or_else(|| miette!("workspace missing from response"))?; - - println!( - "{} Created workspace {}", - "✓".green().bold(), - workspace.object_name().bold() - ); - - Ok(()) -} - -pub async fn workspace_get(server: &str, name: &str, tls: &TlsOptions) -> Result<()> { - use openshell_core::proto::GetWorkspaceRequest; - - let mut client = grpc_client(server, tls).await?; - let response = client - .get_workspace(GetWorkspaceRequest { - name: name.to_string(), - }) - .await - .into_diagnostic()?; - - let workspace = response - .into_inner() - .workspace - .ok_or_else(|| miette!("workspace missing from response"))?; - - println!("{}", "Workspace:".cyan().bold()); - println!(); - println!(" {} {}", "Name:".dimmed(), workspace.object_name()); - if let Some(meta) = &workspace.metadata { - println!(" {} {}", "Id:".dimmed(), meta.id); - println!( - " {} {}", - "Resource version:".dimmed(), - meta.resource_version - ); - if meta.created_at_ms != 0 { - println!( - " {} {}", - "Created:".dimmed(), - format_epoch_ms(meta.created_at_ms) - ); - } - if !meta.labels.is_empty() { - println!( - " {} {}", - "Labels:".dimmed(), - meta.labels - .iter() - .map(|(k, v)| format!("{k}={v}")) - .collect::>() - .join(", ") - ); - } - } - - Ok(()) -} - -pub async fn workspace_list( - server: &str, - limit: u32, - offset: u32, - label_selector: &str, - output: &str, - tls: &TlsOptions, -) -> Result<()> { - use openshell_core::proto::ListWorkspacesRequest; - - let mut client = grpc_client(server, tls).await?; - let response = client - .list_workspaces(ListWorkspacesRequest { - limit, - offset, - label_selector: label_selector.to_string(), - }) - .await - .into_diagnostic()?; - let workspaces = response.into_inner().workspaces; - - if crate::output::print_output_collection(output, &workspaces, workspace_to_json)? { - return Ok(()); - } - - if workspaces.is_empty() { - println!("No workspaces found."); - return Ok(()); - } - - let name_width = workspaces - .iter() - .map(|w| w.object_name().len()) - .max() - .unwrap_or(4) - .max(4); + let name_width = workspaces + .iter() + .map(|w| w.object_name().len()) + .max() + .unwrap_or(4) + .max(4); println!( "{: serde_json ); } if !meta.labels.is_empty() { - obj.insert("labels".to_string(), serde_json::json!(meta.labels)); - } - } - obj.insert( - "status".to_string(), - serde_json::json!(workspace_phase_str(workspace)), - ); - serde_json::Value::Object(obj) -} - -#[allow(clippy::too_many_arguments)] -pub async fn gateway_inference_set( - server: &str, - provider_name: &str, - model_id: &str, - route_name: &str, - no_verify: bool, - timeout_secs: u64, - workspace: &str, - tls: &TlsOptions, -) -> Result<()> { - let progress = if std::io::stdout().is_terminal() { - let spinner = ProgressBar::new_spinner(); - spinner.set_style( - ProgressStyle::with_template("{spinner:.cyan} {msg} ({elapsed})") - .unwrap_or_else(|_| ProgressStyle::default_spinner()), - ); - spinner.set_message("Configuring inference..."); - spinner.enable_steady_tick(Duration::from_millis(120)); - Some(spinner) - } else { - None - }; - - let mut client = grpc_inference_client(server, tls).await?; - let response = client - .set_inference_route(SetInferenceRouteRequest { - provider_name: provider_name.to_string(), - model_id: model_id.to_string(), - route_name: route_name.to_string(), - verify: false, - no_verify, - timeout_secs, - workspace: workspace.to_string(), - }) - .await; - - if let Some(progress) = &progress { - progress.finish_and_clear(); - } - - let response = response.map_err(format_inference_status)?; - - let configured = response.into_inner(); - let label = if configured.route_name == "sandbox-system" { - "System inference configured:" - } else { - "Inference configured:" - }; - println!("{}", label.cyan().bold()); - println!(); - println!(" {} {}", "Workspace:".dimmed(), configured.workspace); - println!(" {} {}", "Route:".dimmed(), configured.route_name); - println!(" {} {}", "Provider:".dimmed(), configured.provider_name); - println!(" {} {}", "Model:".dimmed(), configured.model_id); - println!(" {} {}", "Version:".dimmed(), configured.version); - print_timeout(configured.timeout_secs); - if configured.validation_performed { - println!(" {}", "Validated Endpoints:".dimmed()); - for endpoint in configured.validated_endpoints { - println!(" - {} ({})", endpoint.url, endpoint.protocol); - } - } - Ok(()) -} - -#[allow(clippy::too_many_arguments)] -pub async fn gateway_inference_update( - server: &str, - provider_name: Option<&str>, - model_id: Option<&str>, - route_name: &str, - no_verify: bool, - timeout_secs: Option, - workspace: &str, - tls: &TlsOptions, -) -> Result<()> { - if provider_name.is_none() && model_id.is_none() && timeout_secs.is_none() { - return Err(miette::miette!( - "at least one of --provider, --model, or --timeout must be specified" - )); - } - - let mut client = grpc_inference_client(server, tls).await?; - - // Fetch current config to use as base for the partial update. - let current = client - .get_inference_route(GetInferenceRouteRequest { - route_name: route_name.to_string(), - workspace: workspace.to_string(), - }) - .await - .into_diagnostic()? - .into_inner(); - - let provider = provider_name.unwrap_or(¤t.provider_name); - let model = model_id.unwrap_or(¤t.model_id); - let timeout = timeout_secs.unwrap_or(current.timeout_secs); - - let progress = if std::io::stdout().is_terminal() { - let spinner = ProgressBar::new_spinner(); - spinner.set_style( - ProgressStyle::with_template("{spinner:.cyan} {msg} ({elapsed})") - .unwrap_or_else(|_| ProgressStyle::default_spinner()), - ); - spinner.set_message("Configuring inference..."); - spinner.enable_steady_tick(Duration::from_millis(120)); - Some(spinner) - } else { - None - }; - - let response = client - .set_inference_route(SetInferenceRouteRequest { - provider_name: provider.to_string(), - model_id: model.to_string(), - route_name: route_name.to_string(), - verify: false, - no_verify, - timeout_secs: timeout, - workspace: workspace.to_string(), - }) - .await; - - if let Some(progress) = &progress { - progress.finish_and_clear(); - } - - let response = response.map_err(format_inference_status)?; - - let configured = response.into_inner(); - let label = if configured.route_name == "sandbox-system" { - "System inference updated:" - } else { - "Inference updated:" - }; - println!("{}", label.cyan().bold()); - println!(); - println!(" {} {}", "Workspace:".dimmed(), configured.workspace); - println!(" {} {}", "Route:".dimmed(), configured.route_name); - println!(" {} {}", "Provider:".dimmed(), configured.provider_name); - println!(" {} {}", "Model:".dimmed(), configured.model_id); - println!(" {} {}", "Version:".dimmed(), configured.version); - print_timeout(configured.timeout_secs); - if configured.validation_performed { - println!(" {}", "Validated Endpoints:".dimmed()); - for endpoint in configured.validated_endpoints { - println!(" - {} ({})", endpoint.url, endpoint.protocol); - } - } - Ok(()) -} - -pub async fn gateway_inference_get( - server: &str, - route_name: Option<&str>, - workspace: &str, - tls: &TlsOptions, -) -> Result<()> { - let mut client = grpc_inference_client(server, tls).await?; - - if let Some(name) = route_name { - // Show a single route (--system was specified). - let response = client - .get_inference_route(GetInferenceRouteRequest { - route_name: name.to_string(), - workspace: workspace.to_string(), - }) - .await - .into_diagnostic()?; - - let configured = response.into_inner(); - let label = if name == "sandbox-system" { - "System inference:" - } else { - "Inference:" - }; - println!("{}", label.cyan().bold()); - println!(); - println!(" {} {}", "Workspace:".dimmed(), configured.workspace); - println!(" {} {}", "Provider:".dimmed(), configured.provider_name); - println!(" {} {}", "Model:".dimmed(), configured.model_id); - println!(" {} {}", "Version:".dimmed(), configured.version); - print_timeout(configured.timeout_secs); - } else { - // Show both routes by default. - print_inference_route(&mut client, "Inference", "", workspace).await; - println!(); - print_inference_route(&mut client, "System inference", "sandbox-system", workspace).await; - } - Ok(()) -} - -pub async fn gateway_inference_delete( - server: &str, - route_name: &str, - workspace: &str, - tls: &TlsOptions, -) -> Result<()> { - let mut client = grpc_inference_client(server, tls).await?; - - let response = client - .delete_inference_route(DeleteInferenceRouteRequest { - route_name: route_name.to_string(), - workspace: workspace.to_string(), - }) - .await - .into_diagnostic()?; - - let label = if route_name == "sandbox-system" { - "System inference route" - } else { - "Inference route" - }; - - if response.into_inner().deleted { - println!("{label} deleted."); - } else { - println!("{label} not found (already deleted)."); - } - Ok(()) -} - -async fn print_inference_route( - client: &mut crate::tls::GrpcInferenceClient, - label: &str, - route_name: &str, - workspace: &str, -) { - match client - .get_inference_route(GetInferenceRouteRequest { - route_name: route_name.to_string(), - workspace: workspace.to_string(), - }) - .await - { - Ok(response) => { - let configured = response.into_inner(); - println!("{}", format!("{label}:").cyan().bold()); - println!(); - println!(" {} {}", "Workspace:".dimmed(), configured.workspace); - println!(" {} {}", "Provider:".dimmed(), configured.provider_name); - println!(" {} {}", "Model:".dimmed(), configured.model_id); - println!(" {} {}", "Version:".dimmed(), configured.version); - print_timeout(configured.timeout_secs); - } - Err(e) if e.code() == Code::NotFound => { - println!("{}", format!("{label}:").cyan().bold()); - println!(); - println!(" {}", "Not configured".dimmed()); - } - Err(e) => { - println!("{}", format!("{label}:").cyan().bold()); - println!(); - println!(" {} {}", "Error:".red(), e.message()); - } - } -} - -fn print_timeout(timeout_secs: u64) { - if timeout_secs == 0 { - println!(" {} {}s (default)", "Timeout:".dimmed(), 60); - } else { - println!(" {} {}s", "Timeout:".dimmed(), timeout_secs); - } -} - -fn format_inference_status(status: Status) -> miette::Report { - let message = status.message().trim(); - - if message.is_empty() { - return miette::miette!("inference configuration failed ({})", status.code()); + obj.insert("labels".to_string(), serde_json::json!(meta.labels)); + } } - - miette::miette!("{message}") + obj.insert( + "status".to_string(), + serde_json::json!(workspace_phase_str(workspace)), + ); + serde_json::Value::Object(obj) } pub fn git_repo_root(local_path: &Path) -> Result { @@ -4351,314 +3463,6 @@ pub async fn sandbox_policy_set_global( Ok(()) } -pub async fn sandbox_settings_get( - server: &str, - name: &str, - json: bool, - workspace: &str, - tls: &TlsOptions, -) -> Result<()> { - let mut client = grpc_client(server, tls).await?; - let sandbox = client - .get_sandbox(GetSandboxRequest { - name: name.to_string(), - workspace: workspace.to_string(), - }) - .await - .into_diagnostic()? - .into_inner() - .sandbox - .ok_or_else(|| miette::miette!("sandbox not found"))?; - - let response = client - .get_sandbox_config(GetSandboxConfigRequest { - sandbox_id: sandbox.object_id().to_string(), - }) - .await - .into_diagnostic()? - .into_inner(); - - if json { - let obj = settings_to_json_sandbox(name, workspace, &response); - println!("{}", serde_json::to_string_pretty(&obj).into_diagnostic()?); - return Ok(()); - } - - let policy_source = if response.policy_source == PolicySource::Global as i32 { - "global" - } else { - "sandbox" - }; - - println!("Sandbox: {name}"); - println!("Config Rev: {}", response.config_revision); - println!("Policy Source: {policy_source}"); - println!("Policy Hash: {}", response.policy_hash); - - if response.settings.is_empty() { - println!("Settings: No settings available."); - return Ok(()); - } - - println!("Settings:"); - let mut keys: Vec<_> = response.settings.keys().cloned().collect(); - keys.sort(); - for key in keys { - if let Some(setting) = response.settings.get(&key) { - let scope = match SettingScope::try_from(setting.scope) { - Ok(SettingScope::Global) => "global", - Ok(SettingScope::Sandbox) => "sandbox", - _ => "unset", - }; - println!( - " {} = {} ({})", - key, - format_setting_value(setting.value.as_ref()), - scope - ); - } - } - - Ok(()) -} - -pub async fn gateway_settings_get(server: &str, json: bool, tls: &TlsOptions) -> Result<()> { - let mut client = grpc_client(server, tls).await?; - let response = client - .get_gateway_config(GetGatewayConfigRequest {}) - .await - .into_diagnostic()? - .into_inner(); - - if json { - let obj = settings_to_json_global(&response); - println!("{}", serde_json::to_string_pretty(&obj).into_diagnostic()?); - return Ok(()); - } - - println!("Scope: global"); - println!("Settings Rev: {}", response.settings_revision); - - if response.settings.is_empty() { - println!("Settings: No settings available."); - return Ok(()); - } - - println!("Settings:"); - let mut keys: Vec<_> = response.settings.keys().cloned().collect(); - keys.sort(); - for key in keys { - if let Some(setting) = response.settings.get(&key) { - println!(" {} = {}", key, format_setting_value(Some(setting))); - } - } - Ok(()) -} - -fn settings_to_json_sandbox( - name: &str, - workspace: &str, - response: &GetSandboxConfigResponse, -) -> serde_json::Value { - let policy_source = if response.policy_source == PolicySource::Global as i32 { - "global" - } else { - "sandbox" - }; - - let mut settings = serde_json::Map::new(); - let mut keys: Vec<_> = response.settings.keys().cloned().collect(); - keys.sort(); - for key in keys { - if let Some(setting) = response.settings.get(&key) { - let scope = match SettingScope::try_from(setting.scope) { - Ok(SettingScope::Global) => "global", - Ok(SettingScope::Sandbox) => "sandbox", - _ => "unset", - }; - settings.insert( - key, - serde_json::json!({ - "value": format_setting_value(setting.value.as_ref()), - "scope": scope, - }), - ); - } - } - - serde_json::json!({ - "sandbox": name, - "workspace": workspace, - "config_revision": response.config_revision, - "policy_source": policy_source, - "policy_hash": response.policy_hash, - "settings": settings, - }) -} - -fn settings_to_json_global( - response: &openshell_core::proto::GetGatewayConfigResponse, -) -> serde_json::Value { - let mut settings = serde_json::Map::new(); - let mut keys: Vec<_> = response.settings.keys().cloned().collect(); - keys.sort(); - for key in keys { - if let Some(setting) = response.settings.get(&key) { - settings.insert(key, serde_json::json!(format_setting_value(Some(setting)))); - } - } - - serde_json::json!({ - "scope": "global", - "settings_revision": response.settings_revision, - "settings": settings, - }) -} - -pub async fn gateway_setting_set( - server: &str, - key: &str, - value: &str, - yes: bool, - workspace: &str, - tls: &TlsOptions, -) -> Result<()> { - let setting_value = parse_cli_setting_value(key, value)?; - confirm_global_setting_takeover(key, yes)?; - - let mut client = grpc_client(server, tls).await?; - let response = client - .update_config(UpdateConfigRequest { - name: String::new(), - setting_key: key.to_string(), - setting_value: Some(setting_value), - global: true, - workspace: workspace.to_string(), - ..Default::default() - }) - .await - .into_diagnostic()? - .into_inner(); - - println!( - "{} Set global setting {}={} (revision {})", - "✓".green().bold(), - key, - value, - response.settings_revision - ); - Ok(()) -} - -pub async fn sandbox_setting_set( - server: &str, - name: &str, - key: &str, - value: &str, - workspace: &str, - tls: &TlsOptions, -) -> Result<()> { - let setting_value = parse_cli_setting_value(key, value)?; - - let mut client = grpc_client(server, tls).await?; - let response = client - .update_config(UpdateConfigRequest { - name: name.to_string(), - setting_key: key.to_string(), - setting_value: Some(setting_value), - workspace: workspace.to_string(), - ..Default::default() - }) - .await - .into_diagnostic()? - .into_inner(); - - println!( - "{} Set sandbox setting {}={} for {} (revision {})", - "✓".green().bold(), - key, - value, - name, - response.settings_revision - ); - Ok(()) -} - -pub async fn gateway_setting_delete( - server: &str, - key: &str, - yes: bool, - workspace: &str, - tls: &TlsOptions, -) -> Result<()> { - confirm_global_setting_delete(key, yes)?; - - let mut client = grpc_client(server, tls).await?; - let response = client - .update_config(UpdateConfigRequest { - name: String::new(), - setting_key: key.to_string(), - delete_setting: true, - global: true, - workspace: workspace.to_string(), - ..Default::default() - }) - .await - .into_diagnostic()? - .into_inner(); - - if response.deleted { - println!( - "{} Deleted global setting {} (revision {})", - "✓".green().bold(), - key, - response.settings_revision - ); - } else { - println!("{} Global setting {} not found", "!".yellow(), key); - } - Ok(()) -} - -pub async fn sandbox_setting_delete( - server: &str, - name: &str, - key: &str, - workspace: &str, - tls: &TlsOptions, -) -> Result<()> { - let mut client = grpc_client(server, tls).await?; - let response = client - .update_config(UpdateConfigRequest { - name: name.to_string(), - setting_key: key.to_string(), - delete_setting: true, - workspace: workspace.to_string(), - ..Default::default() - }) - .await - .into_diagnostic()? - .into_inner(); - - if response.deleted { - println!( - "{} Deleted sandbox setting {} for {} (revision {})", - "✓".green().bold(), - key, - name, - response.settings_revision - ); - } else { - println!( - "{} Sandbox setting {} not found for {}", - "!".yellow(), - key, - name, - ); - } - Ok(()) -} - pub async fn sandbox_policy_set( server: &str, name: &str, @@ -5967,11 +4771,10 @@ mod tests { use super::{ PolicyGetView, ProvisioningStep, build_sandbox_resource_limits, dockerfile_sources_supported_for_gateway, format_endpoint, format_log_line, git_sync_files, - has_main_process_result, parse_cli_setting_value, parse_credential_expiry_cli_value, - parse_driver_config_json, parse_secret_material_env_pairs, policy_revision_list_json, - policy_revision_to_json, provisioning_timeout_message, ready_false_condition_message, - resolve_from, sandbox_should_persist, sandbox_upload_plan, service_endpoint_to_json, - service_expose_status_error, service_url_for_gateway, workspace_member_to_json, + has_main_process_result, parse_credential_expiry_cli_value, parse_driver_config_json, + parse_secret_material_env_pairs, policy_revision_list_json, policy_revision_to_json, + provisioning_timeout_message, ready_false_condition_message, resolve_from, + sandbox_should_persist, sandbox_upload_plan, workspace_member_to_json, }; use crate::TEST_ENV_LOCK; use crate::commands::common::{ @@ -5981,7 +4784,6 @@ mod tests { use std::fs; use std::path::Path; use std::process::Command; - use tonic::Status; use openshell_bootstrap::GatewayMetadata; use openshell_core::progress::{ @@ -5993,8 +4795,7 @@ mod tests { ResourceRequirements, Sandbox, SandboxCondition, SandboxPhase, SandboxPolicy, SandboxPolicyRevision, SandboxResources, SandboxStatus, SandboxWorkloadConfig, SandboxWorkloadTemplate, SandboxWorkloadTemplateProvenance, SandboxWorkloadTemplateSpec, - ServiceEndpoint, ServiceEndpointResponse, WorkspaceMember, WorkspaceRole, - datamodel::v1::ObjectMeta, + WorkspaceMember, WorkspaceRole, datamodel::v1::ObjectMeta, }; #[test] @@ -6077,37 +4878,6 @@ mod tests { assert!(unknown[0].get("sandbox").is_none()); } - #[test] - fn service_endpoint_json_has_raw_fields_and_normalized_url() { - let response = ServiceEndpointResponse { - endpoint: Some(ServiceEndpoint { - metadata: Some(ObjectMeta { - workspace: "team-a".to_string(), - ..Default::default() - }), - sandbox_name: "api".to_string(), - service_name: String::new(), - target_port: 8080, - ..Default::default() - }), - url: "https://api.openshell.localhost:3000/".to_string(), - }; - - let value = service_endpoint_to_json(&response, "https://gateway.example:17670") - .expect("service endpoint JSON"); - assert_eq!( - value, - serde_json::json!({ - "workspace": "team-a", - "sandbox": "api", - "service": "", - "target_port": 8080, - "url": "https://api.openshell.localhost:17670/", - }) - ); - assert!(service_endpoint_to_json(&ServiceEndpointResponse::default(), "unused").is_none()); - } - #[test] fn workspace_member_json_uses_stable_role_names() { for (role, expected) in [ @@ -6297,37 +5067,6 @@ mod tests { assert_eq!(progress_step_from_metadata("driver-private-step"), None); } - #[test] - fn parse_cli_setting_value_parses_bool_aliases() { - let yes_value = parse_cli_setting_value("ocsf_json_enabled", "yes").expect("parse yes"); - assert_eq!( - yes_value.value, - Some(openshell_core::proto::setting_value::Value::BoolValue(true)) - ); - - let zero_value = parse_cli_setting_value("ocsf_json_enabled", "0").expect("parse 0"); - assert_eq!( - zero_value.value, - Some(openshell_core::proto::setting_value::Value::BoolValue( - false - )) - ); - } - - #[test] - fn parse_cli_setting_value_rejects_invalid_bool() { - let err = parse_cli_setting_value("ocsf_json_enabled", "maybe") - .expect_err("invalid bool should fail"); - assert!(err.to_string().contains("invalid bool value")); - } - - #[test] - fn parse_cli_setting_value_rejects_unknown_key() { - let err = - parse_cli_setting_value("unknown_key", "value").expect_err("unknown key should fail"); - assert!(err.to_string().contains("unknown setting key")); - } - #[test] fn build_sandbox_resource_limits_sets_limits_only() { let resources = build_sandbox_resource_limits(Some("500m"), Some("2Gi")) @@ -6582,62 +5321,6 @@ mod tests { assert!(dockerfile_sources_supported_for_gateway(None)); } - #[test] - fn service_url_for_gateway_uses_external_gateway_port() { - assert_eq!( - service_url_for_gateway( - "https://quiet-flamingo--notebook.navigator.openshell.localhost:8080/", - "https://127.0.0.1:31886" - ), - "https://quiet-flamingo--notebook.navigator.openshell.localhost:31886/" - ); - } - - #[test] - fn service_url_for_gateway_omits_default_external_port() { - assert_eq!( - service_url_for_gateway( - "https://quiet-flamingo--notebook.navigator.openshell.localhost:8080/", - "https://gateway.example.com" - ), - "https://quiet-flamingo--notebook.navigator.openshell.localhost/" - ); - } - - #[test] - fn service_url_for_gateway_preserves_service_scheme() { - assert_eq!( - service_url_for_gateway( - "http://quiet-flamingo--notebook.navigator.openshell.localhost:8080/", - "https://127.0.0.1:31886" - ), - "http://quiet-flamingo--notebook.navigator.openshell.localhost:31886/" - ); - } - - #[test] - fn service_url_for_gateway_uses_gateway_default_port() { - assert_eq!( - service_url_for_gateway( - "http://quiet-flamingo--notebook.navigator.openshell.localhost:8080/", - "https://gateway.example.com" - ), - "http://quiet-flamingo--notebook.navigator.openshell.localhost:443/" - ); - } - - #[test] - fn service_expose_status_error_mentions_required_scope() { - let report = service_expose_status_error(Status::permission_denied( - "scope 'sandbox:write' required", - )); - - assert_eq!( - report.to_string(), - "expose service failed: permission denied (requires sandbox:write)" - ); - } - #[test] fn ready_false_condition_message_prefers_reason_and_message() { let status = SandboxStatus {