diff --git a/apps/rocm/src/main.rs b/apps/rocm/src/main.rs index 2de7389a..2866f585 100644 --- a/apps/rocm/src/main.rs +++ b/apps/rocm/src/main.rs @@ -11,6 +11,7 @@ mod endpoint_keys; mod logging; mod provider_keys; mod providers; +mod serve_port; mod serve_summary; mod storage; mod therock; @@ -362,9 +363,16 @@ rocm serve qwen2.5-7b-instruct --verbose --device gpu_required")] /// Host address to bind. #[arg(long, default_value = DEFAULT_LOCAL_HOST)] host: String, - /// TCP port to bind. - #[arg(long, default_value_t = rocm_core::DEFAULT_LOCAL_PORT)] - port: u16, + /// TCP port to bind. Omit it to choose an available port automatically, + /// starting at 11435, on the default 127.0.0.1 host. Any other --host + /// requires an explicit --port: proactive collision detection is claimed + /// only for local 127.0.0.1 serving. + // `allow_hyphen_values` so a negative value such as `-1` reaches + // `parse_serve_port` and gets the domain error naming the valid range, + // instead of clap rejecting it as an unknown flag. Mirrors the same + // choice made for `--gpu-memory-utilization` below. + #[arg(long, value_parser = serve_port::parse_serve_port, allow_hyphen_values = true)] + port: Option, /// Attach to the server in this terminal and stream its logs (Ctrl-D to /// detach and leave it running, Ctrl-C to stop). Same as --verbose. #[arg(long, conflicts_with = "managed")] @@ -4709,7 +4717,8 @@ struct ServeArgs { runtime_id: Option, env_id: Option, host: String, - port: u16, + /// `None` when `--port` was omitted: the allocation transaction chooses one. + port: Option, foreground: bool, managed: bool, verbose: bool, @@ -4747,6 +4756,11 @@ fn serve(args: ServeArgs) -> Result<()> { } = args; let _ = managed; // background is now the default; --managed is accepted as an explicit synonym. validate_bind_host(&host, allow_public_bind)?; + // Automatic selection is a canonical-loopback-only capability. Refuse it on a + // custom host here — before engine resolution, GPU probing, or any side + // effect — so the user gets the `--port` guidance immediately. + let port_request = serve_port::PortRequest::from_flag(port); + serve_port::validate_port_request(&host, port_request)?; // Loopback stays credential-free; a public bind must be authenticated. Resolve // (or generate) the endpoint key now so every downstream path — engine spawn, // readiness probe, smoke test, and the client-config we print — shares one value. @@ -4928,7 +4942,13 @@ fn serve(args: ServeArgs) -> Result<()> { println!(" engine: {selected_engine}"); println!("{}", serve_engine_selection_line(&serve_engine)); println!(" host: {host}"); - println!(" port: {port}"); + // The plan is printed before the allocation transaction runs, so an + // automatic request cannot honestly name a port yet. The launch output + // below prints the resolved endpoint. + match port { + Some(port) => println!(" port: {port}"), + None => println!(" {}", serve_port::AUTOMATIC_PORT_DISCLOSURE), + } if let Some(runtime_id) = resolved_selection.runtime_id.as_deref() { println!(" runtime_id: {runtime_id}"); } @@ -5007,7 +5027,7 @@ fn serve(args: ServeArgs) -> Result<()> { &model, &resolve, &host, - port, + port_request, &resolve.device_policy, &gpu_indices, managed_runtime_id.as_deref(), @@ -5087,7 +5107,7 @@ fn serve(args: ServeArgs) -> Result<()> { &model, &resolve, &host, - port, + port_request, &gpu_indices, resolved_selection.runtime_id.as_deref(), resolved_selection.env_id.as_deref(), @@ -5373,11 +5393,132 @@ enum ManagedSpawn { }, } +/// Outcome of the locked allocation step. +/// +/// `Debug` so tests can assert on `Result::expect_err`. +#[derive(Debug)] +enum ManagedPortDecision { + /// An equivalent managed service is already live; nothing may be spawned. + /// Boxed to keep the variants a similar size (clippy::large_enum_variant). + AlreadyLive(Box), + /// The concrete port this launch owns until it hands it to the child. + Lease(rocm_core::LoopbackPortLease), +} + +/// Live managed services, with liveness refreshed by [`load_managed_services`] +/// (dead PIDs demote to "stopped" and therefore neither satisfy the duplicate +/// check nor reserve a port). +/// +/// A read failure degrades to "nothing is live": the OS preflight still guards +/// the port, and refusing to serve because the services directory could not be +/// listed would be a worse failure than launching. +fn live_managed_services(paths: &AppPaths) -> Vec { + load_managed_services(paths) + .unwrap_or_default() + .into_iter() + .filter(managed_service_is_live) + .collect() +} + +/// Ports that live managed services still own. +/// +/// Conservative by design: a record reserves its port regardless of the host it +/// recorded, because a `0.0.0.0` (or `localhost`) listener also blocks the +/// loopback address a new local server would bind. Over-reserving costs one +/// candidate in a 101-port scan; under-reserving costs a collision. +fn reserved_service_ports(live: &[ManagedServiceRecord]) -> Vec { + live.iter() + .map(|record| serve_port::ReservedPort { + port: record.port, + service_id: record.service_id.clone(), + status: record.status.clone(), + }) + .collect() +} + +/// The decision half of the launch transaction: refresh liveness, apply the +/// `(engine, canonical_model_id)` idempotency guard, then lease a concrete port +/// against both live records and the real OS. +/// +/// Callers MUST hold the shared allocation lock across this call *and* the +/// record publication that follows, or two launches can pick the same candidate. +/// `probe` is injected so tests can drive deterministic bind outcomes without a +/// GPU or a spawn. +fn resolve_managed_port_in_transaction( + paths: &AppPaths, + engine: &str, + canonical_model_id: &str, + host: &str, + request: serve_port::PortRequest, + probe: &mut dyn FnMut(u16) -> std::io::Result, +) -> Result { + let live = live_managed_services(paths); + // Idempotency first: an already-live equivalent service means no port is + // needed at all, so it must be answered before any candidate is leased. + if let Some(existing) = existing_live_managed_service_in(&live, engine, canonical_model_id) { + return Ok(ManagedPortDecision::AlreadyLive(Box::new(existing.clone()))); + } + let reserved = reserved_service_ports(&live); + serve_port::resolve_serve_port(host, request, &reserved, probe).map(ManagedPortDecision::Lease) +} + +/// Grace given to a child that must be terminated because its record could not +/// be published. Short: the child is milliseconds old and has nothing to flush. +const MANAGED_LAUNCH_CLEANUP_GRACE: Duration = Duration::from_secs(2); + +/// Undo a failed managed launch attempt before the allocation lock is released, +/// so no failed transaction leaves a live reservation or a stray endpoint key. +/// +/// Order matters: a spawned child is terminated and confirmed dead *first*, so +/// nothing is still holding the port or writing to the log when the artifacts — +/// including the record that reserves the port for other launches — are removed. +/// Every step is best-effort and idempotent; cleanup must never mask the +/// original failure with a second one. +fn abandon_failed_managed_launch( + paths: &AppPaths, + record: &ManagedServiceRecord, + child_pid: Option, +) { + if let Some(pid) = child_pid.filter(|pid| *pid != 0) { + let outcome = rocm_core::terminate_verified( + &rocm_core::ProcessIdentity::capture(pid), + rocm_core::KillScope::Tree, + MANAGED_LAUNCH_CLEANUP_GRACE, + true, + ); + record_cli_audit_event( + paths, + "service", + "managed_service_launch_cleanup", + "warn", + format!( + "terminated engine child pid={pid} after a failed launch publication outcome={}", + outcome.as_str() + ), + Some(&record.service_id), + ); + } + for path in [ + record.manifest_path.as_path(), + record.log_path.as_path(), + record.engine_state_path.as_path(), + ] { + let _ = fs::remove_file(path); + } + endpoint_keys::clear_endpoint_api_key(paths, &record.service_id); +} + /// Spawn the detached engine child shared by the managed (background) and /// attached (`--verbose`/`--foreground`) serve paths. Returns before the HTTP /// readiness wait; callers decide whether to block on readiness /// ([`start_managed_service`]) or start tailing the log immediately /// ([`run_attached_service`]). +/// +/// The whole body is one serialized transaction under the shared allocation +/// lock: duplicate detection, live-reservation refresh, candidate leasing, +/// manifest/log creation, argv preparation, the bounded immediate-exit check, +/// and record publication. Readiness deliberately runs after the lock is +/// released. #[allow(clippy::too_many_arguments)] fn spawn_managed_engine_child( paths: &AppPaths, @@ -5386,7 +5527,7 @@ fn spawn_managed_engine_child( requested_model: &str, resolve: &ResolveModelResponse, host: &str, - port: u16, + port_request: serve_port::PortRequest, device_policy: &DevicePolicy, gpu_indices: &[u32], runtime_id: Option<&str>, @@ -5396,6 +5537,12 @@ fn spawn_managed_engine_child( paths.ensure()?; fs::create_dir_all(paths.services_dir())?; + // Concurrent `rocm serve` launches and `rocmd` recovery all contend here, so + // no two of them can choose the same port or publish competing records for + // one engine+model. Bounded: a stuck peer produces an actionable error + // rather than a hang. Released when this guard drops, and by process death. + let _allocation = rocm_core::lock_service_allocation(paths)?; + // Idempotency guard: if a managed service for this engine+model is already // alive, surface it and spawn nothing. A second `serve --managed` (e.g. the // chat assistant re-issuing the same request) is treated as satisfied, not @@ -5406,37 +5553,46 @@ fn spawn_managed_engine_child( .map(serde_json::to_string) .transpose() .context("failed to encode engine recipe hint")?; - if let Some(existing) = - existing_live_managed_service(paths, engine, &resolve.canonical_model_id) - { - if existing.engine_recipe_json != requested_recipe_json { - bail!( - "managed service `{}` is already running for engine `{engine}` and model `{}` with different serve options (recipe hint, tool-call parser, or generation defaults); stop it and run `rocm serve` again to apply the requested options", - existing.service_id, - resolve.canonical_model_id + let lease = match resolve_managed_port_in_transaction( + paths, + engine, + &resolve.canonical_model_id, + host, + port_request, + &mut serve_port::loopback_bind_probe, + )? { + ManagedPortDecision::AlreadyLive(existing) => { + let existing = *existing; + if existing.engine_recipe_json != requested_recipe_json { + bail!( + "managed service `{}` is already running for engine `{engine}` and model `{}` with different serve options (recipe hint, tool-call parser, or generation defaults); stop it and run `rocm serve` again to apply the requested options", + existing.service_id, + resolve.canonical_model_id + ); + } + record_cli_audit_event( + paths, + "service", + "managed_service_launch_skipped", + "info", + format!( + "skipped duplicate managed launch engine={engine} model={} existing_service_id={} status={}", + resolve.canonical_model_id, existing.service_id, existing.status + ), + Some(&existing.service_id), ); - } - record_cli_audit_event( - paths, - "service", - "managed_service_launch_skipped", - "info", - format!( - "skipped duplicate managed launch engine={engine} model={} existing_service_id={} status={}", - resolve.canonical_model_id, existing.service_id, existing.status - ), - Some(&existing.service_id), - ); - return Ok(ManagedSpawn::AlreadyRunning(ManagedLaunchReport { - service_id: existing.service_id, - endpoint_url: existing.endpoint_url, - status: existing.status, - already_running: true, - child_pid: None, - log_path: None, - manifest_path: None, - })); - } + return Ok(ManagedSpawn::AlreadyRunning(ManagedLaunchReport { + service_id: existing.service_id, + endpoint_url: existing.endpoint_url, + status: existing.status, + already_running: true, + child_pid: None, + log_path: None, + manifest_path: None, + })); + } + ManagedPortDecision::Lease(lease) => lease, + }; let mut record = ManagedServiceRecord::new( paths, @@ -5445,7 +5601,7 @@ fn spawn_managed_engine_child( requested_model, resolve.canonical_model_id.clone(), host, - port, + lease.port(), "managed", 0, runtime_id.map(str::to_owned), @@ -5454,8 +5610,61 @@ fn spawn_managed_engine_child( ); record.gpu_indices = gpu_indices.to_vec(); record.engine_recipe_json = requested_recipe_json; + // Publishing the record inside the lock is what reserves the leased port for + // every other transaction. From here on, any failure must remove it again. record.write()?; + match launch_managed_engine_child_locked( + paths, + &mut record, + lease, + engine, + service_id, + resolve, + host, + device_policy, + gpu_indices, + runtime_id, + env_id, + engine_recipe, + ) { + Ok(child_pid) => Ok(ManagedSpawn::Spawned { + record: Box::new(record), + child_pid, + }), + Err(error) => { + // Pre-spawn failures reach here with nothing running; a post-spawn + // publication failure already terminated its child, and the removals + // are idempotent. + abandon_failed_managed_launch(paths, &record, None); + Err(error) + } + } +} + +/// The side-effecting half of the launch transaction: create the log, prepare +/// concrete argv, hand the leased port to the child, spawn it, reject a child +/// that has already exited, and publish the running record. +/// +/// Split out of [`spawn_managed_engine_child`] so that *every* failure below the +/// record's first write is funnelled through one cleanup path before the +/// allocation lock is released. Returns the child pid. +#[allow(clippy::too_many_arguments)] +fn launch_managed_engine_child_locked( + paths: &AppPaths, + record: &mut ManagedServiceRecord, + lease: rocm_core::LoopbackPortLease, + engine: &str, + service_id: &str, + resolve: &ResolveModelResponse, + host: &str, + device_policy: &DevicePolicy, + gpu_indices: &[u32], + runtime_id: Option<&str>, + env_id: Option<&str>, + engine_recipe: Option<&EngineRecipeHint>, +) -> Result { + let port = record.port; if let Some(parent) = record.engine_state_path.parent() { fs::create_dir_all(parent) .with_context(|| format!("failed to create {}", parent.display()))?; @@ -5494,6 +5703,12 @@ fn spawn_managed_engine_child( // for managed spawns, so enforce the invariant here too rather than relying // on every future caller having done so. ensure_public_service_has_endpoint_key(host, endpoint_key_file.is_some())?; + // Hand the port over: close the probe socket immediately before the spawn, + // still under the allocation lock, so no other ROCm transaction can take it + // in the meantime. On a custom host the lease never held a socket and this + // is a no-op — the engine owns that bind. + debug_assert_eq!(lease.port(), port); + lease.release(); #[cfg(windows)] let child_pid = { let env_values = app_path_env_var_values(paths, engine_envs_root.as_deref()); @@ -5539,12 +5754,15 @@ fn spawn_managed_engine_child( // verifies this exact process rather than a recycled PID. record.supervisor_start_ticks = rocm_core::process_start_ticks(child_pid); record.status = "running".to_owned(); - record.write()?; + // A surviving child is `running`, never `ready`: it holds a process, not a + // proven endpoint. If it cannot be recorded it would be unsupervised and + // invisible while holding the port, so terminate it rather than leak it. + if let Err(error) = record.write() { + abandon_failed_managed_launch(paths, record, Some(child_pid)); + return Err(error.context("failed to publish the managed service record")); + } - Ok(ManagedSpawn::Spawned { - record: Box::new(record), - child_pid, - }) + Ok(child_pid) } #[allow(clippy::too_many_arguments)] @@ -5554,7 +5772,7 @@ fn start_managed_service( requested_model: &str, resolve: &ResolveModelResponse, host: &str, - port: u16, + port_request: serve_port::PortRequest, device_policy: &DevicePolicy, gpu_indices: &[u32], runtime_id: Option<&str>, @@ -5571,7 +5789,7 @@ fn start_managed_service( requested_model, resolve, host, - port, + port_request, device_policy, gpu_indices, runtime_id, @@ -5581,6 +5799,9 @@ fn start_managed_service( ManagedSpawn::AlreadyRunning(report) => return Ok(report), ManagedSpawn::Spawned { record, child_pid } => (*record, child_pid), }; + // The transaction resolved the request to one concrete port; everything + // below — readiness, the endpoint URL, the audit line — uses that. + let port = record.port; #[cfg(windows)] thread::sleep(Duration::from_millis(200)); @@ -5760,7 +5981,7 @@ fn run_attached_service( requested_model: &str, resolve: &ResolveModelResponse, host: &str, - port: u16, + port_request: serve_port::PortRequest, gpu_indices: &[u32], runtime_id: Option<&str>, env_id: Option<&str>, @@ -5775,7 +5996,7 @@ fn run_attached_service( requested_model, resolve, host, - port, + port_request, &resolve.device_policy, gpu_indices, runtime_id, @@ -5783,7 +6004,7 @@ fn run_attached_service( resolve.engine_recipe.as_ref(), )?; - let (service_id, log_path, child_pid) = match spawn { + let (service_id, log_path, port, child_pid) = match spawn { // A server for this engine+model is already live. Don't fight it for the // port — point the user at the existing one instead of tailing a log we // did not start. @@ -5797,9 +6018,14 @@ fn run_attached_service( drop_orphaned_endpoint_key_on_already_running(&paths, service_id, endpoint_api_key); return Ok(()); } - ManagedSpawn::Spawned { record, child_pid } => { - (service_id.to_owned(), record.log_path.clone(), child_pid) - } + // Attached mode is a managed record too, so it shares the transaction's + // resolved concrete port. + ManagedSpawn::Spawned { record, child_pid } => ( + service_id.to_owned(), + record.log_path.clone(), + record.port, + child_pid, + ), }; // The child is a managed service that outlives this session once detached, so @@ -14671,30 +14897,29 @@ pub(crate) fn managed_service_is_live(record: &ManagedServiceRecord) -> bool { } /// Idempotency guard for managed launches: returns an already-live managed -/// service for this engine+model, if any. The caller compares its effective -/// launch recipe before deciding whether reuse is safe. +/// service for this engine+model, if any, from an already-refreshed live set. +/// The caller compares its effective launch recipe before deciding whether reuse +/// is safe. /// /// Keyed on `(engine, canonical_model_id)` — NOT `service_id`. `generate_service_id` /// embeds `unix_time_millis()`, so every launch mints a unique id; matching on it -/// would never catch a duplicate. `load_managed_services` refreshes liveness, so +/// would never catch a duplicate. [`live_managed_services`] refreshes liveness, so /// stale manifests (dead PIDs) demote to "stopped" and are skipped, letting a /// genuine relaunch proceed. Records are sorted newest-first, so `find` returns /// the newest live match. Prevents a second `serve --managed` for the same /// engine+model from spawning a duplicate process once the TUI job-bridge guard /// has cleared. -fn existing_live_managed_service( - paths: &AppPaths, +/// +/// Takes the snapshot rather than the paths so the allocation transaction loads +/// it once and uses it for both this check and the port reservations: duplicate +/// detection and candidate selection can never observe different states. +fn existing_live_managed_service_in<'a>( + live: &'a [ManagedServiceRecord], engine: &str, canonical_model_id: &str, -) -> Option { - load_managed_services(paths) - .ok()? - .into_iter() - .find(|record| { - record.engine == engine - && record.canonical_model_id == canonical_model_id - && managed_service_is_live(record) - }) +) -> Option<&'a ManagedServiceRecord> { + live.iter() + .find(|record| record.engine == engine && record.canonical_model_id == canonical_model_id) } fn managed_service_running_state(status: &str) -> &'static str { @@ -18095,6 +18320,76 @@ mod tests { } } + #[test] + fn serve_without_port_flag_requests_automatic_selection() { + let cli = parse_serve(&[]).expect("bare serve parses"); + match cli.command { + Some(Command::Serve { port, host, .. }) => { + assert_eq!(port, None, "omitted --port must stay unresolved"); + assert_eq!(host, DEFAULT_LOCAL_HOST); + assert!(serve_port::PortRequest::from_flag(port).is_auto()); + } + other => panic!("expected Serve, got {other:?}"), + } + } + + #[test] + fn serve_with_explicit_port_keeps_it_concrete() { + let cli = parse_serve(&["--port", "8000"]).expect("explicit port parses"); + match cli.command { + Some(Command::Serve { port, .. }) => { + assert_eq!(port, Some(8000)); + assert_eq!( + serve_port::PortRequest::from_flag(port), + serve_port::PortRequest::Explicit(8000) + ); + } + other => panic!("expected Serve, got {other:?}"), + } + } + + #[test] + fn serve_rejects_a_zero_or_out_of_range_port() { + for bad in ["0", "65536", "-1", "auto"] { + let error = parse_serve(&["--port", bad]).expect_err("must be rejected"); + assert!( + error.to_string().contains("between 1 and 65535"), + "unexpected error for --port {bad}: {error}" + ); + } + } + + #[test] + fn hidden_engine_serve_command_keeps_a_concrete_default_port() { + // Only the user-facing verb gained automatic selection; the direct engine + // entry point still requires (and defaults to) a concrete port. + let cli = Cli::try_parse_from([ + "rocm", + "__engine-serve-http", + "lemonade", + "svc-1", + "qwen", + "--state-path", + "state.json", + ]) + .expect("hidden engine command parses"); + match cli.command { + Some(Command::EngineServeHttp { port, host, .. }) => { + assert_eq!(port, rocm_core::DEFAULT_LOCAL_PORT); + assert_eq!(host, DEFAULT_LOCAL_HOST); + } + other => panic!("expected EngineServeHttp, got {other:?}"), + } + } + + #[test] + fn serve_port_help_documents_omission_and_the_custom_host_rule() { + let help = serve_arg_help("port"); + assert!(help.contains("11435"), "{help}"); + assert!(help.contains("automatically"), "{help}"); + assert!(help.contains("127.0.0.1"), "{help}"); + } + #[test] fn serve_verbose_conflicts_with_managed() { // `--verbose` streams logs in the foreground; a backgrounded managed @@ -22224,7 +22519,9 @@ install therock"; live.created_at_unix_ms = 2000; live.write()?; - let found = existing_live_managed_service(&paths, "lemonade", "qwen-canonical"); + let live_services = live_managed_services(&paths); + let found = + existing_live_managed_service_in(&live_services, "lemonade", "qwen-canonical").cloned(); let _ = fs::remove_dir_all(root); let found = found.expect("a live managed service should be detected by engine+model"); @@ -22260,7 +22557,8 @@ install therock"; record.engine_pid = Some(999_999_999); record.write()?; - let found = existing_live_managed_service(&paths, "lemonade", "qwen-canonical"); + let live = live_managed_services(&paths); + let found = existing_live_managed_service_in(&live, "lemonade", "qwen-canonical"); let _ = fs::remove_dir_all(root); assert!( @@ -22294,7 +22592,8 @@ install therock"; record.engine_pid = Some(std::process::id()); record.write()?; - let found = existing_live_managed_service(&paths, "lemonade", "qwen-canonical"); + let live = live_managed_services(&paths); + let found = existing_live_managed_service_in(&live, "lemonade", "qwen-canonical"); let _ = fs::remove_dir_all(root); assert!( @@ -22308,11 +22607,393 @@ install therock"; fn missing_manifest_allows_launch() { // No services dir / manifests → nothing to detect, launch proceeds. let (root, paths) = test_paths("dup-managed-missing"); - let found = existing_live_managed_service(&paths, "lemonade", "qwen-canonical"); + let live = live_managed_services(&paths); + let found = existing_live_managed_service_in(&live, "lemonade", "qwen-canonical"); let _ = fs::remove_dir_all(root); assert!(found.is_none()); } + /// A live managed record: own PID + a live status, so liveness refresh keeps + /// it (and therefore its port reservation) in place. + fn write_live_service_record( + paths: &AppPaths, + service_id: &str, + canonical_model_id: &str, + port: u16, + status: &str, + ) -> Result { + let mut record = ManagedServiceRecord::new( + paths, + service_id, + "lemonade", + canonical_model_id, + canonical_model_id.to_owned(), + DEFAULT_LOCAL_HOST, + port, + "managed", + std::process::id(), + None, + None, + None, + ); + record.status = status.to_owned(); + record.engine_pid = Some(std::process::id()); + record.write()?; + Ok(record) + } + + fn never_probe(_port: u16) -> std::io::Result { + panic!("no port may be probed on this path") + } + + fn always_free(port: u16) -> std::io::Result { + Ok(rocm_core::LoopbackPortLease::engine_owned(port)) + } + + #[test] + fn allocation_transaction_answers_an_equivalent_live_service_before_leasing() -> Result<()> { + let (root, paths) = test_paths("alloc-duplicate-first"); + paths.ensure()?; + write_live_service_record( + &paths, + "lemonade-qwen-1", + "qwen-canonical", + 11_501, + "starting", + )?; + + let decision = resolve_managed_port_in_transaction( + &paths, + "lemonade", + "qwen-canonical", + DEFAULT_LOCAL_HOST, + serve_port::PortRequest::Auto, + &mut never_probe, + ); + let _ = fs::remove_dir_all(root); + + match decision? { + ManagedPortDecision::AlreadyLive(existing) => { + assert_eq!(existing.service_id, "lemonade-qwen-1"); + assert_eq!(existing.port, 11_501); + } + ManagedPortDecision::Lease(lease) => { + panic!("expected the duplicate guard, leased {}", lease.port()) + } + } + Ok(()) + } + + #[test] + fn allocation_transaction_skips_ports_reserved_by_other_live_services() -> Result<()> { + let (root, paths) = test_paths("alloc-skip-reserved"); + paths.ensure()?; + write_live_service_record(&paths, "lemonade-a-1", "model-a", 11_435, "running")?; + write_live_service_record(&paths, "lemonade-b-1", "model-b", 11_436, "recovering")?; + + let decision = resolve_managed_port_in_transaction( + &paths, + "lemonade", + "model-c", + DEFAULT_LOCAL_HOST, + serve_port::PortRequest::Auto, + &mut always_free, + ); + let _ = fs::remove_dir_all(root); + + match decision? { + // A `recovering` record owns its port just as firmly as a running + // one: CLI automatic selection must not race daemon recovery for it. + ManagedPortDecision::Lease(lease) => assert_eq!(lease.port(), 11_437), + ManagedPortDecision::AlreadyLive(existing) => { + panic!("unexpected duplicate match {}", existing.service_id) + } + } + Ok(()) + } + + #[test] + fn allocation_transaction_ignores_stopped_records() -> Result<()> { + let (root, paths) = test_paths("alloc-ignore-stopped"); + paths.ensure()?; + let mut stale = ManagedServiceRecord::new( + &paths, + "lemonade-stale-1", + "lemonade", + "model-a", + "model-a".to_owned(), + DEFAULT_LOCAL_HOST, + 11_435, + "managed", + 999_999_999, + None, + None, + None, + ); + stale.status = "ready".to_owned(); + stale.engine_pid = Some(999_999_999); + stale.write()?; + + let decision = resolve_managed_port_in_transaction( + &paths, + "lemonade", + "model-a", + DEFAULT_LOCAL_HOST, + serve_port::PortRequest::Auto, + &mut always_free, + ); + let _ = fs::remove_dir_all(root); + + match decision? { + // Dead PIDs demote to "stopped", so neither the duplicate guard nor + // the reservation applies: the first candidate is free again. + ManagedPortDecision::Lease(lease) => assert_eq!(lease.port(), 11_435), + ManagedPortDecision::AlreadyLive(existing) => { + panic!( + "a stopped record must not block relaunch: {}", + existing.service_id + ) + } + } + Ok(()) + } + + #[test] + fn allocation_transaction_rejects_an_explicit_port_a_recovering_service_holds() -> Result<()> { + let (root, paths) = test_paths("alloc-explicit-recovering"); + paths.ensure()?; + write_live_service_record(&paths, "lemonade-rec-1", "model-a", 11_435, "recovering")?; + + let decision = resolve_managed_port_in_transaction( + &paths, + "lemonade", + "model-b", + DEFAULT_LOCAL_HOST, + serve_port::PortRequest::Explicit(11_435), + &mut never_probe, + ); + let _ = fs::remove_dir_all(root); + + let error = decision.expect_err("a recovering service still owns its port"); + let message = format!("{error:#}"); + assert!(message.contains("lemonade-rec-1"), "{message}"); + assert!(message.contains("recovering"), "{message}"); + Ok(()) + } + + #[test] + fn concurrent_auto_transactions_publish_distinct_ports() -> Result<()> { + use std::sync::{Arc, Barrier}; + + let (root, paths) = test_paths("alloc-concurrent-distinct"); + paths.ensure()?; + fs::create_dir_all(paths.services_dir())?; + let barrier = Arc::new(Barrier::new(2)); + + let mut handles = Vec::new(); + for model in ["model-a", "model-b"] { + let paths = paths.clone(); + let barrier = Arc::clone(&barrier); + handles.push(thread::spawn(move || -> Result { + barrier.wait(); + // Exactly the production transaction: one lock around decision + // and publication, real sockets for the preflight. + let _allocation = rocm_core::lock_service_allocation(&paths)?; + let lease = match resolve_managed_port_in_transaction( + &paths, + "lemonade", + model, + DEFAULT_LOCAL_HOST, + serve_port::PortRequest::Auto, + &mut serve_port::loopback_bind_probe, + )? { + ManagedPortDecision::Lease(lease) => lease, + ManagedPortDecision::AlreadyLive(existing) => { + anyhow::bail!("unexpected duplicate {}", existing.service_id) + } + }; + let port = + write_live_service_record(&paths, model, model, lease.port(), "starting")?.port; + lease.release(); + Ok(port) + })); + } + + let ports: Vec = handles + .into_iter() + .map(|handle| handle.join().expect("transaction thread")) + .collect::>>()?; + let _ = fs::remove_dir_all(root); + + assert_eq!(ports.len(), 2); + assert_ne!( + ports[0], ports[1], + "two concurrent automatic launches must not publish the same port" + ); + for port in ports { + assert!((serve_port::AUTO_PORT_FIRST..=serve_port::AUTO_PORT_LAST).contains(&port)); + } + Ok(()) + } + + #[test] + fn concurrent_equivalent_transactions_converge_on_one_record() -> Result<()> { + use std::sync::{Arc, Barrier}; + + let (root, paths) = test_paths("alloc-concurrent-equivalent"); + paths.ensure()?; + fs::create_dir_all(paths.services_dir())?; + let barrier = Arc::new(Barrier::new(2)); + + let mut handles = Vec::new(); + for id in ["lemonade-dup-1", "lemonade-dup-2"] { + let paths = paths.clone(); + let barrier = Arc::clone(&barrier); + handles.push(thread::spawn(move || -> Result { + barrier.wait(); + let _allocation = rocm_core::lock_service_allocation(&paths)?; + match resolve_managed_port_in_transaction( + &paths, + "lemonade", + "shared-model", + DEFAULT_LOCAL_HOST, + serve_port::PortRequest::Auto, + &mut serve_port::loopback_bind_probe, + )? { + ManagedPortDecision::Lease(lease) => { + write_live_service_record( + &paths, + id, + "shared-model", + lease.port(), + "starting", + )?; + lease.release(); + Ok(true) + } + ManagedPortDecision::AlreadyLive(_) => Ok(false), + } + })); + } + + let launched: Vec = handles + .into_iter() + .map(|handle| handle.join().expect("transaction thread")) + .collect::>>()?; + let published = live_managed_services(&paths).len(); + let _ = fs::remove_dir_all(root); + + assert_eq!( + launched.iter().filter(|launched| **launched).count(), + 1, + "exactly one of two equivalent requests may launch" + ); + assert_eq!(published, 1, "the loser must not publish a second record"); + Ok(()) + } + + #[test] + fn abandoned_launch_leaves_no_record_log_or_endpoint_key() -> Result<()> { + let (root, paths) = test_paths("alloc-abandon-artifacts"); + paths.ensure()?; + let record = + write_live_service_record(&paths, "lemonade-fail-1", "model-a", 11_435, "starting")?; + fs::create_dir_all(record.log_path.parent().expect("log parent"))?; + fs::write(&record.log_path, b"engine log")?; + if let Some(parent) = record.engine_state_path.parent() { + fs::create_dir_all(parent)?; + } + fs::write(&record.engine_state_path, b"{}")?; + endpoint_keys::store_endpoint_api_key(&paths, &record.service_id, "secret-key")?; + + abandon_failed_managed_launch(&paths, &record, None); + + let manifest_exists = record.manifest_path.exists(); + let log_exists = record.log_path.exists(); + let state_exists = record.engine_state_path.exists(); + let key = endpoint_keys::endpoint_api_key(&paths, &record.service_id); + // The reservation must be gone too: nothing may still claim the port. + let live = live_managed_services(&paths); + let _ = fs::remove_dir_all(root); + + assert!(!manifest_exists, "a failed attempt must not leave a record"); + assert!(!log_exists, "a failed attempt must not leave a log"); + assert!( + !state_exists, + "a failed attempt must not leave engine state" + ); + assert_eq!(key, None, "a failed attempt must not leave an endpoint key"); + assert!( + live.is_empty(), + "a failed attempt must leave no reservation" + ); + Ok(()) + } + + #[test] + #[ignore = "spawned as a child process by abandoned_launch_terminates_the_spawned_child"] + fn managed_launch_cleanup_sleeper_child() { + if std::env::var_os("ROCM_TEST_SLEEPER_CHILD").is_none() { + return; + } + thread::sleep(Duration::from_mins(2)); + } + + #[test] + fn abandoned_launch_terminates_the_spawned_child() -> Result<()> { + let (root, paths) = test_paths("alloc-abandon-child"); + paths.ensure()?; + let record = + write_live_service_record(&paths, "lemonade-fail-2", "model-a", 11_435, "starting")?; + + // A real, long-lived child stands in for an engine process that was + // spawned but whose record could not be published. + let mut child = ProcessCommand::new(std::env::current_exe().expect("test binary")) + .arg("tests::managed_launch_cleanup_sleeper_child") + .arg("--exact") + .arg("--ignored") + .env("ROCM_TEST_SLEEPER_CHILD", "1") + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .expect("spawn sleeper child"); + let child_pid = child.id(); + + abandon_failed_managed_launch(&paths, &record, Some(child_pid)); + + // On Unix a killed direct child stays a zombie — and therefore "running" + // to a `kill(pid, 0)` probe — until its parent reaps it, so prove death + // by reaping, then confirm the pid is gone. + let deadline = std::time::Instant::now() + Duration::from_secs(10); + let mut exited = None; + while exited.is_none() && std::time::Instant::now() < deadline { + exited = child.try_wait().expect("query sleeper child"); + if exited.is_none() { + thread::sleep(Duration::from_millis(50)); + } + } + let terminated = exited.is_some(); + if !terminated { + // Cleanup did not kill it: do so here rather than leaking a sleeper + // for two minutes, and reap it so no zombie outlives the test. + let _ = child.kill(); + } + // Reaped on every path, including the deadline path above. + let _ = child.wait(); + let manifest_exists = record.manifest_path.exists(); + let still_running = !terminated || process_is_running(child_pid); + let _ = fs::remove_dir_all(root); + + assert!( + terminated, + "a child that could not be published must be terminated" + ); + assert!(!still_running, "the child pid must be gone after cleanup"); + assert!(!manifest_exists, "its record must be removed as well"); + Ok(()) + } + #[test] fn spawn_managed_engine_child_blocks_reuse_with_mismatched_recipe() -> Result<()> { // A live service recorded with one recipe (e.g. a tool-call parser flag) @@ -22375,7 +23056,7 @@ install therock"; "qwen", &resolve, "127.0.0.1", - 11511, + serve_port::PortRequest::Explicit(11511), &resolve.device_policy, &[], None, diff --git a/apps/rocm/src/serve_port.rs b/apps/rocm/src/serve_port.rs new file mode 100644 index 00000000..baefa089 --- /dev/null +++ b/apps/rocm/src/serve_port.rs @@ -0,0 +1,475 @@ +// Copyright © Advanced Micro Devices, Inc., or its affiliates. +// +// SPDX-License-Identifier: MIT + +//! Central serve-port policy for the user-facing `rocm serve` verb. +//! +//! `--port` is an explicit-or-automatic *request*, not a concrete default. This +//! module turns that request into one concrete, leased port before any manifest, +//! log, endpoint key, or child process exists: +//! +//! - **Automatic** (`--port` omitted) is accepted only on the canonical default +//! loopback host `127.0.0.1`, and scans [`AUTO_PORT_FIRST`]..=[`AUTO_PORT_LAST`] +//! for a candidate that is neither reserved by a live managed service nor +//! occupied on the OS. +//! - **Explicit on canonical loopback** goes through the same reservation and +//! OS preflight, so a collision fails immediately with a clear message instead +//! of surfacing as an engine bind error minutes into startup. +//! - **Explicit on a custom host** (`localhost`, `::1`, `0.0.0.0`, any other +//! IPv4/IPv6 address) is passed through untouched: bind success or failure +//! stays engine-owned, because an IPv4-loopback probe proves nothing about +//! those addresses. Automatic is refused there rather than guessing. +//! +//! Everything downstream of this module — engine requests, service records, +//! readiness, supervision, recovery — keeps its concrete `u16` contract. + +use anyhow::{Result, bail}; +use rocm_core::{DEFAULT_LOCAL_HOST, DEFAULT_LOCAL_PORT, LoopbackPortLease}; +use std::io; + +/// First automatic candidate. Also the legacy well-known local endpoint, so an +/// otherwise-idle machine keeps serving on the port users already know. +pub(crate) const AUTO_PORT_FIRST: u16 = DEFAULT_LOCAL_PORT; +/// Number of candidates *after* the first one. The scanned range is inclusive on +/// both ends: `11435..=11535`, 101 candidates. +pub(crate) const AUTO_PORT_SPAN: u16 = 100; +/// Last automatic candidate. +pub(crate) const AUTO_PORT_LAST: u16 = AUTO_PORT_FIRST + AUTO_PORT_SPAN; + +/// The one phrase used wherever a port must be described *before* it is +/// resolved. The CLI plan output and the Dash approval card share it verbatim so +/// neither promises a concrete port it has not yet acquired. +pub(crate) const AUTOMATIC_PORT_DISCLOSURE: &str = "Port: automatic; endpoint shown after launch"; + +/// What the user asked for on the command line. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) enum PortRequest { + /// `--port` omitted: choose an available local port. + Auto, + /// `--port N`: use exactly this port. + Explicit(u16), +} + +impl PortRequest { + #[must_use] + pub(crate) const fn from_flag(port: Option) -> Self { + match port { + Some(port) => Self::Explicit(port), + None => Self::Auto, + } + } + + #[must_use] + pub(crate) const fn is_auto(self) -> bool { + matches!(self, Self::Auto) + } + + /// The concrete port, when the user named one. + #[must_use] + pub(crate) const fn explicit(self) -> Option { + match self { + Self::Auto => None, + Self::Explicit(port) => Some(port), + } + } +} + +/// A port a live managed service still owns. Carried with its service id so a +/// collision names the thing actually holding the port. +#[derive(Clone, Debug, PartialEq, Eq)] +pub(crate) struct ReservedPort { + pub(crate) port: u16, + pub(crate) service_id: String, + pub(crate) status: String, +} + +/// clap value parser for every user-facing `--port`. +/// +/// Rejects `0` explicitly: the OS reads it as "any ephemeral port", which would +/// silently produce an endpoint nobody asked for. +/// +/// # Errors +/// Returns a user-facing message for anything outside 1–65535. +pub(crate) fn parse_serve_port(value: &str) -> Result { + match value.trim().parse::() { + Ok(0) | Err(_) => Err(format!( + "port must be a number between 1 and 65535, got '{value}'" + )), + Ok(port) => Ok(port), + } +} + +/// Reject an automatic request on a host this policy cannot preflight, before +/// any engine resolution or side effect happens. +/// +/// # Errors +/// Fails when `--port` was omitted for anything but the canonical loopback host. +pub(crate) fn validate_port_request(host: &str, request: PortRequest) -> Result<()> { + if request.is_auto() && !rocm_core::is_canonical_loopback_host(host) { + bail!( + "`rocm serve --host {host}` needs an explicit `--port <1-65535>`. Automatic port \ + selection is supported only on the default host {DEFAULT_LOCAL_HOST}, where ROCm \ + can prove a port is free before launching; on any other address the engine owns \ + the bind." + ); + } + Ok(()) +} + +/// Production bind probe: take a real `127.0.0.1:` lease. +/// +/// # Errors +/// Propagates the OS bind error verbatim so the caller can distinguish +/// "occupied" from "not permitted"/"not available". +pub(crate) fn loopback_bind_probe(port: u16) -> io::Result { + rocm_core::lease_loopback_port(port) +} + +/// Turn a [`PortRequest`] into one concrete, leased port. +/// +/// Callers must hold the shared managed-service allocation lock across this call +/// *and* the record publication that follows it, so a concurrent launch cannot +/// pick the same candidate between the probe and the spawn. +/// +/// `probe` is injected so tests can drive deterministic OS outcomes; production +/// passes [`loopback_bind_probe`]. +/// +/// # Errors +/// Fails when an explicit port is reserved or occupied, when the automatic range +/// is exhausted, or when the OS reports anything other than "address in use" — +/// a permission or address-availability error must never be papered over by +/// quietly choosing a different port. +pub(crate) fn resolve_serve_port( + host: &str, + request: PortRequest, + reserved: &[ReservedPort], + probe: &mut dyn FnMut(u16) -> io::Result, +) -> Result { + if !rocm_core::is_canonical_loopback_host(host) { + // Custom host: an explicit port passes through untouched so bind success + // or failure stays engine-owned. Automatic is refused here as well as in + // `serve()`, because this is the shared choke point every caller reaches + // — and it is refused by returning an error, never by panicking. + let Some(port) = request.explicit() else { + validate_port_request(host, request)?; + unreachable!("custom-host automatic requests are rejected above"); + }; + return Ok(LoopbackPortLease::engine_owned(port)); + } + + match request { + PortRequest::Explicit(port) => { + if let Some(owner) = reserved.iter().find(|entry| entry.port == port) { + bail!( + "{DEFAULT_LOCAL_HOST}:{port} is already reserved by managed service {} \ + ({}). Stop it with `rocm services stop {} --yes`, or pass a different \ + `--port`.", + owner.service_id, + owner.status, + owner.service_id + ); + } + probe(port).map_err(|error| explicit_bind_error(port, &error)) + } + PortRequest::Auto => { + for port in AUTO_PORT_FIRST..=AUTO_PORT_LAST { + if reserved.iter().any(|entry| entry.port == port) { + continue; + } + match probe(port) { + Ok(lease) => return Ok(lease), + // Occupied is the only outcome that advances the scan. + Err(error) if error.kind() == io::ErrorKind::AddrInUse => {} + Err(error) => bail!( + "failed to reserve {DEFAULT_LOCAL_HOST}:{port} while choosing an \ + automatic port: {error}" + ), + } + } + bail!( + "no free TCP port for {DEFAULT_LOCAL_HOST} in {AUTO_PORT_FIRST}–{AUTO_PORT_LAST}; \ + every candidate is reserved by a live managed service or already in use. Stop an \ + unused server with `rocm services stop --yes`, or pass an explicit \ + `--port`." + ); + } + } +} + +fn explicit_bind_error(port: u16, error: &io::Error) -> anyhow::Error { + if error.kind() == io::ErrorKind::AddrInUse { + return anyhow::anyhow!( + "{DEFAULT_LOCAL_HOST}:{port} is already in use by another process. Choose a free \ + `--port`, stop the process holding it, or omit `--port` to let ROCm pick an \ + available port starting at {AUTO_PORT_FIRST}." + ); + } + anyhow::anyhow!("failed to reserve {DEFAULT_LOCAL_HOST}:{port}: {error}") +} + +#[cfg(test)] +mod tests { + use super::*; + use std::net::TcpListener; + + fn reserved(port: u16, service_id: &str, status: &str) -> ReservedPort { + ReservedPort { + port, + service_id: service_id.to_owned(), + status: status.to_owned(), + } + } + + /// A probe driven by a table of scripted per-port failures. Any port absent + /// from the table binds successfully. + fn scripted( + outcomes: Vec<(u16, io::ErrorKind)>, + ) -> impl FnMut(u16) -> io::Result { + move |port| match outcomes.iter().find(|(candidate, _)| *candidate == port) { + Some((_, kind)) => Err(io::Error::new(*kind, "scripted")), + None => Ok(LoopbackPortLease::engine_owned(port)), + } + } + + #[test] + fn port_request_distinguishes_omitted_from_explicit() { + assert_eq!(PortRequest::from_flag(None), PortRequest::Auto); + assert!(PortRequest::from_flag(None).is_auto()); + assert_eq!(PortRequest::from_flag(None).explicit(), None); + assert_eq!( + PortRequest::from_flag(Some(8000)), + PortRequest::Explicit(8000) + ); + assert!(!PortRequest::from_flag(Some(8000)).is_auto()); + assert_eq!(PortRequest::from_flag(Some(8000)).explicit(), Some(8000)); + } + + #[test] + fn port_parser_rejects_zero_and_out_of_range_values() { + assert_eq!(parse_serve_port("8000"), Ok(8000)); + assert_eq!(parse_serve_port(" 11435 "), Ok(11_435)); + assert_eq!(parse_serve_port("65535"), Ok(65_535)); + for bad in ["0", "65536", "-1", "eleven", ""] { + let error = parse_serve_port(bad).expect_err("must be rejected"); + assert!( + error.contains("between 1 and 65535"), + "unexpected message for {bad:?}: {error}" + ); + } + } + + #[test] + fn automatic_range_is_exactly_the_documented_window() { + assert_eq!(AUTO_PORT_FIRST, 11_435); + assert_eq!(AUTO_PORT_LAST, 11_535); + assert_eq!((AUTO_PORT_FIRST..=AUTO_PORT_LAST).count(), 101); + } + + #[test] + fn automatic_is_refused_on_every_custom_host() { + validate_port_request(DEFAULT_LOCAL_HOST, PortRequest::Auto).expect("canonical host is ok"); + for host in ["localhost", "::1", "0.0.0.0", "127.0.0.2", "10.0.0.5"] { + let error = validate_port_request(host, PortRequest::Auto) + .expect_err("automatic must be refused on a custom host"); + let message = format!("{error:#}"); + assert!(message.contains("explicit `--port"), "{message}"); + assert!(message.contains(host), "{message}"); + // An explicit port is always accepted there. + validate_port_request(host, PortRequest::Explicit(8000)) + .expect("explicit port is valid on a custom host"); + } + } + + #[test] + fn custom_host_explicit_port_bypasses_the_loopback_preflight() { + let mut probe = |_port: u16| -> io::Result { + panic!("custom hosts must not be preflighted against IPv4 loopback") + }; + // Even a port a live loopback service reserves is passed straight + // through: the engine owns bind on that address. + let lease = resolve_serve_port( + "0.0.0.0", + PortRequest::Explicit(11_435), + &[reserved(11_435, "svc-a", "running")], + &mut probe, + ) + .expect("custom host explicit port is engine-owned"); + assert_eq!(lease.port(), 11_435); + assert!(!lease.is_held()); + } + + #[test] + fn resolving_auto_on_a_custom_host_errors_instead_of_panicking() { + // The choke point must refuse Auto on its own, even when a caller skipped + // the earlier `validate_port_request` gate — as a Result, never a panic. + let mut probe = |_port: u16| -> io::Result { + panic!("a custom host must never be preflighted") + }; + for host in ["localhost", "::1", "0.0.0.0", "10.0.0.5"] { + let error = resolve_serve_port(host, PortRequest::Auto, &[], &mut probe) + .expect_err("automatic selection is not supported on a custom host"); + let message = format!("{error:#}"); + assert!(message.contains("explicit `--port"), "{message}"); + assert!(message.contains(host), "{message}"); + } + } + + #[test] + fn explicit_loopback_port_outside_the_auto_range_is_retained() { + let mut probe = scripted(vec![]); + let lease = resolve_serve_port( + DEFAULT_LOCAL_HOST, + PortRequest::Explicit(8000), + &[], + &mut probe, + ) + .expect("a free explicit port is kept"); + assert_eq!(lease.port(), 8000); + } + + #[test] + fn explicit_loopback_port_that_is_occupied_is_rejected() { + let mut probe = scripted(vec![(8000, io::ErrorKind::AddrInUse)]); + let error = resolve_serve_port( + DEFAULT_LOCAL_HOST, + PortRequest::Explicit(8000), + &[], + &mut probe, + ) + .expect_err("an occupied explicit port must fail"); + let message = format!("{error:#}"); + assert!(message.contains("127.0.0.1:8000"), "{message}"); + assert!(message.contains("already in use"), "{message}"); + } + + #[test] + fn explicit_loopback_port_reserved_by_a_live_service_names_that_service() { + let mut probe = |_port: u16| -> io::Result { + panic!("a reserved port must be rejected before probing") + }; + let error = resolve_serve_port( + DEFAULT_LOCAL_HOST, + PortRequest::Explicit(11_435), + &[reserved(11_435, "vllm-qwen-1", "starting")], + &mut probe, + ) + .expect_err("a reserved explicit port must fail"); + let message = format!("{error:#}"); + assert!(message.contains("vllm-qwen-1"), "{message}"); + assert!(message.contains("starting"), "{message}"); + } + + #[test] + fn automatic_takes_the_first_candidate_when_it_is_free() { + let mut probe = scripted(vec![]); + let lease = resolve_serve_port(DEFAULT_LOCAL_HOST, PortRequest::Auto, &[], &mut probe) + .expect("first candidate is free"); + assert_eq!(lease.port(), AUTO_PORT_FIRST); + } + + #[test] + fn automatic_skips_occupied_candidates_until_one_binds() { + let mut probe = scripted(vec![ + (11_435, io::ErrorKind::AddrInUse), + (11_436, io::ErrorKind::AddrInUse), + (11_437, io::ErrorKind::AddrInUse), + ]); + let lease = resolve_serve_port(DEFAULT_LOCAL_HOST, PortRequest::Auto, &[], &mut probe) + .expect("scan advances past occupied candidates"); + assert_eq!(lease.port(), 11_438); + } + + #[test] + fn automatic_skips_ports_reserved_by_live_records_without_probing_them() { + let mut probed: Vec = Vec::new(); + let lease = { + let mut probe = |port: u16| -> io::Result { + probed.push(port); + Ok(LoopbackPortLease::engine_owned(port)) + }; + resolve_serve_port( + DEFAULT_LOCAL_HOST, + PortRequest::Auto, + &[ + reserved(11_435, "svc-ready", "ready"), + reserved(11_436, "svc-recovering", "recovering"), + ], + &mut probe, + ) + .expect("scan skips reserved candidates") + }; + assert_eq!(lease.port(), 11_437); + assert_eq!(probed, vec![11_437]); + } + + #[test] + fn automatic_stops_immediately_on_a_non_collision_bind_error() { + for kind in [ + io::ErrorKind::PermissionDenied, + io::ErrorKind::AddrNotAvailable, + ] { + let mut probe = scripted(vec![(11_435, kind)]); + let error = resolve_serve_port(DEFAULT_LOCAL_HOST, PortRequest::Auto, &[], &mut probe) + .expect_err("only AddrInUse may advance the scan"); + let message = format!("{error:#}"); + assert!(message.contains("127.0.0.1:11435"), "{message}"); + } + } + + #[test] + fn explicit_non_collision_bind_error_reports_host_port_and_os_detail() { + let mut probe = |_port: u16| -> io::Result { + Err(io::Error::new( + io::ErrorKind::PermissionDenied, + "permission denied", + )) + }; + let error = resolve_serve_port( + DEFAULT_LOCAL_HOST, + PortRequest::Explicit(80), + &[], + &mut probe, + ) + .expect_err("a privileged port must fail"); + let message = format!("{error:#}"); + assert!(message.contains("127.0.0.1:80"), "{message}"); + assert!(message.contains("permission denied"), "{message}"); + } + + #[test] + fn automatic_exhaustion_reports_the_exact_scanned_range() { + let mut probe = |_port: u16| -> io::Result { + Err(io::Error::new(io::ErrorKind::AddrInUse, "occupied")) + }; + let error = resolve_serve_port(DEFAULT_LOCAL_HOST, PortRequest::Auto, &[], &mut probe) + .expect_err("an exhausted range must fail"); + let message = format!("{error:#}"); + assert!(message.contains("11435–11535"), "{message}"); + } + + #[test] + fn automatic_selects_against_real_loopback_listeners() { + // Real sockets, not a script: hold the first two candidates with live + // listeners and prove the production probe walks past them. If either is + // already owned by something outside this test the scenario cannot be set + // up deterministically, so skip rather than assert on a moving target. + let (Ok(first), Ok(second)) = ( + TcpListener::bind(("127.0.0.1", AUTO_PORT_FIRST)), + TcpListener::bind(("127.0.0.1", AUTO_PORT_FIRST + 1)), + ) else { + return; + }; + let mut probe = loopback_bind_probe; + let lease = resolve_serve_port(DEFAULT_LOCAL_HOST, PortRequest::Auto, &[], &mut probe) + .expect("a candidate above the held ones is free"); + assert!( + lease.port() >= AUTO_PORT_FIRST + 2, + "selected {} while {AUTO_PORT_FIRST} and {} were held", + lease.port(), + AUTO_PORT_FIRST + 1 + ); + assert!(lease.is_held(), "the production probe must hold the socket"); + // Guards outlive every assertion above. + drop((first, second)); + } +} diff --git a/apps/rocmd/src/lib.rs b/apps/rocmd/src/lib.rs index 1e638528..c50056db 100644 --- a/apps/rocmd/src/lib.rs +++ b/apps/rocmd/src/lib.rs @@ -4790,7 +4790,25 @@ fn handle_server_recover_event_with_record( Some(record.service_id.clone()), ); } - restart_managed_service(paths, &mut *record)?; + // A contended allocation lock or an occupied recorded port is a + // recoverable, per-service condition — report it and move on rather + // than letting one service take the whole 30s watcher tick (and the + // daemon) down. + if let Err(error) = restart_managed_service(paths, &mut *record) { + return record_event( + paths, + state, + "server-recover", + "error", + "restart_managed_service_failed", + &format!( + "failed to recover managed service {} on {}:{} after \ + {recovery_reason_display}: {error:#}", + record.service_id, record.host, record.port + ), + Some(record.service_id.clone()), + ); + } record_event( paths, state, @@ -4937,7 +4955,33 @@ fn manifest_service_recovery_reason( } } -fn restart_managed_service(_paths: &AppPaths, record: &mut ManagedServiceRecord) -> Result<()> { +/// Respawn the supervisor for a managed service that needs recovery. +/// +/// Recovery re-claims the concrete port already written in the record, so it +/// runs inside the same bounded cross-process allocation lock as a fresh +/// `rocm serve` launch: a CLI automatic selection and a recovery re-bind can +/// never interleave, and the CLI sees this record's `recovering` reservation. +fn restart_managed_service(paths: &AppPaths, record: &mut ManagedServiceRecord) -> Result<()> { + let _allocation = rocm_core::lock_service_allocation(paths)?; + + // On the canonical loopback host, prove the recorded port is genuinely free + // before publishing `recovering` or spawning anything: a supervisor that + // cannot bind would replace a truthful record with a lie. Custom hosts keep + // the existing engine-owned bind behavior — the record transition is still + // serialized, but the bind is not preflighted. + let lease = if rocm_core::is_canonical_loopback_host(&record.host) { + Some( + rocm_core::lease_loopback_port(record.port).with_context(|| { + format!( + "cannot recover managed service {} on {}:{}: its recorded port is in use", + record.service_id, record.host, record.port + ) + })?, + ) + } else { + None + }; + let rocmd_binary = std::env::current_exe().context("failed to resolve current rocmd executable path")?; let log_file = fs::OpenOptions::new() @@ -4949,6 +4993,11 @@ fn restart_managed_service(_paths: &AppPaths, record: &mut ManagedServiceRecord) .try_clone() .context("failed to clone service log file handle")?; + // The truthful state to fall back to. Any failure below must leave the + // operator with this, not a half-applied `recovering` row — and, once the + // supervisor exists, must not leave that supervisor running unrecorded. + let prior = record.clone(); + record.status = "recovering".to_owned(); // Counts the restart and drops the previous run's inference verification. // The respawned child writes a fresh record of its own, and "recovering" is @@ -4958,7 +5007,18 @@ fn restart_managed_service(_paths: &AppPaths, record: &mut ManagedServiceRecord) // true at the one site that reuses a record across restarts. record.reset_for_restart(); record.supervisor_pid = std::process::id(); - record.write()?; + if let Err(error) = record.write() { + // Nothing was spawned and the on-disk record was never replaced; just + // undo the in-memory transition so the caller does not report recovery + // that never happened. + *record = prior; + return Err(error.context("failed to publish the recovering service record")); + } + // Hand the port to the supervisor: close the probe socket immediately before + // the spawn, still holding the allocation lock. + if let Some(lease) = lease { + lease.release(); + } let mut child = detached_rocmd_command(&rocmd_binary) .args(recovery_supervise_args(record)) @@ -4969,7 +5029,26 @@ fn restart_managed_service(_paths: &AppPaths, record: &mut ManagedServiceRecord) .context("failed to spawn recovery supervisor")?; record.supervisor_pid = child.id(); - record.write()?; + if let Err(error) = record.write() { + // The supervisor is alive but unrecorded: it would hold the port while + // being invisible to every stop/recovery path. Kill it, then put the + // prior truthful record back. + terminate_recovery_supervisor(&mut child); + *record = prior; + let error = error.context("failed to publish the recovering service record"); + if let Err(rollback) = record.write() { + // Both writes failed: the on-disk record now still claims + // `recovering` for a supervisor that no longer exists. That is worse + // than the original failure and must not be swallowed — the watcher + // and the manual restart tool both surface whatever comes back. + return Err(error.context(format!( + "rollback also failed: managed service {} is left recorded as recovering \ + with no supervisor ({rollback:#})", + record.service_id + ))); + } + return Err(error); + } thread::sleep(Duration::from_millis(200)); if let Some(status) = child @@ -4987,6 +5066,26 @@ fn restart_managed_service(_paths: &AppPaths, record: &mut ManagedServiceRecord) Ok(()) } +/// Grace given to a recovery supervisor that must be terminated because its +/// record could not be published. Short: it is milliseconds old. +const RECOVERY_CLEANUP_GRACE: Duration = Duration::from_secs(2); + +/// Terminate and reap a recovery supervisor that could not be recorded. +/// +/// Tree scope, forced: the supervisor may already have started an engine child, +/// and leaving either alive would keep the recorded port occupied by a process +/// no stop or recovery path can find. Reaping afterwards keeps the daemon free +/// of zombies. +fn terminate_recovery_supervisor(child: &mut std::process::Child) { + let _outcome = rocm_core::terminate_verified( + &rocm_core::ProcessIdentity::capture(child.id()), + rocm_core::KillScope::Tree, + RECOVERY_CLEANUP_GRACE, + true, + ); + let _ = child.wait(); +} + fn recovery_supervise_args(record: &ManagedServiceRecord) -> Vec { let mut args = vec![ "supervise".to_owned(), @@ -5562,6 +5661,167 @@ mod tests { ); } + #[test] + fn recovery_waits_for_the_allocation_lock_and_refuses_an_occupied_port() -> Result<()> { + use std::net::TcpListener; + use std::sync::Arc; + use std::sync::atomic::{AtomicBool, Ordering}; + + let (_root, paths) = temp_app_paths("recovery-port-collision"); + paths.ensure()?; + fs::create_dir_all(paths.services_dir())?; + + // Something else already owns the port this service recorded. + let occupant = TcpListener::bind(("127.0.0.1", 0))?; + let port = occupant.local_addr()?.port(); + + let mut record = ManagedServiceRecord::new( + &paths, + "lemonade-recover-1", + "lemonade", + "qwen", + "qwen-canonical".to_owned(), + DEFAULT_LOCAL_HOST, + port, + "managed", + 0, + None, + None, + None, + ); + record.status = "failed".to_owned(); + record.write()?; + + let guard = rocm_core::lock_service_allocation(&paths)?; + let finished = Arc::new(AtomicBool::new(false)); + let handle = { + let paths = paths.clone(); + let finished = Arc::clone(&finished); + let mut record = record.clone(); + thread::spawn(move || { + let result = restart_managed_service(&paths, &mut record); + finished.store(true, Ordering::SeqCst); + result + }) + }; + + thread::sleep(Duration::from_millis(250)); + assert!( + !finished.load(Ordering::SeqCst), + "recovery must serialize on the shared allocation lock" + ); + drop(guard); + + let error = handle + .join() + .expect("recovery thread") + .expect_err("an occupied recorded port must fail recovery"); + let message = format!("{error:#}"); + assert!(message.contains("recorded port is in use"), "{message}"); + + // Nothing was spawned and the prior truthful record is untouched: it must + // not have been rewritten as `recovering`. + let on_disk = load_managed_services(&paths)? + .into_iter() + .find(|candidate| candidate.service_id == "lemonade-recover-1") + .expect("record still present"); + assert_eq!(on_disk.status, "failed"); + assert_eq!(on_disk.restart_count, record.restart_count); + drop(occupant); + Ok(()) + } + + #[test] + fn recovery_publication_failure_leaves_the_prior_record_intact() -> Result<()> { + use std::net::TcpListener; + + let (_root, paths) = temp_app_paths("recovery-publication-failure"); + paths.ensure()?; + fs::create_dir_all(paths.services_dir())?; + + // A free recorded port, so the lease succeeds and the failure under test + // is the record publication itself. + let probe = TcpListener::bind(("127.0.0.1", 0))?; + let port = probe.local_addr()?.port(); + drop(probe); + + let mut record = ManagedServiceRecord::new( + &paths, + "lemonade-recover-2", + "lemonade", + "qwen", + "qwen-canonical".to_owned(), + DEFAULT_LOCAL_HOST, + port, + "managed", + 4242, + None, + None, + None, + ); + record.status = "failed".to_owned(); + record.write()?; + + // Make the manifest unwritable by replacing it with a directory. The + // `recovering` transition can no longer be published. + fs::remove_file(&record.manifest_path)?; + fs::create_dir_all(&record.manifest_path)?; + + let mut attempt = record.clone(); + let error = restart_managed_service(&paths, &mut attempt) + .expect_err("an unpublishable record must fail recovery"); + let message = format!("{error:#}"); + assert!(message.contains("recovering service record"), "{message}"); + // Nothing was spawned and the on-disk row was never replaced, so the + // error must NOT claim a failed rollback: that phrasing is reserved for + // the post-spawn case where the record really is left mid-transition. + assert!(!message.contains("rollback also failed"), "{message}"); + + // The caller's record must not claim a recovery that never happened. + assert_eq!(attempt.status, "failed"); + assert_eq!(attempt.restart_count, record.restart_count); + assert_eq!(attempt.supervisor_pid, record.supervisor_pid); + fs::remove_dir_all(&record.manifest_path)?; + Ok(()) + } + + #[test] + #[ignore = "spawned as a child process by terminate_recovery_supervisor_kills_and_reaps_the_child"] + fn recovery_cleanup_sleeper_child() { + if std::env::var_os("ROCMD_TEST_SLEEPER_CHILD").is_none() { + return; + } + thread::sleep(Duration::from_mins(2)); + } + + #[test] + fn terminate_recovery_supervisor_kills_and_reaps_the_child() { + // A real long-lived process stands in for a supervisor that was spawned + // but whose record could not be published. + let mut child = ProcessCommand::new(std::env::current_exe().expect("test binary")) + .arg("tests::recovery_cleanup_sleeper_child") + .arg("--exact") + .arg("--ignored") + .env("ROCMD_TEST_SLEEPER_CHILD", "1") + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .expect("spawn sleeper child"); + let child_pid = child.id(); + + terminate_recovery_supervisor(&mut child); + + // The helper reaps, so the pid must be gone rather than a zombie. + assert!( + !rocm_core::process_is_running(child_pid), + "an unrecordable supervisor must be terminated and reaped" + ); + // The helper already reaped it; wait again so the child is provably + // handled on every path in this function (and reaping is idempotent). + assert!(child.wait().is_ok()); + } + #[test] fn rocm_mcp_tools_include_bridge_gaps() { let tools = rocm_mcp_tools(); diff --git a/crates/rocm-core/src/lib.rs b/crates/rocm-core/src/lib.rs index 1846e013..41c6d7d2 100644 --- a/crates/rocm-core/src/lib.rs +++ b/crates/rocm-core/src/lib.rs @@ -11,7 +11,7 @@ use std::ffi::OsStr; use std::ffi::OsString; use std::fs; use std::io::{IsTerminal, Read, Write, stdin, stdout}; -use std::net::{IpAddr, TcpStream, ToSocketAddrs}; +use std::net::{IpAddr, TcpListener, TcpStream, ToSocketAddrs}; #[cfg(windows)] use std::os::windows::ffi::OsStrExt; use std::path::{Path, PathBuf}; @@ -7447,6 +7447,189 @@ pub fn unix_time_millis() -> u128 { .as_millis() } +// --------------------------------------------------------------------------- +// Managed-service port leasing and the shared allocation lock. +// +// Both `rocm serve` and `rocmd` recovery contend for the same local TCP ports +// and the same service records, so the primitives that arbitrate them live here +// rather than in either binary. +// --------------------------------------------------------------------------- + +/// True only for the exact canonical default loopback host. +/// +/// Proactive OS collision detection (leasing a real socket before launch) is +/// claimed only for this one host. `localhost`, `::1`, `0.0.0.0`, and other +/// IPv4 addresses are "custom hosts": they require an explicit port and keep the +/// existing engine-owned bind/error behavior, because an IPv4-loopback probe +/// would not prove anything about the address the engine will actually bind. +#[must_use] +pub fn is_canonical_loopback_host(host: &str) -> bool { + host.trim() == DEFAULT_LOCAL_HOST +} + +/// A real, held `127.0.0.1:` listener proving the port is bindable now. +/// +/// The lease is the collision check: it is acquired before any manifest, log, or +/// child process exists, and [`LoopbackPortLease::release`] drops the socket +/// immediately before the engine child is spawned so the child can bind it. The +/// window between release and the child's bind is unavoidable without engine +/// socket-passing; it is far narrower than the previous "bind and hope" launch. +#[derive(Debug)] +pub struct LoopbackPortLease { + port: u16, + listener: Option, +} + +impl LoopbackPortLease { + #[must_use] + pub const fn port(&self) -> u16 { + self.port + } + + /// True while this lease still holds the socket open. + #[must_use] + pub const fn is_held(&self) -> bool { + self.listener.is_some() + } + + /// Close the socket and hand the port to the caller. Call this immediately + /// before spawning the process that will bind it. + pub fn release(mut self) -> u16 { + self.listener = None; + self.port + } + + /// A lease over a port this process never probed, for custom hosts where the + /// engine — not the CLI — owns bind success or failure. + #[must_use] + pub const fn engine_owned(port: u16) -> Self { + Self { + port, + listener: None, + } + } +} + +/// Try to take an exclusive IPv4-loopback lease on `port`. +/// +/// Errors are returned verbatim so callers can distinguish "occupied" +/// ([`std::io::ErrorKind::AddrInUse`], the only condition that should advance an +/// automatic scan) from a privileged or unavailable address, which must fail +/// immediately rather than silently choosing a different port. +pub fn lease_loopback_port(port: u16) -> std::io::Result { + let listener = TcpListener::bind(std::net::SocketAddrV4::new( + std::net::Ipv4Addr::LOCALHOST, + port, + ))?; + Ok(LoopbackPortLease { + port, + listener: Some(listener), + }) +} + +/// File name of the shared allocation lock, inside [`AppPaths::services_dir`]. +/// +/// Not a `.json` file, so `load_managed_services` never mistakes it for a +/// service manifest. +pub const SERVICE_ALLOCATION_LOCK_FILE: &str = "allocation.lock"; +/// Total time a launch or recovery waits for the allocation lock before failing +/// with an actionable error instead of hanging. +pub const SERVICE_ALLOCATION_LOCK_TIMEOUT: Duration = Duration::from_secs(5); +const SERVICE_ALLOCATION_LOCK_POLL: Duration = Duration::from_millis(25); + +#[must_use] +pub fn service_allocation_lock_path(paths: &AppPaths) -> PathBuf { + paths.services_dir().join(SERVICE_ALLOCATION_LOCK_FILE) +} + +/// Cross-process guard serializing the managed-service allocation transaction: +/// duplicate detection, live-reservation refresh, port leasing, child start +/// confirmation, and record publication. +/// +/// The lock is an OS advisory file lock, so it is released both by dropping this +/// guard and by process death — a crashed launcher never wedges the next one. +/// Full HTTP/model readiness deliberately runs *outside* the guard: it takes up +/// to 45 seconds and holds no allocation decision. +#[derive(Debug)] +pub struct ServiceAllocationLock { + file: Option, + path: PathBuf, +} + +impl ServiceAllocationLock { + #[must_use] + pub fn path(&self) -> &Path { + &self.path + } +} + +impl Drop for ServiceAllocationLock { + fn drop(&mut self) { + if let Some(file) = self.file.take() { + let _ = file.unlock(); + } + } +} + +/// Acquire the shared allocation lock for `paths`, waiting at most +/// [`SERVICE_ALLOCATION_LOCK_TIMEOUT`]. +/// +/// # Errors +/// Fails when the lock file cannot be created/locked, or when another process +/// held it for the whole timeout. +pub fn lock_service_allocation(paths: &AppPaths) -> Result { + lock_service_allocation_at( + &service_allocation_lock_path(paths), + SERVICE_ALLOCATION_LOCK_TIMEOUT, + ) +} + +/// [`lock_service_allocation`] against an explicit path and timeout. Exposed for +/// tests that need a private lock file or a short deadline. +/// +/// # Errors +/// See [`lock_service_allocation`]. +pub fn lock_service_allocation_at(path: &Path, timeout: Duration) -> Result { + if let Some(parent) = path.parent() { + fs::create_dir_all(parent) + .with_context(|| format!("failed to create {}", parent.display()))?; + } + let file = fs::OpenOptions::new() + .create(true) + .read(true) + .write(true) + .truncate(false) + .open(path) + .with_context(|| format!("failed to open {}", path.display()))?; + let deadline = Instant::now() + timeout; + loop { + match file.try_lock() { + Ok(()) => { + return Ok(ServiceAllocationLock { + file: Some(file), + path: path.to_path_buf(), + }); + } + Err(std::fs::TryLockError::WouldBlock) => {} + Err(std::fs::TryLockError::Error(error)) => { + return Err( + anyhow::Error::new(error).context(format!("failed to lock {}", path.display())) + ); + } + } + let now = Instant::now(); + if now >= deadline { + bail!( + "timed out after {timeout:?} waiting for the managed-service allocation lock at \ + {path}; another `rocm serve` launch or `rocmd` recovery is mid-transaction. \ + Retry in a moment, and check `rocm services` for a stuck launch.", + path = path.display() + ); + } + thread::sleep(SERVICE_ALLOCATION_LOCK_POLL.min(deadline - now)); + } +} + #[cfg(test)] mod tests { use super::*; @@ -11545,4 +11728,204 @@ last_installed_runtime_id = "therock-release" None ); } + + // -- managed-service allocation lock and loopback leasing --------------- + + const LOCK_CHILD_PATH_ENV: &str = "ROCM_CORE_TEST_LOCK_PATH"; + const LOCK_CHILD_READY_ENV: &str = "ROCM_CORE_TEST_LOCK_READY"; + const HANDOFF_PORT_ENV: &str = "ROCM_CORE_TEST_HANDOFF_PORT"; + + fn lock_test_dir(label: &str) -> PathBuf { + let dir = workspace_test_artifact_dir().join(format!( + "{label}-{}-{}", + std::process::id(), + unix_time_millis() + )); + fs::create_dir_all(&dir).expect("create lock test dir"); + dir + } + + /// Re-run this test binary, executing exactly one `#[ignore]`d helper test in + /// a genuinely separate process. Cross-process locking and socket handoff + /// cannot be proven inside one process. + fn spawn_helper_test(name: &str, envs: &[(&str, String)]) -> std::process::Child { + let exe = std::env::current_exe().expect("test binary path"); + let mut command = Command::new(exe); + command + .arg(name) + .arg("--exact") + .arg("--ignored") + .arg("--test-threads") + .arg("1") + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::null()); + for (key, value) in envs { + command.env(key, value); + } + command.spawn().expect("spawn helper test process") + } + + #[test] + fn canonical_loopback_host_is_only_the_exact_default_address() { + assert!(is_canonical_loopback_host(DEFAULT_LOCAL_HOST)); + assert!(is_canonical_loopback_host(" 127.0.0.1 ")); + // Everything else is a "custom host": explicit port, engine-owned bind. + for host in [ + "localhost", + "::1", + "0.0.0.0", + "127.0.0.2", + "192.168.1.4", + "", + ] { + assert!( + !is_canonical_loopback_host(host), + "{host} must not be treated as the canonical loopback host" + ); + } + } + + #[test] + fn allocation_lock_path_lives_in_services_dir_and_is_not_a_manifest() { + let paths = AppPaths { + config_dir: PathBuf::from("/tmp/cfg"), + data_dir: PathBuf::from("/tmp/data"), + cache_dir: PathBuf::from("/tmp/cache"), + }; + let path = service_allocation_lock_path(&paths); + assert_eq!(path.parent(), Some(paths.services_dir().as_path())); + // `load_managed_services` only reads `*.json`, so the lock must not be one. + assert_ne!( + path.extension().and_then(|value| value.to_str()), + Some("json") + ); + } + + #[test] + fn allocation_lock_excludes_independent_handles_and_fails_closed_on_timeout() { + let path = lock_test_dir("core-alloc-lock-exclusive").join("allocation.lock"); + let held = lock_service_allocation_at(&path, Duration::from_secs(5)).expect("first holder"); + assert_eq!(held.path(), path.as_path()); + + // A second, independent file handle must contend on the OS lock rather + // than sail through, and must give up with an actionable error. + let started = Instant::now(); + let error = lock_service_allocation_at(&path, Duration::from_millis(200)) + .expect_err("a held lock must not be acquired twice"); + assert!( + started.elapsed() >= Duration::from_millis(200), + "waiter returned before its deadline" + ); + let message = format!("{error:#}"); + assert!(message.contains("timed out"), "{message}"); + assert!(message.contains("allocation lock"), "{message}"); + + drop(held); + lock_service_allocation_at(&path, Duration::from_secs(5)) + .expect("lock is free once the guard is dropped"); + } + + #[test] + #[ignore = "spawned as a child process by allocation_lock_is_released_when_the_holder_process_exits"] + fn allocation_lock_child_holder() { + let Ok(path) = std::env::var(LOCK_CHILD_PATH_ENV) else { + return; + }; + let ready = std::env::var(LOCK_CHILD_READY_ENV).expect("ready-file path"); + let guard = lock_service_allocation_at(Path::new(&path), Duration::from_secs(10)) + .expect("child acquires the shared lock"); + fs::write(&ready, "held").expect("signal that the child holds the lock"); + thread::sleep(Duration::from_millis(300)); + // Leak the guard so the lock is released by process exit alone — that is + // the property a crashed launcher depends on. + std::mem::forget(guard); + } + + #[test] + fn allocation_lock_is_released_when_the_holder_process_exits() { + let dir = lock_test_dir("core-alloc-lock-crossproc"); + let path = dir.join("allocation.lock"); + let ready = dir.join("child-ready"); + let mut child = spawn_helper_test( + "tests::allocation_lock_child_holder", + &[ + (LOCK_CHILD_PATH_ENV, path.display().to_string()), + (LOCK_CHILD_READY_ENV, ready.display().to_string()), + ], + ); + + let deadline = Instant::now() + Duration::from_mins(1); + while !ready.exists() { + assert!( + Instant::now() < deadline, + "child never reported holding the lock" + ); + thread::sleep(Duration::from_millis(25)); + } + + let busy = lock_service_allocation_at(&path, Duration::from_millis(50)) + .expect_err("the lock must be exclusive across processes"); + assert!(format!("{busy:#}").contains("allocation lock")); + + let status = child.wait().expect("child holder exits"); + assert!(status.success(), "child holder failed: {status}"); + lock_service_allocation_at(&path, Duration::from_secs(10)) + .expect("the OS releases the lock when the holding process exits"); + } + + #[test] + fn loopback_lease_blocks_a_second_bind_until_released() { + let ephemeral = TcpListener::bind(("127.0.0.1", 0)).expect("ephemeral bind"); + let port = ephemeral.local_addr().expect("local addr").port(); + drop(ephemeral); + + let lease = lease_loopback_port(port).expect("free port leases"); + assert_eq!(lease.port(), port); + assert!(lease.is_held()); + let conflict = lease_loopback_port(port).expect_err("a held port must not lease twice"); + assert_eq!(conflict.kind(), std::io::ErrorKind::AddrInUse); + + assert_eq!(lease.release(), port); + lease_loopback_port(port).expect("the port is bindable again after release"); + } + + #[test] + fn engine_owned_lease_holds_no_socket() { + // Custom hosts get a port carrier, never a loopback probe. + let lease = LoopbackPortLease::engine_owned(8000); + assert_eq!(lease.port(), 8000); + assert!(!lease.is_held()); + } + + #[test] + #[ignore = "spawned as a child process by released_loopback_lease_is_immediately_bindable_by_another_process"] + fn loopback_handoff_child() { + let Ok(port) = std::env::var(HANDOFF_PORT_ENV) else { + return; + }; + let port: u16 = port.parse().expect("handoff port"); + let lease = lease_loopback_port(port).expect("child binds the released port"); + assert_eq!(lease.port(), port); + } + + #[test] + fn released_loopback_lease_is_immediately_bindable_by_another_process() { + let ephemeral = TcpListener::bind(("127.0.0.1", 0)).expect("ephemeral bind"); + let port = ephemeral.local_addr().expect("local addr").port(); + drop(ephemeral); + let lease = lease_loopback_port(port).expect("parent leases the port"); + assert_eq!(lease.release(), port); + + let status = spawn_helper_test( + "tests::loopback_handoff_child", + &[(HANDOFF_PORT_ENV, port.to_string())], + ) + .wait() + .expect("handoff child exits"); + assert!( + status.success(), + "a separate process must bind the port immediately after the lease is released" + ); + } } diff --git a/crates/rocm-dash-tui/src/app/mod.rs b/crates/rocm-dash-tui/src/app/mod.rs index f046c0b8..8a58aaf1 100644 --- a/crates/rocm-dash-tui/src/app/mod.rs +++ b/crates/rocm-dash-tui/src/app/mod.rs @@ -984,7 +984,8 @@ impl AppState { /// (and is ignored while a job runs) before it can eject the manager. fn active_overlay_at_root(&self) -> bool { self.serve_wizard.as_ref().is_none_or(|w| { - w.browser.is_none() + w.editor.is_none() + && w.browser.is_none() && w.picker.is_none() && w.approval.is_none() && w.active_job.is_none() @@ -4331,6 +4332,34 @@ mod tests { ); } + #[test] + fn serve_wizard_editor_handles_escape_before_overlay_back_out() { + let mut state = AppState::new("t".into(), "default-dark".into()); + state.active_tab = ActiveTab::Serving; + let mut wizard = crate::ui::serve_wizard::ServeWizardState { + model: "typed-model".into(), + ..Default::default() + }; + wizard.editor = Some(crate::ui::serve_wizard::InlineEditor { + field: crate::ui::serve_wizard::Field::Host, + original: "127.0.0.1".into(), + value: "localhost".into(), + cursor: 9, + selected: false, + }); + state.serve_wizard = Some(wizard); + + assert!( + !state.should_pane_back_out(KeyCode::Esc), + "Esc must reach the inline editor instead of closing the wizard" + ); + assert_eq!( + state.serve_wizard.as_ref().unwrap().model, + "typed-model", + "the in-progress form remains available to the editor handler" + ); + } + #[test] fn esc_opens_menu_when_idle_but_not_on_chat() { // Idle (non-Chat) tabs: Esc opens the btop main menu. diff --git a/crates/rocm-dash-tui/src/ui/serve_wizard.rs b/crates/rocm-dash-tui/src/ui/serve_wizard.rs index 4e474804..739ab2a3 100644 --- a/crates/rocm-dash-tui/src/ui/serve_wizard.rs +++ b/crates/rocm-dash-tui/src/ui/serve_wizard.rs @@ -5,11 +5,18 @@ //! Serve wizard overlay (Phase 3 Wave 1). //! //! The headline operational screen rebuilt on the Wave-0 primitives: a compact -//! form that builds a `rocm serve … --managed` invocation and runs it **through -//! the approval gate and the job-bridge** — never inline, never with a legacy -//! `std::thread::spawn` + `try_recv`. A served model launched here surfaces in -//! the services manager and the dashboard's live `gen_tps` (the D7 wire is -//! already in place via `rocm serve --managed`). +//! **model-first** form that builds a `rocm serve …` invocation and runs it +//! **through the approval gate and the job-bridge** — never inline, never with +//! a legacy `std::thread::spawn` + `try_recv`. A served model launched here +//! surfaces in the services manager and the dashboard's live `gen_tps`. +//! +//! Progressive disclosure: the default form is Model → `Advanced settings` → +//! Launch. Everything else (engine, device policy, host, port, mode) is +//! automatic and lives behind the inline `Advanced settings` row — expanded in +//! place, never as a second modal. Because the defaults are genuinely +//! automatic, the default invocation is exactly `rocm serve MODEL`: the CLI +//! stays the single resolution authority for engine choice, the GPU-required +//! policy, the loopback host, the port, and managed mode. //! //! The model field can be typed directly (a recipe name, alias, or path) or //! filled from the reusable Wave-0 [`FolderBrowser`] for a local model path @@ -36,28 +43,63 @@ use crate::ui::model_picker::{ModelPicker, ModelRecipeSummary, PickerOutcome, dr use crate::ui::panel::{self, BoxRole}; use crate::ui::theme::Theme; -/// Engine inventory — names mirror `apps/rocm` `engine_inventory()`. Kept -/// TUI-local (a stable, small list) so this layer needs no `rocm-core` dep. -pub const ENGINES: &[&str] = &["lemonade", "vllm"]; - -/// Device-policy choices. Index 0 omits `--device` entirely (engine default); -/// the rest mirror `rocm-core`'s validated `gpu_required|gpu_preferred|cpu_only`. -pub const DEVICES: &[&str] = &[ - "(engine default)", - "gpu_required", - "gpu_preferred", - "cpu_only", -]; +/// Engine inventory — index 0 is the automatic choice. +/// +/// Automatic emits no `--engine` and lets the CLI resolve one; the rest mirror +/// `apps/rocm` `engine_inventory()`. Kept TUI-local (a stable, small list) so +/// this layer needs no `rocm-core` dep. +pub const ENGINES: &[&str] = &["automatic", "lemonade", "vllm"]; + +/// Index into [`ENGINES`] that means "let the CLI decide". +pub const ENGINE_AUTO: usize = 0; + +/// The device policy this wizard offers: read-only, GPU-required, automatic. +/// +/// ROCm never falls back to CPU, so there is nothing here to choose — omitting +/// `--device` already resolves to `gpu_required` in the CLI. +pub const DEVICE_POLICY: &str = "GPU required (automatic)"; /// Mirrors `rocm-core::DEFAULT_LOCAL_HOST` / `DEFAULT_LOCAL_PORT` (TUI-local to -/// avoid the dep; the CLI re-applies its own defaults if these are cleared). +/// avoid the dep). The default host is omitted from argv; the default port text +/// is only the starting point for a *Custom* port. const DEFAULT_HOST: &str = "127.0.0.1"; const DEFAULT_PORT: &str = "11435"; -/// The form fields, in vertical order. +/// The shared phrase for an automatic port. +/// +/// The wizard cannot truthfully promise a concrete port before the CLI leases +/// one, so the approval card says exactly this and the launch output prints the +/// resolved endpoint. +pub const AUTO_PORT_NOTE: &str = "Port: automatic; endpoint shown after launch"; + +/// The exact cross-field guidance for a custom host left on an automatic port. +/// +/// Automatic selection is only supported on the canonical loopback host. +pub const CUSTOM_HOST_NEEDS_PORT: &str = "Custom hosts require a custom port; set Port to Custom."; + +/// Loopback hosts the CLI accepts without `--allow-public-bind`. +/// +/// Dash has no public-bind confirmation or endpoint-key UI, so a public bind +/// must be requested deliberately from the CLI instead. +pub const LOOPBACK_HOSTS: &[&str] = &["127.0.0.1", "localhost", "::1"]; + +/// The guidance shown when a public / non-loopback host is typed into Dash. +pub const PUBLIC_HOST_NEEDS_CLI: &str = "Dash serves on loopback only; for a public bind run \ + `rocm serve --host … --port … --allow-public-bind` from the CLI."; + +/// Whether `host` is one of the exact loopback spellings accepted by the CLI. +#[must_use] +pub fn is_loopback_host(host: &str) -> bool { + LOOPBACK_HOSTS.contains(&host) +} + +/// The form fields. Not every field is visible at once — see +/// [`ServeWizardState::visible_fields`]. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum Field { Model, + /// Inline progressive-disclosure row (`Advanced settings ▸/▾`). + Advanced, Engine, Device, Host, @@ -66,9 +108,10 @@ pub enum Field { Launch, } -/// Field order; `state.field` indexes this. +/// Full (expanded) field order. pub const FIELDS: &[Field] = &[ Field::Model, + Field::Advanced, Field::Engine, Field::Device, Field::Host, @@ -77,13 +120,98 @@ pub const FIELDS: &[Field] = &[ Field::Launch, ]; +/// Collapsed (default) field order — model-first, then disclosure, then launch. +pub const BASIC_FIELDS: &[Field] = &[Field::Model, Field::Advanced, Field::Launch]; + +/// Port policy: automatic (the CLI leases a free local port) or an explicit +/// custom port typed by the user. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub enum PortMode { + #[default] + Auto, + Custom, +} + +/// Inline single-line text editor state for the Host / Custom-port rows. +/// +/// On entry the existing value is *selected*: the first printable character +/// replaces it wholesale (the common "retype it" case) while Left/Right, +/// Backspace, and further typing fall back to ordinary cursor editing. +#[derive(Debug, Clone)] +pub struct InlineEditor { + pub field: Field, + /// Value to restore on Escape. + pub original: String, + pub value: String, + /// Cursor position, in characters, in `0..=value.chars().count()`. + pub cursor: usize, + /// Whether the initial value is still fully selected. + pub selected: bool, +} + +impl InlineEditor { + fn new(field: Field, value: &str) -> Self { + Self { + field, + original: value.to_string(), + value: value.to_string(), + cursor: value.chars().count(), + selected: true, + } + } + + fn len(&self) -> usize { + self.value.chars().count() + } + + fn byte_at(&self, cursor: usize) -> usize { + self.value + .char_indices() + .nth(cursor) + .map_or(self.value.len(), |(b, _)| b) + } + + fn insert(&mut self, c: char) { + if self.selected { + self.value.clear(); + self.cursor = 0; + self.selected = false; + } + let at = self.byte_at(self.cursor); + self.value.insert(at, c); + self.cursor += 1; + } + + fn backspace(&mut self) { + if self.selected { + self.value.clear(); + self.cursor = 0; + self.selected = false; + return; + } + if self.cursor == 0 { + return; + } + let from = self.byte_at(self.cursor - 1); + let to = self.byte_at(self.cursor); + self.value.replace_range(from..to, ""); + self.cursor -= 1; + } + + fn move_cursor(&mut self, delta: isize) { + self.selected = false; + let max = self.len().cast_signed(); + self.cursor = (self.cursor.cast_signed() + delta).clamp(0, max) as usize; + } +} + /// An approved-but-not-yet-launched serve invocation. #[derive(Debug, Clone)] pub struct PendingServe { /// Resolved `rocm` binary path (captured at approval time so a later /// `current_exe()` failure can't silently drop an approved launch). pub cmd: String, - /// The argv after the binary (`["serve", model, "--engine", …]`). + /// The argv after the binary (`["serve", model, …]`). pub args: Vec, pub request: ApprovalRequest, pub choice: ApprovalChoice, @@ -92,13 +220,19 @@ pub struct PendingServe { /// Overlay state. `None` on `AppState` means the wizard is closed. #[derive(Debug, Clone)] pub struct ServeWizardState { + /// Index into [`ServeWizardState::visible_fields`]. pub field: usize, pub model: String, pub engine_idx: usize, - pub device_idx: usize, pub host: String, + pub port_mode: PortMode, + /// The custom port text (only meaningful when `port_mode` is `Custom`). pub port: String, pub managed: bool, + /// Whether the inline `Advanced settings` rows are showing. + pub advanced_expanded: bool, + /// Inline text editor; `Some` while editing Host or the custom port. + pub editor: Option, /// Local-path picker (Wave-0 primitive); `Some` while browsing. pub browser: Option, /// Model-recipe picker sub-step; `Some` while choosing a recipe. @@ -116,11 +250,13 @@ impl Default for ServeWizardState { Self { field: 0, model: String::new(), - engine_idx: 0, - device_idx: 0, + engine_idx: ENGINE_AUTO, host: DEFAULT_HOST.to_string(), + port_mode: PortMode::Auto, port: DEFAULT_PORT.to_string(), managed: true, + advanced_expanded: false, + editor: None, browser: None, picker: None, approval: None, @@ -131,91 +267,173 @@ impl Default for ServeWizardState { } impl ServeWizardState { - fn current_field(&self) -> Field { - FIELDS[self.field.min(FIELDS.len() - 1)] + /// The rows the user can actually see and reach right now. + #[must_use] + pub const fn visible_fields(&self) -> &'static [Field] { + if self.advanced_expanded { + FIELDS + } else { + BASIC_FIELDS + } + } + + /// The focused row (clamped — collapsing can never strand focus). + #[must_use] + pub fn current_field(&self) -> Field { + let vis = self.visible_fields(); + vis[self.field.min(vis.len() - 1)] } fn move_field(&mut self, delta: isize) { - let max = FIELDS.len().cast_signed() - 1; - self.field = (self.field.cast_signed() + delta).clamp(0, max) as usize; + let vis = self.visible_fields(); + let cur = self.field.min(vis.len() - 1); + let max = vis.len().cast_signed() - 1; + self.field = (cur.cast_signed() + delta).clamp(0, max) as usize; + } + + /// Focus the given field, expanding Advanced when it is hidden. + fn focus(&mut self, field: Field) { + if !self.visible_fields().contains(&field) { + self.advanced_expanded = true; + } + if let Some(idx) = self.visible_fields().iter().position(|f| *f == field) { + self.field = idx; + } } fn cycle(&mut self, delta: isize) { match self.current_field() { + // Advanced is a choice row too: ← collapses, → expands. Only + // reachable while Advanced owns focus, and Advanced sits at the + // same index in both orders, so focus is preserved either way. + Field::Advanced => self.advanced_expanded = delta > 0, Field::Engine => self.engine_idx = cycle_idx(self.engine_idx, ENGINES.len(), delta), - Field::Device => self.device_idx = cycle_idx(self.device_idx, DEVICES.len(), delta), + Field::Port => { + self.port_mode = match self.port_mode { + PortMode::Auto => PortMode::Custom, + PortMode::Custom => PortMode::Auto, + }; + } Field::Mode => self.managed = !self.managed, _ => {} } } + /// Direct typing only applies to the Model row; Host and the custom port + /// are edited through the explicit inline editor. fn type_char(&mut self, c: char) { - match self.current_field() { - Field::Model => self.model.push(c), - Field::Host => self.host.push(c), - // Port accepts digits only — never builds an unparseable `--port`. - Field::Port if c.is_ascii_digit() => self.port.push(c), - _ => {} + if self.current_field() == Field::Model { + self.model.push(c); } } fn backspace(&mut self) { - match self.current_field() { - Field::Model => { - self.model.pop(); - } - Field::Host => { - self.host.pop(); - } - Field::Port => { - self.port.pop(); - } - _ => {} + if self.current_field() == Field::Model { + self.model.pop(); } } - /// Build the `rocm` argv for the current form, or an error message. - fn build_args(&self) -> Result, String> { - let model = self.model.trim(); - if model.is_empty() { - return Err("model is required".to_string()); + /// Whether any advanced setting deviates from the automatic defaults. + #[must_use] + pub fn is_customized(&self) -> bool { + self.engine_idx != ENGINE_AUTO + || self.host.trim() != DEFAULT_HOST + || self.port_mode != PortMode::Auto + || !self.managed + } + + /// The `Advanced settings` summary text — the one place a *collapsed* form + /// still tells the truth about overrides, including a broken hidden one. + #[must_use] + pub fn advanced_summary(&self) -> &'static str { + if !self.is_customized() { + return "Automatic"; + } + if self.port_mode == PortMode::Custom && parse_port(&self.port).is_none() { + "Customized · port needs attention" + } else { + "Customized" + } + } + + /// Validate the whole form. On success returns the concrete port to emit + /// (`None` for automatic); on failure the offending field plus a plain fix. + fn validate(&self) -> Result, (Field, String)> { + if self.model.trim().is_empty() { + return Err((Field::Model, "model is required".to_string())); } + let host = self.host.trim(); + if host.is_empty() { + return Err(( + Field::Host, + format!("host is required; type an address or restore {DEFAULT_HOST}"), + )); + } + // Automatic port selection is only supported on the canonical loopback + // host — anything else must name its own port. Never invent one. + if host != DEFAULT_HOST && self.port_mode == PortMode::Auto { + return Err((Field::Port, CUSTOM_HOST_NEEDS_PORT.to_string())); + } + // Dash has no `--allow-public-bind` confirmation and no endpoint-key + // UI, so it must never stage a public bind — even with an explicit + // port. That deliberate step belongs to the CLI. + if !is_loopback_host(host) { + return Err((Field::Host, PUBLIC_HOST_NEEDS_CLI.to_string())); + } + if self.port_mode == PortMode::Custom { + let Some(port) = parse_port(&self.port) else { + let shown = self.port.trim(); + return Err(( + Field::Port, + format!("port `{shown}` is not a valid 1–65535 value"), + )); + }; + return Ok(Some(port)); + } + Ok(None) + } + + /// Build the `rocm` argv for the current form, or the offending field and + /// an error message. Only explicit overrides are emitted: the automatic + /// form is exactly `serve MODEL`. + fn build_args(&self) -> Result, (Field, String)> { + let port = self.validate()?; + let model = self.model.trim(); let mut args = vec!["serve".to_string(), model.to_string()]; - args.push("--engine".to_string()); - args.push(ENGINES[self.engine_idx.min(ENGINES.len() - 1)].to_string()); - // Index 0 = engine default → omit --device. - if self.device_idx > 0 { - args.push("--device".to_string()); - args.push(DEVICES[self.device_idx.min(DEVICES.len() - 1)].to_string()); + // Automatic engine → the CLI resolves it. + if self.engine_idx != ENGINE_AUTO { + args.push("--engine".to_string()); + args.push(ENGINES[self.engine_idx.min(ENGINES.len() - 1)].to_string()); } + // Device policy is automatic and GPU-required; omitting `--device` + // already means `gpu_required` in the CLI. let host = self.host.trim(); - if !host.is_empty() { + if host != DEFAULT_HOST { args.push("--host".to_string()); args.push(host.to_string()); } - let port = self.port.trim(); - if !port.is_empty() { - // u16 accepts 0, but 0 is not a bindable listen port — reject it so - // the error surfaces in the form, not as a downstream bind failure. - match port.parse::() { - Ok(p) if p > 0 => { - args.push("--port".to_string()); - args.push(p.to_string()); - } - _ => return Err(format!("port `{port}` is not a valid 1–65535 value")), - } + if let Some(port) = port { + args.push("--port".to_string()); + args.push(port.to_string()); } - // Managed (default) hands supervision to the daemon → it shows up in the - // services manager + dashboard gen_tps. Foreground runs in this job. - if self.managed { - args.push("--managed".to_string()); - } else { + // Managed (the default) hands supervision to the daemon → it shows up + // in the services manager + dashboard gen_tps, and needs no flag. + if !self.managed { args.push("--foreground".to_string()); } Ok(args) } } +/// Parse a custom port: digits only, 1–65535. `0` parses as `u16` but is not a +/// bindable listen port, so it is rejected here rather than downstream. +fn parse_port(raw: &str) -> Option { + match raw.trim().parse::() { + Ok(p) if p > 0 => Some(p), + _ => None, + } +} + const fn cycle_idx(cur: usize, len: usize, delta: isize) -> usize { if len == 0 { return 0; @@ -242,14 +460,10 @@ pub fn on_key( if let Some(picker) = w.picker.as_mut() { match picker.on_key(key.code, recipes) { PickerOutcome::Chosen(summary) => { + // Only the model is filled. A recipe's preferred engine is NOT + // forced onto the form: the CLI stays the resolution authority + // and the wizard keeps emitting an automatic engine. w.model = summary.id; - // Pre-select the recipe's preferred engine when it is one this - // wizard lists; otherwise leave the engine choice untouched. - if let Some(eng) = summary.preferred_engine - && let Some(idx) = ENGINES.iter().position(|e| *e == eng) - { - w.engine_idx = idx; - } w.picker = None; } PickerOutcome::Cancelled => w.picker = None, @@ -271,7 +485,7 @@ pub fn on_key( return Vec::new(); } - // 2) Approval modal has focus. + // 3) Approval modal has focus. if let Some(pending) = w.approval.as_mut() { let (choice, verdict) = approval_key(key.code, pending.choice); pending.choice = choice; @@ -287,7 +501,7 @@ pub fn on_key( return Vec::new(); } - // 3) A launch job is showing in the console. + // 4) A launch job is showing in the console. if let Some(job_id) = w.active_job.clone() { match on_console_key(&job_id, jobs, key) { ConsoleOutcome::Cancelled(fx) => return fx, @@ -298,7 +512,15 @@ pub fn on_key( return Vec::new(); } - // 4) Form editing. + // 5) Inline text editor (Host / custom Port) owns the keyboard. It is not a + // modal: the form stays painted underneath, and form navigation keys + // cannot reach hidden rows from here. + if w.editor.is_some() { + editor_key(w, key.code); + return Vec::new(); + } + + // 6) Form editing. match key.code { KeyCode::Esc => *wizard = None, KeyCode::Up => w.move_field(-1), @@ -306,15 +528,24 @@ pub fn on_key( KeyCode::Left => w.cycle(-1), KeyCode::Right => w.cycle(1), KeyCode::Char(' ') if w.current_field() == Field::Mode => w.cycle(1), + KeyCode::Char(' ') if w.current_field() == Field::Advanced => { + w.advanced_expanded = !w.advanced_expanded; + } // Tab on the Model field opens the local-path picker (Wave-0 primitive). KeyCode::Tab if w.current_field() == Field::Model => { let start = std::env::current_dir().unwrap_or_else(|_| std::path::PathBuf::from("/")); w.browser = Some(FolderBrowser::new("Pick a local model path", start)); } - KeyCode::Enter => { - if w.current_field() == Field::Launch { - request_launch(w); - } else if w.current_field() == Field::Model && !recipes.is_empty() { + KeyCode::Enter => match w.current_field() { + Field::Launch => request_launch(w), + // Expand/collapse only ever happens while Advanced owns focus, so + // focus lands back on Advanced in both directions. + Field::Advanced => w.advanced_expanded = !w.advanced_expanded, + Field::Host => w.editor = Some(InlineEditor::new(Field::Host, &w.host)), + Field::Port if w.port_mode == PortMode::Custom => { + w.editor = Some(InlineEditor::new(Field::Port, &w.port)); + } + Field::Model if !recipes.is_empty() => { // On the Model field, Enter opens the recipe picker (the // model_picker sub-step); free-text typing + Tab-browse remain. // Seed the filter with anything already typed so the picker @@ -323,10 +554,9 @@ pub fn on_key( query: w.model.trim().to_string(), selected: 0, }); - } else { - w.move_field(1); } - } + _ => w.move_field(1), + }, KeyCode::Backspace => w.backspace(), KeyCode::Char(c) => w.type_char(c), _ => {} @@ -334,25 +564,114 @@ pub fn on_key( Vec::new() } +/// Drive the inline editor. Enter accepts, Escape restores, and nothing here +/// can open another overlay. +fn editor_key(w: &mut ServeWizardState, code: KeyCode) { + if w.editor.is_none() { + return; + } + match code { + KeyCode::Esc => { + // Restore the original — an abandoned edit never mutates the form. + if let Some(ed) = w.editor.take() { + match ed.field { + Field::Host => w.host = ed.original, + Field::Port => w.port = ed.original, + _ => {} + } + } + refresh_message(w); + } + KeyCode::Left | KeyCode::Right => { + let delta = if code == KeyCode::Left { -1 } else { 1 }; + if let Some(ed) = w.editor.as_mut() { + ed.move_cursor(delta); + } + } + KeyCode::Backspace => { + if let Some(ed) = w.editor.as_mut() { + ed.backspace(); + } + } + KeyCode::Enter => { + let Some((field, value)) = w.editor.as_ref().map(|e| (e.field, e.value.clone())) else { + return; + }; + match field { + Field::Host => { + let trimmed = value.trim(); + if trimmed.is_empty() { + // Keep the editor open with a plain fix rather than + // silently accepting an unusable host. + w.message = Some(format!( + "host cannot be empty; type an address or press Esc to restore {DEFAULT_HOST}" + )); + return; + } + w.host = trimmed.to_string(); + } + Field::Port => { + if value.trim().is_empty() { + // Mirror the Host rule: an empty custom port is not a + // value, so keep the editor open with a plain fix. + w.message = Some( + "port cannot be empty; type a 1–65535 value or press Esc to cancel" + .to_string(), + ); + return; + } + // Digits-only is enforced on input; range validation happens + // at review time so an in-progress value stays visible. + w.port = value; + } + _ => {} + } + refresh_message(w); + w.editor = None; + } + KeyCode::Char(c) => { + if let Some(ed) = w.editor.as_mut() { + if ed.field == Field::Port && !c.is_ascii_digit() { + return; + } + ed.insert(c); + } + } + _ => {} + } +} + +/// Re-evaluate an *outstanding* validation complaint against the current form: +/// a genuine fix clears it, a still-broken form re-states the reason. Silent +/// when nothing has been complained about yet, so ordinary edits stay quiet. +fn refresh_message(w: &mut ServeWizardState) { + if w.message.is_some() { + w.message = w.validate().err().map(|(_, msg)| msg); + } +} + /// Validate the form and stage an approval (no job runs until approved). fn request_launch(w: &mut ServeWizardState) { match w.build_args() { Ok(args) => { let cmd = resolve_exe(); let cmdline = format!("{} {}", exe_label(&cmd), args.join(" ")); - let request = ApprovalRequest::new( - format!("serve “{}”", w.model.trim()), - vec![ - cmdline, - String::new(), - "This launches a local model server through the ROCm CLI.".to_string(), - if w.managed { - "Managed: it will appear in the services manager and dashboard.".to_string() - } else { - "Foreground: it runs in this job console until stopped.".to_string() - }, - ], - ); + let mut body = vec![ + cmdline, + String::new(), + "This launches a local model server through the ROCm CLI.".to_string(), + ]; + // Only claim an automatic port once the combination is valid — the + // concrete endpoint is printed by the CLI after it leases one. + if w.port_mode == PortMode::Auto { + body.push(AUTO_PORT_NOTE.to_string()); + } + body.push(if w.managed { + "Managed: it will appear in the services manager and dashboard.".to_string() + } else { + "Foreground: it runs in this job console until stopped.".to_string() + }); + let request = ApprovalRequest::new(format!("serve “{}”", w.model.trim()), body); w.message = None; w.approval = Some(PendingServe { cmd, @@ -361,7 +680,12 @@ fn request_launch(w: &mut ServeWizardState) { choice: ApprovalChoice::default(), }); } - Err(msg) => w.message = Some(msg), + Err((field, msg)) => { + // Move focus to the problem — expanding Advanced when the offending + // row is hidden — so the fix is always visible, never guessed at. + w.focus(field); + w.message = Some(msg); + } } } @@ -428,10 +752,12 @@ pub fn draw_serve_wizard( .split(inner); let has_recipes = !recipes.is_empty(); - let lines: Vec = FIELDS + let vis = w.visible_fields(); + let focused = w.field.min(vis.len() - 1); + let lines: Vec = vis .iter() .enumerate() - .map(|(i, field)| field_line(*field, i == w.field, w, has_recipes, theme)) + .map(|(i, field)| field_line(*field, i == focused, w, has_recipes, theme)) .collect(); f.render_widget(Paragraph::new(lines), rows[0]); @@ -444,14 +770,9 @@ pub fn draw_serve_wizard( rows[1], ); - let hint = if recipes.is_empty() { - "↑↓ field · ←→ change · Tab browse (model) · Enter next/launch · Esc close" - } else { - "↑↓ field · ←→ change · Enter pick (model)/next/launch · Tab browse · Esc close" - }; f.render_widget( Paragraph::new(Line::from(Span::styled( - hint, + footer_hint(w, has_recipes), Style::default().fg(theme.muted), ))), rows[2], @@ -469,6 +790,41 @@ pub fn draw_serve_wizard( } } +/// Footer help, derived from focus and edit state. Every hint names the row's +/// role in words (`choice`, `editable`, `read only`) so meaning never depends +/// on colour or glyphs alone. +fn footer_hint(w: &ServeWizardState, has_recipes: bool) -> &'static str { + if let Some(ed) = &w.editor { + return match ed.field { + Field::Port => { + "editing port · ←→ cursor · Backspace delete · Enter accept · Esc cancel" + } + _ => "editing host · ←→ cursor · Backspace delete · Enter accept · Esc cancel", + }; + } + match w.current_field() { + Field::Model if has_recipes => { + "editable · Enter pick a recipe · type a name · Tab browse a path · Esc close" + } + Field::Model => "editable · type a name or path · Tab browse a path · Esc close", + Field::Advanced if w.advanced_expanded => { + "Enter or ← collapse advanced settings · defaults stay automatic" + } + Field::Advanced => "Enter or → expand advanced settings · defaults stay automatic", + Field::Engine => "choice · ←→ pick an engine · automatic lets the CLI decide", + Field::Device => "read only · GPU required · ROCm never falls back to CPU", + Field::Host => "editable · Enter to edit the host · loopback only in Dash", + Field::Port if w.port_mode == PortMode::Custom => { + "choice · ←→ back to automatic · Enter to edit the port" + } + Field::Port => "choice · ←→ switch to custom · automatic picks a free local port", + Field::Mode => "choice · ←→ managed or foreground", + Field::Launch => "Enter to review the exact command · nothing runs before you approve", + } +} + +/// Render one row. Choice values carry chevrons, editable values brackets, and +/// the automatic device policy is plain read-only text. fn field_line<'a>( field: Field, selected: bool, @@ -476,34 +832,6 @@ fn field_line<'a>( has_recipes: bool, theme: &Theme, ) -> Line<'a> { - let model_placeholder = if has_recipes { - "(Enter to pick a recipe · type a name · Tab to browse)" - } else { - "(type a name / path, or Tab to browse)" - }; - let (label, value): (&str, String) = match field { - Field::Model => ("Model", display_value(&w.model, model_placeholder)), - Field::Engine => ( - "Engine", - ENGINES[w.engine_idx.min(ENGINES.len() - 1)].to_string(), - ), - Field::Device => ( - "Device", - DEVICES[w.device_idx.min(DEVICES.len() - 1)].to_string(), - ), - Field::Host => ("Host", display_value(&w.host, "(engine default)")), - Field::Port => ("Port", display_value(&w.port, "(engine default)")), - Field::Mode => ( - "Mode", - if w.managed { - "managed".to_string() - } else { - "foreground".to_string() - }, - ), - Field::Launch => ("", String::new()), - }; - if field == Field::Launch { let style = if selected { Style::default() @@ -529,21 +857,127 @@ fn field_line<'a>( } else { Style::default().fg(theme.fg) }; - Line::from(vec![ + + if field == Field::Advanced { + let chevron = if w.advanced_expanded { "▾" } else { "▸" }; + return Line::from(vec![ + Span::styled(marker, label_style), + Span::styled("Advanced settings ", label_style), + Span::styled(chevron, label_style), + Span::styled(format!(" {}", w.advanced_summary()), value_style), + ]); + } + + let model_placeholder = if has_recipes { + "(Enter to pick a recipe · type a name · Tab to browse)" + } else { + "(type a name / path, or Tab to browse)" + }; + + let label = match field { + Field::Model => "Model", + Field::Engine => "Engine", + Field::Device => "Device", + Field::Host => "Host", + Field::Port => "Port", + Field::Mode => "Mode", + Field::Advanced | Field::Launch => unreachable!(), + }; + + let mut spans = vec![ Span::styled(marker, label_style), Span::styled(format!("{label:<8}"), label_style), - Span::styled(value, value_style), - ]) + ]; + + // The editor, when open, replaces its own row's value with a live caret. + let editing = w.editor.as_ref().filter(|e| e.field == field); + + match field { + Field::Model => spans.push(Span::styled( + bracketed(&w.model, model_placeholder), + value_style, + )), + Field::Engine => spans.push(Span::styled( + chevroned(ENGINES[w.engine_idx.min(ENGINES.len() - 1)]), + value_style, + )), + // Read-only: no chevrons, no brackets — nothing to change here. + Field::Device => spans.push(Span::styled(DEVICE_POLICY, value_style)), + Field::Host => { + if let Some(ed) = editing { + spans.extend(editor_spans(ed, value_style, theme)); + } else { + spans.push(Span::styled(bracketed(&w.host, "(unset)"), value_style)); + } + } + Field::Port => { + let mode = match w.port_mode { + PortMode::Auto => "automatic", + PortMode::Custom => "custom", + }; + spans.push(Span::styled(chevroned(mode), value_style)); + if w.port_mode == PortMode::Custom { + spans.push(Span::styled(" ", value_style)); + if let Some(ed) = editing { + spans.extend(editor_spans(ed, value_style, theme)); + } else { + spans.push(Span::styled(bracketed(&w.port, "(unset)"), value_style)); + } + } + } + Field::Mode => spans.push(Span::styled( + chevroned(if w.managed { "managed" } else { "foreground" }), + value_style, + )), + Field::Advanced | Field::Launch => unreachable!(), + } + + Line::from(spans) } -fn display_value(v: &str, placeholder: &'static str) -> String { +/// Editable value with a visible caret (and a reversed run while the initial +/// value is still selected), rendered so the text stays contiguous. +fn editor_spans<'a>(ed: &'a InlineEditor, base: Style, theme: &Theme) -> Vec> { + let caret = Style::default() + .bg(theme.accent) + .fg(theme.bg) + .add_modifier(Modifier::BOLD); + let mut spans = vec![Span::styled("[", base)]; + if ed.selected && !ed.value.is_empty() { + spans.push(Span::styled(ed.value.as_str(), caret)); + } else { + let chars: Vec = ed.value.chars().collect(); + let cut = ed.cursor.min(chars.len()); + let before: String = chars[..cut].iter().collect(); + if !before.is_empty() { + spans.push(Span::styled(before, base)); + } + if cut < chars.len() { + spans.push(Span::styled(chars[cut].to_string(), caret)); + let after: String = chars[cut + 1..].iter().collect(); + if !after.is_empty() { + spans.push(Span::styled(after, base)); + } + } else { + spans.push(Span::styled(" ", caret)); + } + } + spans.push(Span::styled("]", base)); + spans +} + +fn bracketed(v: &str, placeholder: &'static str) -> String { if v.is_empty() { placeholder.to_string() } else { - v.to_string() + format!("[{v}]") } } +fn chevroned(v: &str) -> String { + format!("‹ {v} ›") +} + #[cfg(test)] mod tests { use super::*; @@ -557,6 +991,17 @@ mod tests { s.chars().map(|c| key(KeyCode::Char(c))).collect() } + fn feed(wiz: &mut Option, jobs: &mut State, keys: &[KeyCode]) { + for c in keys { + on_key(wiz, jobs, &[], key(*c)); + } + } + + /// Put focus on a row, expanding Advanced when that row is hidden. + fn focus_field(w: &mut ServeWizardState, field: Field) { + w.focus(field); + } + #[test] fn cycle_idx_wraps_both_directions() { assert_eq!(cycle_idx(0, 3, -1), 2); @@ -564,113 +1009,649 @@ mod tests { assert_eq!(cycle_idx(0, 0, 1), 0); } + // ---------------------------------------------------------------- defaults + #[test] - fn default_form_targets_managed_lemonade() { + fn default_form_is_fully_automatic_and_collapsed() { let w = ServeWizardState::default(); - assert!(w.managed); - assert_eq!(ENGINES[w.engine_idx], "lemonade"); - assert_eq!(w.device_idx, 0); // engine default → no --device + assert!(!w.advanced_expanded); + assert_eq!(w.visible_fields(), BASIC_FIELDS); + assert_eq!(w.engine_idx, ENGINE_AUTO); + assert_eq!(ENGINES[w.engine_idx], "automatic"); assert_eq!(w.host, "127.0.0.1"); - assert_eq!(w.port, "11435"); + assert_eq!(w.port_mode, PortMode::Auto); + assert!(w.managed); + assert_eq!(w.advanced_summary(), "Automatic"); + assert!(!w.is_customized()); } + // ------------------------------------------------------------------- argv + #[test] fn build_args_requires_a_model() { let w = ServeWizardState::default(); - assert_eq!(w.build_args().unwrap_err(), "model is required"); + let (field, msg) = w.build_args().unwrap_err(); + assert_eq!(field, Field::Model); + assert_eq!(msg, "model is required"); + } + + #[test] + fn default_argv_is_bare_serve_model() { + let w = ServeWizardState { + model: "qwen".into(), + ..Default::default() + }; + assert_eq!(w.build_args().unwrap(), vec!["serve", "qwen"]); } #[test] - fn build_args_emits_managed_serve_with_defaults() { + fn automatic_defaults_omit_engine_device_host_port_and_managed() { let w = ServeWizardState { model: "qwen".into(), ..Default::default() }; let args = w.build_args().unwrap(); + for flag in ["--engine", "--device", "--host", "--port", "--managed"] { + assert!(!args.contains(&flag.to_string()), "{flag} must be omitted"); + } + } + + #[test] + fn explicit_overrides_emit_only_their_own_flags() { + let w = ServeWizardState { + model: "glm".into(), + engine_idx: 2, // vllm + host: "localhost".into(), + port_mode: PortMode::Custom, + port: "8000".into(), + managed: false, + ..Default::default() + }; assert_eq!( - args, + w.build_args().unwrap(), vec![ "serve", - "qwen", + "glm", "--engine", - "lemonade", + "vllm", "--host", - "127.0.0.1", + "localhost", "--port", - "11435", - "--managed", + "8000", + "--foreground", ] ); } #[test] - fn build_args_includes_device_when_not_default_and_foreground() { + fn custom_port_on_default_host_emits_port_only() { let w = ServeWizardState { - model: "glm".into(), - device_idx: 1, // gpu_required - managed: false, + model: "m".into(), + port_mode: PortMode::Custom, + port: "11500".into(), ..Default::default() }; - let args = w.build_args().unwrap(); - assert!(args.windows(2).any(|p| p == ["--device", "gpu_required"])); - assert!(args.contains(&"--foreground".to_string())); - assert!(!args.contains(&"--managed".to_string())); + assert_eq!( + w.build_args().unwrap(), + vec!["serve", "m", "--port", "11500"] + ); + } + + #[test] + fn build_args_rejects_out_of_range_and_zero_ports() { + for bad in ["99999", "0", "", " "] { + let w = ServeWizardState { + model: "m".into(), + port_mode: PortMode::Custom, + port: bad.into(), + ..Default::default() + }; + let (field, msg) = w.build_args().unwrap_err(); + assert_eq!(field, Field::Port, "{bad:?}"); + assert!(msg.contains("1–65535"), "{bad:?}: {msg}"); + } } #[test] - fn build_args_rejects_bad_port() { + fn empty_host_is_rejected_with_a_plain_fix() { let w = ServeWizardState { model: "m".into(), - port: "99999".into(), + host: " ".into(), ..Default::default() }; - assert!(w.build_args().unwrap_err().contains("port")); + let (field, msg) = w.build_args().unwrap_err(); + assert_eq!(field, Field::Host); + assert!(msg.contains("host is required"), "{msg}"); + } + + #[test] + fn noncanonical_host_with_auto_port_is_blocked_with_the_contract_message() { + for host in ["localhost", "::1", "0.0.0.0", "192.168.1.5"] { + let w = ServeWizardState { + model: "m".into(), + host: host.into(), + port_mode: PortMode::Auto, + ..Default::default() + }; + let (field, msg) = w.build_args().unwrap_err(); + assert_eq!(field, Field::Port, "{host}"); + assert_eq!(msg, CUSTOM_HOST_NEEDS_PORT, "{host}"); + } + } + + #[test] + fn exact_cli_loopback_hosts_with_custom_port_are_allowed() { + for host in ["127.0.0.1", "localhost", "::1"] { + let w = ServeWizardState { + model: "m".into(), + host: host.into(), + port_mode: PortMode::Custom, + port: "9000".into(), + ..Default::default() + }; + let expected = if host == DEFAULT_HOST { + vec!["serve", "m", "--port", "9000"] + } else { + vec!["serve", "m", "--host", host, "--port", "9000"] + }; + assert_eq!(w.build_args().unwrap(), expected, "{host}"); + } + } + + #[test] + fn public_host_never_reaches_approval_even_with_a_custom_port() { + // Dash has no --allow-public-bind confirmation and no endpoint-key UI, + // so a public bind must be requested from the CLI instead. + for host in [ + "0.0.0.0", + "192.168.1.5", + "::", + "example.invalid", + "LocalHost", + "[::1]", + ] { + let w = ServeWizardState { + model: "m".into(), + host: host.into(), + port_mode: PortMode::Custom, + port: "8000".into(), + ..Default::default() + }; + let (field, msg) = w.build_args().unwrap_err(); + assert_eq!(field, Field::Host, "{host}"); + assert_eq!(msg, PUBLIC_HOST_NEEDS_CLI, "{host}"); + } + } + + #[test] + fn public_host_launch_stages_no_approval() { + let mut wiz = Some(ServeWizardState { + model: "m".into(), + host: "0.0.0.0".into(), + port_mode: PortMode::Custom, + port: "8000".into(), + ..Default::default() + }); + let mut jobs = State::default(); + wiz.as_mut().unwrap().focus(Field::Launch); + let fx = on_key(&mut wiz, &mut jobs, &[], key(KeyCode::Enter)); + assert!(fx.is_empty()); + let w = wiz.as_ref().unwrap(); + assert!(w.approval.is_none(), "a public bind never reaches approval"); + assert!(w.advanced_expanded, "the offending row is revealed"); + assert_eq!(w.current_field(), Field::Host); + assert_eq!(w.message.as_deref(), Some(PUBLIC_HOST_NEEDS_CLI)); + assert!(jobs.jobs.is_empty()); } #[test] - fn build_args_rejects_port_zero() { - // u16 parses 0, but it is not a bindable listen port — must be rejected - // in the form, matching the "1–65535" message. + fn is_loopback_host_matches_the_cli_exactly() { + for ok in ["127.0.0.1", "localhost", "::1"] { + assert!(is_loopback_host(ok), "{ok}"); + } + for bad in [ + "0.0.0.0", + "127.0.0.2", + "::", + "192.168.1.5", + "", + "LocalHost", + "[::1]", + ] { + assert!(!is_loopback_host(bad), "{bad}"); + } + } + + // -------------------------------------------------------- disclosure/focus + + #[test] + fn advanced_expands_and_collapses_only_while_focused_and_keeps_focus() { + let mut wiz = Some(ServeWizardState::default()); + let mut jobs = State::default(); + // Model has focus: Right must not expand anything. + feed(&mut wiz, &mut jobs, &[KeyCode::Right]); + assert!(!wiz.as_ref().unwrap().advanced_expanded); + + // Down → Advanced, Enter expands, focus stays on Advanced. + feed(&mut wiz, &mut jobs, &[KeyCode::Down, KeyCode::Enter]); + { + let w = wiz.as_ref().unwrap(); + assert!(w.advanced_expanded); + assert_eq!(w.visible_fields(), FIELDS); + assert_eq!(w.current_field(), Field::Advanced); + } + // Enter collapses again, still on Advanced. + feed(&mut wiz, &mut jobs, &[KeyCode::Enter]); + { + let w = wiz.as_ref().unwrap(); + assert!(!w.advanced_expanded); + assert_eq!(w.current_field(), Field::Advanced); + } + // ← collapses / → expands from the same row. + feed(&mut wiz, &mut jobs, &[KeyCode::Right]); + assert!(wiz.as_ref().unwrap().advanced_expanded); + feed(&mut wiz, &mut jobs, &[KeyCode::Left]); + assert!(!wiz.as_ref().unwrap().advanced_expanded); + assert_eq!(wiz.as_ref().unwrap().current_field(), Field::Advanced); + } + + #[test] + fn expanded_focus_order_is_model_advanced_engine_device_host_port_mode_launch() { + let mut w = ServeWizardState { + advanced_expanded: true, + ..Default::default() + }; + let mut seen = Vec::new(); + for _ in 0..FIELDS.len() { + seen.push(w.current_field()); + w.move_field(1); + } + assert_eq!(seen, FIELDS.to_vec()); + // Clamped at the end. + assert_eq!(w.current_field(), Field::Launch); + } + + #[test] + fn collapsing_preserves_overrides_and_reports_customized() { + let mut w = ServeWizardState { + advanced_expanded: true, + engine_idx: 1, + port_mode: PortMode::Custom, + port: "8000".into(), + ..Default::default() + }; + w.advanced_expanded = false; + assert_eq!(w.engine_idx, 1); + assert_eq!(w.port, "8000"); + assert_eq!(w.advanced_summary(), "Customized"); + } + + #[test] + fn invalid_hidden_custom_port_says_it_needs_attention() { let w = ServeWizardState { model: "m".into(), - port: "0".into(), + port_mode: PortMode::Custom, + port: "abc".into(), ..Default::default() }; - assert!(w.build_args().unwrap_err().contains("port")); + assert!(!w.advanced_expanded); + assert_eq!(w.advanced_summary(), "Customized · port needs attention"); + assert!(w.build_args().is_err(), "launch stays blocked"); } #[test] - fn port_field_accepts_digits_only() { - let mut w = ServeWizardState::default(); - w.port.clear(); - w.field = FIELDS.iter().position(|f| *f == Field::Port).unwrap(); - for k in typed("80a0") { - // route through type_char via the field - if let KeyCode::Char(c) = k.code { - w.type_char(c); - } + fn out_of_range_hidden_port_also_needs_attention() { + let w = ServeWizardState { + port_mode: PortMode::Custom, + port: "70000".into(), + ..Default::default() + }; + assert_eq!(w.advanced_summary(), "Customized · port needs attention"); + } + + #[test] + fn invalid_combination_expands_advanced_and_focuses_the_problem() { + let mut wiz = Some(ServeWizardState { + model: "m".into(), + host: "localhost".into(), + ..Default::default() + }); + let mut jobs = State::default(); + wiz.as_mut().unwrap().field = 2; // Launch (collapsed order) + let fx = on_key(&mut wiz, &mut jobs, &[], key(KeyCode::Enter)); + assert!(fx.is_empty(), "nothing may run"); + let w = wiz.as_ref().unwrap(); + assert!(w.approval.is_none(), "invalid combos never reach approval"); + assert!(w.advanced_expanded, "the hidden problem is revealed"); + assert_eq!(w.current_field(), Field::Port); + assert_eq!(w.message.as_deref(), Some(CUSTOM_HOST_NEEDS_PORT)); + assert_eq!(w.host, "localhost", "both values preserved"); + assert_eq!(w.port_mode, PortMode::Auto); + } + + // --------------------------------------------------------------- controls + + #[test] + fn left_right_cycles_engine_including_automatic() { + let mut wiz = Some(ServeWizardState { + advanced_expanded: true, + ..Default::default() + }); + let mut jobs = State::default(); + focus_field(wiz.as_mut().unwrap(), Field::Engine); + feed(&mut wiz, &mut jobs, &[KeyCode::Right]); + assert_eq!(ENGINES[wiz.as_ref().unwrap().engine_idx], "lemonade"); + feed(&mut wiz, &mut jobs, &[KeyCode::Right]); + assert_eq!(ENGINES[wiz.as_ref().unwrap().engine_idx], "vllm"); + feed(&mut wiz, &mut jobs, &[KeyCode::Right]); + assert_eq!(wiz.as_ref().unwrap().engine_idx, ENGINE_AUTO, "wraps"); + } + + #[test] + fn left_right_toggles_port_mode_only_when_focused() { + let mut wiz = Some(ServeWizardState { + advanced_expanded: true, + ..Default::default() + }); + let mut jobs = State::default(); + focus_field(wiz.as_mut().unwrap(), Field::Device); + feed(&mut wiz, &mut jobs, &[KeyCode::Right]); + assert_eq!( + wiz.as_ref().unwrap().port_mode, + PortMode::Auto, + "Device is read-only and cannot change the port" + ); + focus_field(wiz.as_mut().unwrap(), Field::Port); + feed(&mut wiz, &mut jobs, &[KeyCode::Right]); + assert_eq!(wiz.as_ref().unwrap().port_mode, PortMode::Custom); + feed(&mut wiz, &mut jobs, &[KeyCode::Left]); + assert_eq!(wiz.as_ref().unwrap().port_mode, PortMode::Auto); + } + + #[test] + fn mode_toggles_between_managed_and_foreground() { + let mut wiz = Some(ServeWizardState { + advanced_expanded: true, + ..Default::default() + }); + let mut jobs = State::default(); + focus_field(wiz.as_mut().unwrap(), Field::Mode); + feed(&mut wiz, &mut jobs, &[KeyCode::Char(' ')]); + assert!(!wiz.as_ref().unwrap().managed); + feed(&mut wiz, &mut jobs, &[KeyCode::Left]); + assert!(wiz.as_ref().unwrap().managed); + } + + #[test] + fn typing_only_edits_the_model_row() { + let mut wiz = Some(ServeWizardState { + advanced_expanded: true, + ..Default::default() + }); + let mut jobs = State::default(); + focus_field(wiz.as_mut().unwrap(), Field::Host); + for k in typed("zzz") { + on_key(&mut wiz, &mut jobs, &[], k); + } + assert_eq!( + wiz.as_ref().unwrap().host, + "127.0.0.1", + "host only changes through the inline editor" + ); + focus_field(wiz.as_mut().unwrap(), Field::Model); + for k in typed("qwen") { + on_key(&mut wiz, &mut jobs, &[], k); + } + assert_eq!(wiz.as_ref().unwrap().model, "qwen"); + feed(&mut wiz, &mut jobs, &[KeyCode::Backspace]); + assert_eq!(wiz.as_ref().unwrap().model, "qwe"); + } + + // ----------------------------------------------------------- inline editor + + #[test] + fn host_editor_selects_on_entry_and_first_char_replaces() { + let mut wiz = Some(ServeWizardState { + advanced_expanded: true, + ..Default::default() + }); + let mut jobs = State::default(); + focus_field(wiz.as_mut().unwrap(), Field::Host); + feed(&mut wiz, &mut jobs, &[KeyCode::Enter]); + { + let ed = wiz.as_ref().unwrap().editor.as_ref().unwrap(); + assert_eq!(ed.field, Field::Host); + assert!(ed.selected); + assert_eq!(ed.value, "127.0.0.1"); + } + for k in typed("0.0.0.0") { + on_key(&mut wiz, &mut jobs, &[], k); + } + assert_eq!( + wiz.as_ref().unwrap().editor.as_ref().unwrap().value, + "0.0.0.0" + ); + feed(&mut wiz, &mut jobs, &[KeyCode::Enter]); + let w = wiz.as_ref().unwrap(); + assert!(w.editor.is_none()); + assert_eq!(w.host, "0.0.0.0"); + } + + #[test] + fn editor_cursor_insert_and_backspace() { + let mut wiz = Some(ServeWizardState { + advanced_expanded: true, + host: "abc".into(), + ..Default::default() + }); + let mut jobs = State::default(); + focus_field(wiz.as_mut().unwrap(), Field::Host); + feed(&mut wiz, &mut jobs, &[KeyCode::Enter, KeyCode::Left]); + // Left cleared the selection and moved the cursor to before 'c'. + { + let ed = wiz.as_ref().unwrap().editor.as_ref().unwrap(); + assert!(!ed.selected); + assert_eq!(ed.cursor, 2); + } + on_key(&mut wiz, &mut jobs, &[], key(KeyCode::Char('X'))); + assert_eq!(wiz.as_ref().unwrap().editor.as_ref().unwrap().value, "abXc"); + feed(&mut wiz, &mut jobs, &[KeyCode::Backspace]); + assert_eq!(wiz.as_ref().unwrap().editor.as_ref().unwrap().value, "abc"); + feed(&mut wiz, &mut jobs, &[KeyCode::Right, KeyCode::Right]); + assert_eq!(wiz.as_ref().unwrap().editor.as_ref().unwrap().cursor, 3); + feed(&mut wiz, &mut jobs, &[KeyCode::Enter]); + assert_eq!(wiz.as_ref().unwrap().host, "abc"); + } + + #[test] + fn editor_escape_restores_the_original_value() { + let mut wiz = Some(ServeWizardState { + advanced_expanded: true, + ..Default::default() + }); + let mut jobs = State::default(); + focus_field(wiz.as_mut().unwrap(), Field::Host); + feed(&mut wiz, &mut jobs, &[KeyCode::Enter]); + for k in typed("nonsense") { + on_key(&mut wiz, &mut jobs, &[], k); + } + feed(&mut wiz, &mut jobs, &[KeyCode::Esc]); + let w = wiz.as_ref().unwrap(); + assert!(w.editor.is_none(), "Esc closes the editor"); + assert!(wiz.is_some(), "Esc in the editor never closes the overlay"); + assert_eq!(w.host, "127.0.0.1", "original restored"); + } + + #[test] + fn host_editor_trims_and_rejects_blank() { + let mut wiz = Some(ServeWizardState { + advanced_expanded: true, + ..Default::default() + }); + let mut jobs = State::default(); + focus_field(wiz.as_mut().unwrap(), Field::Host); + feed(&mut wiz, &mut jobs, &[KeyCode::Enter]); + for k in typed(" ") { + on_key(&mut wiz, &mut jobs, &[], k); + } + feed(&mut wiz, &mut jobs, &[KeyCode::Enter]); + { + let w = wiz.as_ref().unwrap(); + assert!(w.editor.is_some(), "blank host keeps the editor open"); + assert_eq!(w.host, "127.0.0.1", "form untouched"); + assert!( + w.message + .as_deref() + .unwrap() + .contains("host cannot be empty") + ); + } + for k in typed(" 10.0.0.7 ") { + on_key(&mut wiz, &mut jobs, &[], k); + } + feed(&mut wiz, &mut jobs, &[KeyCode::Enter]); + let w = wiz.as_ref().unwrap(); + assert!(w.editor.is_none()); + assert_eq!(w.host, "10.0.0.7", "trimmed on accept"); + } + + #[test] + fn port_editor_accepts_digits_only() { + let mut wiz = Some(ServeWizardState { + advanced_expanded: true, + port_mode: PortMode::Custom, + ..Default::default() + }); + let mut jobs = State::default(); + focus_field(wiz.as_mut().unwrap(), Field::Port); + feed(&mut wiz, &mut jobs, &[KeyCode::Enter]); + for k in typed("80a0!") { + on_key(&mut wiz, &mut jobs, &[], k); } - assert_eq!(w.port, "800"); + assert_eq!(wiz.as_ref().unwrap().editor.as_ref().unwrap().value, "800"); + feed(&mut wiz, &mut jobs, &[KeyCode::Enter]); + assert_eq!(wiz.as_ref().unwrap().port, "800"); } + #[test] + fn port_editor_rejects_an_empty_value_like_host_does() { + let mut wiz = Some(ServeWizardState { + advanced_expanded: true, + model: "m".into(), + port_mode: PortMode::Custom, + ..Default::default() + }); + let mut jobs = State::default(); + focus_field(wiz.as_mut().unwrap(), Field::Port); + // Enter selects "11435"; Backspace clears the whole selection. + feed(&mut wiz, &mut jobs, &[KeyCode::Enter, KeyCode::Backspace]); + assert_eq!(wiz.as_ref().unwrap().editor.as_ref().unwrap().value, ""); + feed(&mut wiz, &mut jobs, &[KeyCode::Enter]); + { + let w = wiz.as_ref().unwrap(); + assert!(w.editor.is_some(), "empty port keeps the editor open"); + assert_eq!(w.port, "11435", "form untouched"); + let msg = w.message.as_deref().unwrap(); + assert!(msg.contains("port cannot be empty"), "{msg}"); + assert!(msg.contains("1–65535"), "{msg}"); + } + // Typing a real value accepts normally and clears the complaint. + for k in typed("9001") { + on_key(&mut wiz, &mut jobs, &[], k); + } + feed(&mut wiz, &mut jobs, &[KeyCode::Enter]); + let w = wiz.as_ref().unwrap(); + assert!(w.editor.is_none()); + assert_eq!(w.port, "9001"); + assert!(w.message.is_none(), "a fixed form stops complaining"); + } + + #[test] + fn escaping_an_editor_recomputes_an_outstanding_complaint() { + // A blank-host complaint must not survive the Esc that restores a + // perfectly good host… + let mut wiz = Some(ServeWizardState { + advanced_expanded: true, + model: "m".into(), + ..Default::default() + }); + let mut jobs = State::default(); + focus_field(wiz.as_mut().unwrap(), Field::Host); + feed( + &mut wiz, + &mut jobs, + &[KeyCode::Enter, KeyCode::Backspace, KeyCode::Enter], + ); + assert!(wiz.as_ref().unwrap().message.is_some()); + feed(&mut wiz, &mut jobs, &[KeyCode::Esc]); + let w = wiz.as_ref().unwrap(); + assert_eq!(w.host, "127.0.0.1"); + assert!(w.message.is_none(), "stale complaint cleared by the fix"); + + // …but a still-broken form keeps saying why. + let mut wiz = Some(ServeWizardState { + advanced_expanded: true, + model: "m".into(), + host: "localhost".into(), + ..Default::default() + }); + wiz.as_mut().unwrap().message = Some(CUSTOM_HOST_NEEDS_PORT.to_string()); + focus_field(wiz.as_mut().unwrap(), Field::Host); + feed(&mut wiz, &mut jobs, &[KeyCode::Enter, KeyCode::Esc]); + let w = wiz.as_ref().unwrap(); + assert_eq!(w.host, "localhost"); + assert_eq!(w.message.as_deref(), Some(CUSTOM_HOST_NEEDS_PORT)); + } + + #[test] + fn enter_on_auto_port_advances_instead_of_editing() { + let mut wiz = Some(ServeWizardState { + advanced_expanded: true, + ..Default::default() + }); + let mut jobs = State::default(); + focus_field(wiz.as_mut().unwrap(), Field::Port); + feed(&mut wiz, &mut jobs, &[KeyCode::Enter]); + let w = wiz.as_ref().unwrap(); + assert!(w.editor.is_none()); + assert_eq!(w.current_field(), Field::Mode); + } + + #[test] + fn editor_navigation_keys_cannot_move_form_focus() { + let mut wiz = Some(ServeWizardState { + advanced_expanded: true, + ..Default::default() + }); + let mut jobs = State::default(); + focus_field(wiz.as_mut().unwrap(), Field::Host); + let before = wiz.as_ref().unwrap().field; + feed( + &mut wiz, + &mut jobs, + &[KeyCode::Enter, KeyCode::Up, KeyCode::Down], + ); + let w = wiz.as_ref().unwrap(); + assert!(w.editor.is_some()); + assert_eq!(w.field, before, "form navigation is inert while editing"); + } + + // -------------------------------------------------------------- approval + #[test] fn launch_requires_approval_then_spawns_job() { let mut wiz = Some(ServeWizardState::default()); let mut jobs = State::default(); - // Fill the model by typing on the Model field (field 0 by default). for k in typed("qwen") { on_key(&mut wiz, &mut jobs, &[], k); } assert_eq!(wiz.as_ref().unwrap().model, "qwen"); - // Jump to Launch and press Enter → approval staged, NO job yet. - let launch_idx = FIELDS.iter().position(|f| *f == Field::Launch).unwrap(); - wiz.as_mut().unwrap().field = launch_idx; + wiz.as_mut().unwrap().focus(Field::Launch); let fx = on_key(&mut wiz, &mut jobs, &[], key(KeyCode::Enter)); assert!(fx.is_empty(), "launch must not run before approval"); assert!(wiz.as_ref().unwrap().approval.is_some()); assert!(jobs.jobs.is_empty()); - // Approve → exactly one SpawnJob, job registered, console active. let fx = on_key(&mut wiz, &mut jobs, &[], key(KeyCode::Char('y'))); assert_eq!(fx.len(), 1); assert!(matches!(fx[0], SideEffect::SpawnJob { .. })); @@ -680,17 +1661,49 @@ mod tests { assert_eq!(jobs.jobs.len(), 1); } + #[test] + fn approval_shows_exact_argv_and_the_automatic_port_phrase() { + let mut w = ServeWizardState { + model: "qwen".into(), + ..Default::default() + }; + request_launch(&mut w); + let pending = w.approval.as_ref().unwrap(); + assert_eq!(pending.args, vec!["serve", "qwen"]); + let body = pending.request.body.join("\n"); + assert!(body.contains("serve qwen"), "{body}"); + assert!(body.contains(AUTO_PORT_NOTE), "{body}"); + } + + #[test] + fn approval_omits_the_automatic_phrase_for_a_custom_port() { + let mut w = ServeWizardState { + model: "qwen".into(), + port_mode: PortMode::Custom, + port: "8000".into(), + ..Default::default() + }; + request_launch(&mut w); + let pending = w.approval.as_ref().unwrap(); + assert!(pending.args.windows(2).any(|p| p == ["--port", "8000"])); + assert!(!pending.request.body.join("\n").contains(AUTO_PORT_NOTE)); + } + #[test] fn empty_model_launch_sets_message_not_approval() { let mut wiz = Some(ServeWizardState::default()); let mut jobs = State::default(); - let launch_idx = FIELDS.iter().position(|f| *f == Field::Launch).unwrap(); - wiz.as_mut().unwrap().field = launch_idx; + wiz.as_mut().unwrap().focus(Field::Launch); let fx = on_key(&mut wiz, &mut jobs, &[], key(KeyCode::Enter)); assert!(fx.is_empty()); let w = wiz.as_ref().unwrap(); assert!(w.approval.is_none()); assert_eq!(w.message.as_deref(), Some("model is required")); + assert_eq!( + w.current_field(), + Field::Model, + "focus lands on the problem" + ); } #[test] @@ -698,7 +1711,7 @@ mod tests { let mut wiz = Some(ServeWizardState::default()); let mut jobs = State::default(); wiz.as_mut().unwrap().model = "m".into(); - wiz.as_mut().unwrap().field = FIELDS.iter().position(|f| *f == Field::Launch).unwrap(); + wiz.as_mut().unwrap().focus(Field::Launch); on_key(&mut wiz, &mut jobs, &[], key(KeyCode::Enter)); let fx = on_key(&mut wiz, &mut jobs, &[], key(KeyCode::Char('n'))); assert!(fx.is_empty()); @@ -719,7 +1732,7 @@ mod tests { let mut wiz = Some(ServeWizardState::default()); let mut jobs = State::default(); wiz.as_mut().unwrap().model = "m".into(); - wiz.as_mut().unwrap().field = FIELDS.iter().position(|f| *f == Field::Launch).unwrap(); + wiz.as_mut().unwrap().focus(Field::Launch); on_key(&mut wiz, &mut jobs, &[], key(KeyCode::Enter)); on_key(&mut wiz, &mut jobs, &[], key(KeyCode::Char('y'))); assert!(wiz.as_ref().unwrap().active_job.is_some()); @@ -734,7 +1747,7 @@ mod tests { let mut wiz = Some(ServeWizardState::default()); let mut jobs = State::default(); wiz.as_mut().unwrap().model = "m".into(); - wiz.as_mut().unwrap().field = FIELDS.iter().position(|f| *f == Field::Launch).unwrap(); + wiz.as_mut().unwrap().focus(Field::Launch); on_key(&mut wiz, &mut jobs, &[], key(KeyCode::Enter)); on_key(&mut wiz, &mut jobs, &[], key(KeyCode::Char('y'))); assert!(wiz.as_ref().unwrap().active_job.is_some()); @@ -745,7 +1758,7 @@ mod tests { let mut wiz = Some(ServeWizardState::default()); let mut jobs = State::default(); wiz.as_mut().unwrap().model = "m".into(); - wiz.as_mut().unwrap().field = FIELDS.iter().position(|f| *f == Field::Launch).unwrap(); + wiz.as_mut().unwrap().focus(Field::Launch); on_key(&mut wiz, &mut jobs, &[], key(KeyCode::Enter)); on_key(&mut wiz, &mut jobs, &[], key(KeyCode::Char('y'))); let job_id = wiz.as_ref().unwrap().active_job.clone().unwrap(); @@ -767,13 +1780,12 @@ mod tests { // NOT claim success (set active_job) when no SpawnJob was emitted. let mut wiz = Some(ServeWizardState::default()); let mut jobs = State::default(); - let launch = FIELDS.iter().position(|f| *f == Field::Launch).unwrap(); // First launch of "qwen": real spawn. for k in typed("qwen") { on_key(&mut wiz, &mut jobs, &[], k); } - wiz.as_mut().unwrap().field = launch; + wiz.as_mut().unwrap().focus(Field::Launch); on_key(&mut wiz, &mut jobs, &[], key(KeyCode::Enter)); on_key(&mut wiz, &mut jobs, &[], key(KeyCode::Char('y'))); assert_eq!( @@ -787,7 +1799,7 @@ mod tests { for k in typed("qwen") { on_key(&mut wiz2, &mut jobs, &[], k); } - wiz2.as_mut().unwrap().field = launch; + wiz2.as_mut().unwrap().focus(Field::Launch); on_key(&mut wiz2, &mut jobs, &[], key(KeyCode::Enter)); let fx = on_key(&mut wiz2, &mut jobs, &[], key(KeyCode::Char('y'))); // No new SpawnJob, no stale console, an informative message instead. @@ -803,6 +1815,8 @@ mod tests { assert_eq!(jobs.jobs.len(), 1, "still just the one job"); } + // ------------------------------------------------------- picker / browser + #[test] fn tab_on_model_opens_folder_browser() { let mut wiz = Some(ServeWizardState::default()); @@ -816,18 +1830,7 @@ mod tests { } #[test] - fn left_right_cycles_engine() { - let mut wiz = Some(ServeWizardState::default()); - let mut jobs = State::default(); - wiz.as_mut().unwrap().field = FIELDS.iter().position(|f| *f == Field::Engine).unwrap(); - on_key(&mut wiz, &mut jobs, &[], key(KeyCode::Right)); - assert_eq!(wiz.as_ref().unwrap().engine_idx, 1); - on_key(&mut wiz, &mut jobs, &[], key(KeyCode::Left)); - assert_eq!(wiz.as_ref().unwrap().engine_idx, 0); - } - - #[test] - fn enter_on_model_opens_picker_and_choice_fills_model_and_engine() { + fn enter_on_model_opens_picker_and_choice_fills_model_only() { let recipes = vec![ModelRecipeSummary { id: "GLM-4".into(), aliases: vec!["glm".into()], @@ -836,18 +1839,17 @@ mod tests { }]; let mut wiz = Some(ServeWizardState::default()); // field 0 = Model let mut jobs = State::default(); - // Enter on Model opens the picker when recipes exist. on_key(&mut wiz, &mut jobs, &recipes, key(KeyCode::Enter)); assert!(wiz.as_ref().unwrap().picker.is_some()); - // Enter in the picker chooses the (only) recipe → fills model + engine. on_key(&mut wiz, &mut jobs, &recipes, key(KeyCode::Enter)); let w = wiz.as_ref().unwrap(); assert!(w.picker.is_none()); assert_eq!(w.model, "GLM-4"); assert_eq!( - ENGINES[w.engine_idx], "vllm", - "preferred engine pre-selected" + w.engine_idx, ENGINE_AUTO, + "a recipe never forces an engine override — the CLI resolves it" ); + assert_eq!(w.build_args().unwrap(), vec!["serve", "GLM-4"]); } #[test] @@ -856,29 +1858,10 @@ mod tests { let mut jobs = State::default(); on_key(&mut wiz, &mut jobs, &[], key(KeyCode::Enter)); assert!(wiz.as_ref().unwrap().picker.is_none()); - assert_eq!(wiz.as_ref().unwrap().field, 1, "Enter advances to Engine"); - } - - #[test] - fn recipe_with_unknown_preferred_engine_leaves_engine_idx() { - // Pins the silent-fallback contract: a preferred_engine the wizard does - // not list must NOT crash and must leave the engine choice untouched - // (model still filled). Guards future ENGINES vs rocm-core divergence. - let recipes = vec![ModelRecipeSummary { - id: "some-model".into(), - aliases: vec![], - task: "chat".into(), - preferred_engine: Some("not-in-engines-list".into()), - }]; - let mut wiz = Some(ServeWizardState::default()); // engine_idx 0 = lemonade - let mut jobs = State::default(); - on_key(&mut wiz, &mut jobs, &recipes, key(KeyCode::Enter)); // open picker - on_key(&mut wiz, &mut jobs, &recipes, key(KeyCode::Enter)); // choose first - let w = wiz.as_ref().unwrap(); - assert_eq!(w.model, "some-model"); assert_eq!( - w.engine_idx, 0, - "unknown preferred engine leaves the choice" + wiz.as_ref().unwrap().current_field(), + Field::Advanced, + "Enter advances to the disclosure row" ); } @@ -900,7 +1883,6 @@ mod tests { ]; let mut wiz = Some(ServeWizardState::default()); let mut jobs = State::default(); - // Type "qwen" on the Model field, then Enter to open the picker. for k in typed("qwen") { on_key(&mut wiz, &mut jobs, &recipes, k); } @@ -908,7 +1890,6 @@ mod tests { let picker = wiz.as_ref().unwrap().picker.as_ref().unwrap(); assert_eq!(picker.query, "qwen"); assert_eq!(picker.filtered(&recipes).len(), 1, "pre-narrowed to Qwen"); - // Enter chooses the single match. on_key(&mut wiz, &mut jobs, &recipes, key(KeyCode::Enter)); assert_eq!(wiz.as_ref().unwrap().model, "Qwen3-4B"); } @@ -931,6 +1912,8 @@ mod tests { assert!(wiz.is_some(), "picker Esc keeps the wizard open"); } + // ---------------------------------------------------------------- render + fn render(w: &ServeWizardState, jobs: &State) -> String { use ratatui::Terminal; use ratatui::backend::TestBackend; @@ -946,14 +1929,98 @@ mod tests { .collect() } + /// A rendered form row starts with the 2-cell focus marker and an + /// 8-cell-padded label, so `Mode` is `"Mode "` and can never be matched + /// by the `Model` row. Bare substrings are not precise enough here. + fn row_label(label: &str) -> String { + format!("{label:<8}") + } + #[test] - fn snapshot_renders_form_fields() { + fn collapsed_render_shows_only_model_advanced_and_launch() { let w = ServeWizardState::default(); let out = render(&w, &State::default()); assert!(out.contains("Serve a model"), "titled overlay"); - assert!(out.contains("Model"), "model field"); - assert!(out.contains("lemonade"), "default engine shown"); - assert!(out.contains("Launch"), "launch action"); + assert!(out.contains(&row_label("Model")), "model row"); + assert!(out.contains("Advanced settings ▸"), "collapsed disclosure"); + assert!(out.contains("Automatic"), "automatic summary"); + assert!(out.contains("[ Launch ]"), "launch action"); + for hidden in ["Engine", "Device", "Host", "Port", "Mode"] { + assert!( + !out.contains(&row_label(hidden)), + "{hidden} row must stay hidden: {out:?}" + ); + } + } + + #[test] + fn expanded_render_inlines_the_advanced_rows_without_a_second_modal() { + let w = ServeWizardState { + advanced_expanded: true, + ..Default::default() + }; + let out = render(&w, &State::default()); + assert!(out.contains("Advanced settings ▾"), "expanded disclosure"); + for shown in ["Model", "Engine", "Device", "Host", "Port", "Mode"] { + assert!( + out.contains(&row_label(shown)), + "{shown} row missing: {out:?}" + ); + } + assert!(out.contains("[ Launch ]"), "launch action"); + // Role affordances: chevrons for choices, brackets for editable text, + // plain read-only text for the GPU-required policy. + assert!(out.contains("‹ automatic ›"), "choice chevrons: {out:?}"); + assert!(out.contains("[127.0.0.1]"), "editable brackets: {out:?}"); + assert!( + out.contains(DEVICE_POLICY), + "read-only device text: {out:?}" + ); + assert!(!out.contains("Review:"), "no second modal"); + } + + #[test] + fn collapsed_render_reports_a_broken_hidden_port() { + let w = ServeWizardState { + model: "m".into(), + port_mode: PortMode::Custom, + port: "abc".into(), + ..Default::default() + }; + let out = render(&w, &State::default()); + assert!(out.contains("port needs attention"), "{out:?}"); + } + + #[test] + fn footer_names_the_role_of_the_focused_row() { + let mut w = ServeWizardState { + advanced_expanded: true, + ..Default::default() + }; + w.focus(Field::Engine); + assert!(render(&w, &State::default()).contains("choice")); + w.focus(Field::Device); + assert!(render(&w, &State::default()).contains("read only")); + w.focus(Field::Host); + assert!(render(&w, &State::default()).contains("editable")); + // Port names the direction it would move in, not a generic pair. + w.focus(Field::Port); + assert!(render(&w, &State::default()).contains("choice · ←→ switch to custom")); + w.port_mode = PortMode::Custom; + assert!( + render(&w, &State::default()) + .contains("choice · ←→ back to automatic · Enter to edit the port") + ); + w.focus(Field::Host); + let host = w.host.clone(); + w.editor = Some(InlineEditor::new(Field::Host, &host)); + assert!(render(&w, &State::default()).contains("Esc cancel")); + } + + #[test] + fn no_generic_change_footer_remains() { + let w = ServeWizardState::default(); + assert!(!render(&w, &State::default()).contains("←→ change")); } #[test] @@ -961,11 +2028,12 @@ mod tests { let mut wiz = Some(ServeWizardState::default()); let mut jobs = State::default(); wiz.as_mut().unwrap().model = "qwen".into(); - wiz.as_mut().unwrap().field = FIELDS.iter().position(|f| *f == Field::Launch).unwrap(); + wiz.as_mut().unwrap().focus(Field::Launch); on_key(&mut wiz, &mut jobs, &[], key(KeyCode::Enter)); let out = render(wiz.as_ref().unwrap(), &jobs); assert!(out.contains("Review:"), "approval modal shown"); assert!(out.contains("serve"), "describes the gated launch"); assert!(out.contains("Approve"), "approve button present"); + assert!(out.contains("endpoint shown after launch"), "{out:?}"); } } diff --git a/crates/rocm-dash-tui/src/ui/tabs/serving.rs b/crates/rocm-dash-tui/src/ui/tabs/serving.rs index d3afaceb..888c70fe 100644 --- a/crates/rocm-dash-tui/src/ui/tabs/serving.rs +++ b/crates/rocm-dash-tui/src/ui/tabs/serving.rs @@ -25,11 +25,11 @@ pub const VERBS: &[Verb] = &[ summary: "Launch a model on a serving engine and expose an OpenAI-style endpoint.", steps: &[ "Pick a model", - "Choose an engine — Lemonade · vLLM", - "Set GPU placement (required / preferred / CPU-only)", - "Launch on 127.0.0.1:11435 and watch it come up", + "Leave the rest automatic, or open Advanced settings", + "GPU is required — ROCm never falls back to CPU", + "Launch on a local endpoint picked for you and watch it come up", ], - cmd: "rocm serve --engine …", + cmd: "rocm serve ", read_only: false, badge: None, }, @@ -155,6 +155,28 @@ mod tests { .collect() } + #[test] + fn serving_first_view_promises_no_concrete_endpoint() { + // The CLI leases the local port at launch time, so the first thing a + // user reads must not promise a fixed `127.0.0.1:11435` endpoint. + let mut s = AppState::new("t".into(), "default-dark".into()); + s.active_tab = ActiveTab::Serving; + s.serving_sel = 0; // "Serve a model" + let out = render(&s, 200, 30); + assert!( + !out.contains("11435"), + "first-view copy must not promise a fixed port: {out:?}" + ); + assert!( + !out.contains("127.0.0.1"), + "first-view copy must not promise a fixed host: {out:?}" + ); + assert!( + out.contains("Launch on a local endpoint picked for you"), + "automatic endpoint wording missing: {out:?}" + ); + } + #[test] fn serving_renders_its_six_rows() { let mut s = AppState::new("t".into(), "default-dark".into()); diff --git a/tests/e2e-cucumber/features/model_serving.feature b/tests/e2e-cucumber/features/model_serving.feature index a9edaab9..41c52a65 100644 --- a/tests/e2e-cucumber/features/model_serving.feature +++ b/tests/e2e-cucumber/features/model_serving.feature @@ -29,6 +29,17 @@ Feature: Model serving When the user lists running services Then the connection details match the actual server port + # Real CLI coverage for automatic port allocation. A loopback listener holds + # the legacy default while one real managed GPU serve starts; the reported + # endpoint must advance to the next candidate rather than failing late in the + # engine. Runs in the merge queue because it launches a real model server. + @id:serve-auto-port-skips-occupied @requires-gpu @merge-queue + Scenario: 15 - Automatic serving skips an occupied default port + Given a managed runtime is active + And the default serve port is occupied + When the user serves a model with automatic port selection + Then the service uses the next automatic port + # vLLM serve + inference (safetensors model). Engine coverage: vLLM. This is the # deliberate vLLM half of a per-engine pair with `serve-lemonade-inference` # below, so it stays pinned to vLLM (the slug names the engine). It is also the diff --git a/tests/e2e-cucumber/tests/e2e.rs b/tests/e2e-cucumber/tests/e2e.rs index ea7ea3d1..7b3c0314 100644 --- a/tests/e2e-cucumber/tests/e2e.rs +++ b/tests/e2e-cucumber/tests/e2e.rs @@ -38,6 +38,9 @@ pub struct E2eWorld { /// Loopback file server used by artifact-prefetch scenarios. Kept on the /// World so it remains alive while the real `rocmd` subprocess downloads. pub artifact_server: Option, + /// Listener that deliberately occupies the default serve port while a real + /// `rocm serve` invocation proves automatic allocation advances past it. + pub automatic_port_guard: Option, /// Cache-marker destination discovered from `rocmd`'s own JSON report. pub artifact_marker_path: Option, pub endpoint: Option, @@ -174,6 +177,7 @@ impl Default for E2eWorld { Self { mock: None, artifact_server: None, + automatic_port_guard: None, artifact_marker_path: None, endpoint: None, model_name: None, diff --git a/tests/e2e-cucumber/tests/e2e/serving_steps.rs b/tests/e2e-cucumber/tests/e2e/serving_steps.rs index ca7b9da3..7bcca19d 100644 --- a/tests/e2e-cucumber/tests/e2e/serving_steps.rs +++ b/tests/e2e-cucumber/tests/e2e/serving_steps.rs @@ -308,6 +308,18 @@ async fn setup_mock_custom_port(world: &mut E2eWorld) { world.mock = Some(mock); } +#[given("the default serve port is occupied")] +async fn occupy_default_serve_port(world: &mut E2eWorld) { + let _ = ensure_serve_port_free().await; + let next = std::net::TcpListener::bind(("127.0.0.1", SERVE_PORT + 1)) + .expect("the next automatic port must be free before this scenario starts"); + drop(next); + world.automatic_port_guard = Some( + std::net::TcpListener::bind(("127.0.0.1", SERVE_PORT)) + .expect("failed to occupy the default serve port"), + ); +} + /// The (model, engine, ready-substring) this host should serve for an /// engine-agnostic "serve a real model" precondition. /// @@ -698,6 +710,17 @@ async fn user_serves_vllm_capable_default(world: &mut E2eWorld) { world.cli_rc = Some(rc); } +#[when("the user serves a model with automatic port selection")] +async fn user_serves_with_automatic_port(world: &mut E2eWorld) { + let (model, engine, _) = host_serve_target(); + let (stdout, stderr, rc) = + crate::run_rocm(world, &["serve", model, "--engine", engine, "--managed"]); + world.automatic_port_guard.take(); + world.cli_output = Some(stdout); + world.cli_stderr = Some(stderr); + world.cli_rc = Some(rc); +} + #[when("the user sends a chat completion request")] async fn user_sends_completion(world: &mut E2eWorld) { crate::send_chat(world).await; @@ -880,6 +903,28 @@ async fn assert_endpoint_port(world: &mut E2eWorld) { ); } +#[then("the service uses the next automatic port")] +async fn assert_next_automatic_port(world: &mut E2eWorld) { + let stdout = world.cli_output.as_deref().unwrap_or(""); + let stderr = world.cli_stderr.as_deref().unwrap_or(""); + let rc = world.cli_rc.expect("no serve command was run"); + assert!( + rc == 0, + "{}", + e2e_cucumber::cli_failure_report( + &["serve", "", "--engine", "", "--managed"], + rc, + stdout, + stderr, + ) + ); + let expected = format!("endpoint: http://127.0.0.1:{}/v1", SERVE_PORT + 1); + assert!( + stdout.lines().any(|line| line.trim() == expected), + "expected `{expected}` in real serve output:\n{stdout}" + ); +} + /// Extract the engine name from a serve plan's `engine: ` line. fn selected_engine(output: &str) -> &str { let Some(engine) = output