From d48861a2ea018e5d8166067d0c7ae234d111f37e Mon Sep 17 00:00:00 2001 From: IvGolovach <20299097+IvGolovach@users.noreply.github.com> Date: Fri, 29 May 2026 11:49:37 -0700 Subject: [PATCH] Upgrade demanded local models during reconciliation Validation * Validation tier: Tier 3 - opt-in runtime reconciliation can replace a lower-demand local model with a locally present, fresh active-demand target; config admission, planner policy, runtime unload/load sequencing, API target signals, and docs are refreshed on current main after PR #753 landed. * git fetch --no-tags origin main:refs/remotes/origin/main: PASS, origin/main at c0f8990bb17b887f4a1fbaa00d6224c9af086f8d. * git rebase origin/main: PASS after resolving one test-module conflict in crates/mesh-llm-host-runtime/src/runtime/mod.rs by preserving both the newly landed mDNS relay-policy tests and this PR's model-target replacement sequencing test. * git diff --check origin/main...HEAD: PASS, no output. * git diff --check: PASS, no output. * git diff --cached --check: PASS, no output. * cargo fmt --all -- --check: PASS. * cargo test -p mesh-llm-config --lib -- --test-threads=1: PASS, 10 passed. * LLAMA_STAGE_BUILD_DIR=/Users/Funtland/Downloads/mesh-llm/.deps/llama-build/build-stage-abi-metal cargo test -p mesh-llm-host-runtime model_target_reconciliation --lib -- --test-threads=1: PASS, 22 passed. * LLAMA_STAGE_BUILD_DIR=/Users/Funtland/Downloads/mesh-llm/.deps/llama-build/build-stage-abi-metal cargo test -p mesh-llm-host-runtime api::model_targets --lib -- --test-threads=1: PASS, 3 passed. * LLAMA_STAGE_BUILD_DIR=/Users/Funtland/Downloads/mesh-llm/.deps/llama-build/build-stage-abi-metal cargo test -p mesh-llm-host-runtime mdns_discovery --lib -- --test-threads=1: PASS, 3 passed. * LLAMA_STAGE_BUILD_DIR=/Users/Funtland/Downloads/mesh-llm/.deps/llama-build/build-stage-abi-metal cargo check -p mesh-llm: PASS. * LLAMA_STAGE_BUILD_DIR=/Users/Funtland/Downloads/mesh-llm/.deps/llama-build/build-stage-abi-metal /opt/homebrew/bin/cargo-clippy clippy -p mesh-llm-config -p mesh-llm-host-runtime --all-targets -- -D warnings: PASS. * Ledger: not applicable - not required for selected validation tier/change family. * Version: not applicable - no release/version sync required for this non-release opt-in runtime reconciliation change. * Not run: live multi-node demand/reconciliation smoke - no local multi-node runtime endpoint and local GGUF set were available; config, model-target signal, planner, runtime-control, and conflict-adjacent mDNS tests cover the changed branches. * Not run: full workspace suite locally - mandatory PR CI is the final full-suite proof for the pushed SHA. Rollback * git revert HEAD --- ROADMAP.md | 2 +- crates/mesh-llm-config/src/lib.rs | 6 + crates/mesh-llm-config/src/model.rs | 32 ++- .../mesh-llm-host-runtime/src/runtime/mod.rs | 179 +++++++++++--- .../runtime/model_target_reconciliation.rs | 219 +++++++++++++++++- docs/design/DESIGN.md | 10 +- docs/design/TESTING.md | 2 +- 7 files changed, 411 insertions(+), 39 deletions(-) diff --git a/ROADMAP.md b/ROADMAP.md index ae81b4c5e..ce52e7df0 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -51,7 +51,7 @@ MTP work with llama.cpp ongoing, but should be part of this to accelerate infere ## Demand-based rebalancing -Partially done. Unified demand map via gossip, standby nodes promote to serve. Next: large-VRAM hosts auto-upgrade models when demand warrants it. +Partially done. Unified demand map via gossip, standby nodes promote to serve, and large-VRAM hosts can opt into fresh active-demand upgrades for local artifacts. Next: download-backed upgrades, split-aware upgrades, and replica-count balancing. ## Blackboard ✅ diff --git a/crates/mesh-llm-config/src/lib.rs b/crates/mesh-llm-config/src/lib.rs index 711785d07..38f70120f 100644 --- a/crates/mesh-llm-config/src/lib.rs +++ b/crates/mesh-llm-config/src/lib.rs @@ -201,11 +201,17 @@ version = 1 [runtime] reconcile_model_targets = true +reconcile_model_target_demand_upgrades = true +model_target_demand_upgrade_min_requests = 4 +model_target_demand_upgrade_max_age_secs = 900 "#, ) .unwrap(); assert!(config.runtime.reconcile_model_targets); + assert!(config.runtime.reconcile_model_target_demand_upgrades); + assert_eq!(config.runtime.model_target_demand_upgrade_min_requests, 4); + assert_eq!(config.runtime.model_target_demand_upgrade_max_age_secs, 900); } #[test] diff --git a/crates/mesh-llm-config/src/model.rs b/crates/mesh-llm-config/src/model.rs index b0d96acfc..55629852a 100644 --- a/crates/mesh-llm-config/src/model.rs +++ b/crates/mesh-llm-config/src/model.rs @@ -44,10 +44,40 @@ pub struct GpuConfig { pub parallel: Option, } -#[derive(Clone, Debug, Default, Deserialize, Serialize, PartialEq, Eq)] +pub const DEFAULT_MODEL_TARGET_DEMAND_UPGRADE_MIN_REQUESTS: u64 = 2; +pub const DEFAULT_MODEL_TARGET_DEMAND_UPGRADE_MAX_AGE_SECS: u64 = 60 * 60; + +fn default_model_target_demand_upgrade_min_requests() -> u64 { + DEFAULT_MODEL_TARGET_DEMAND_UPGRADE_MIN_REQUESTS +} + +fn default_model_target_demand_upgrade_max_age_secs() -> u64 { + DEFAULT_MODEL_TARGET_DEMAND_UPGRADE_MAX_AGE_SECS +} + +#[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)] pub struct RuntimeConfig { #[serde(default)] pub reconcile_model_targets: bool, + #[serde(default)] + pub reconcile_model_target_demand_upgrades: bool, + #[serde(default = "default_model_target_demand_upgrade_min_requests")] + pub model_target_demand_upgrade_min_requests: u64, + #[serde(default = "default_model_target_demand_upgrade_max_age_secs")] + pub model_target_demand_upgrade_max_age_secs: u64, +} + +impl Default for RuntimeConfig { + fn default() -> Self { + Self { + reconcile_model_targets: false, + reconcile_model_target_demand_upgrades: false, + model_target_demand_upgrade_min_requests: + DEFAULT_MODEL_TARGET_DEMAND_UPGRADE_MIN_REQUESTS, + model_target_demand_upgrade_max_age_secs: + DEFAULT_MODEL_TARGET_DEMAND_UPGRADE_MAX_AGE_SECS, + } + } } #[derive(Clone, Debug, Default, Deserialize, Serialize, PartialEq, Eq)] diff --git a/crates/mesh-llm-host-runtime/src/runtime/mod.rs b/crates/mesh-llm-host-runtime/src/runtime/mod.rs index ade734fd1..94ca63b45 100644 --- a/crates/mesh-llm-host-runtime/src/runtime/mod.rs +++ b/crates/mesh-llm-host-runtime/src/runtime/mod.rs @@ -31,9 +31,10 @@ use self::local::{ stop_split_generation_cleanup, withdraw_advertised_model, }; use self::model_target_reconciliation::{ - ModelTargetReconciliationCandidate, ModelTargetReconciliationCapacityState, - ModelTargetReconciliationInput, ModelTargetReconciliationPolicy, - ModelTargetReconciliationState, plan_model_target_reconciliation, + ModelTargetReconciliationAction, ModelTargetReconciliationCandidate, + ModelTargetReconciliationCapacityState, ModelTargetReconciliationInput, + ModelTargetReconciliationPolicy, ModelTargetReconciliationState, + plan_model_target_reconciliation, }; use self::proxy::{api_proxy, bootstrap_proxy}; #[cfg(test)] @@ -3496,6 +3497,9 @@ fn model_target_reconciliation_policy( ) -> ModelTargetReconciliationPolicy { ModelTargetReconciliationPolicy { enabled: config.runtime.reconcile_model_targets, + demand_upgrades_enabled: config.runtime.reconcile_model_target_demand_upgrades, + demand_upgrade_min_request_count: config.runtime.model_target_demand_upgrade_min_requests, + demand_upgrade_max_age_secs: config.runtime.model_target_demand_upgrade_max_age_secs, ..ModelTargetReconciliationPolicy::default() } } @@ -3533,21 +3537,26 @@ async fn reconcile_model_targets_once(ctx: ReconcileModelTargetsContext<'_>) { .await .into_iter() .collect::>(); - if local_interest_model_refs.is_empty() { + let loaded_model_refs = runtime_loaded_model_refs(runtime_models, managed_models); + if local_interest_model_refs.is_empty() && loaded_model_refs.is_empty() { state.prune_expired(runtime_unix_secs()); return; } let target_lookup = console_state.model_target_lookup().await; - let loaded_model_refs = runtime_loaded_model_refs(runtime_models, managed_models); let local_vram_bytes = node.vram_bytes(); let targets = target_lookup .targets .into_iter() .map(|target| { + let demand_upgrade_target = model_target_reconciliation_demand_upgrade_candidate( + policy, + &loaded_model_refs, + &target, + ); let local_path = if target.wanted && target.serving_node_count == 0 - && local_interest_model_refs.contains(&target.model_ref) + && (local_interest_model_refs.contains(&target.model_ref) || demand_upgrade_target) && target.capacity_advice.state == api::status::ModelTargetCapacityAdviceState::SingleNodeFit && model_target_reconciliation_local_fit(&target, local_vram_bytes) @@ -3557,9 +3566,13 @@ async fn reconcile_model_targets_once(ctx: ReconcileModelTargetsContext<'_>) { None }; ModelTargetReconciliationCandidate { + rank: target.rank, model_ref: target.model_ref, model_name: target.model_name, wanted: target.wanted, + wanted_reason: target.wanted_reason, + request_count: target.request_count, + last_active_secs_ago: target.last_active_secs_ago, serving_node_count: target.serving_node_count, capacity_state: ModelTargetReconciliationCapacityState::from( target.capacity_advice.state, @@ -3584,43 +3597,84 @@ async fn reconcile_model_targets_once(ctx: ReconcileModelTargetsContext<'_>) { for action in actions { let load_spec = action.load_spec.to_string_lossy().to_string(); - let (resp_tx, resp_rx) = tokio::sync::oneshot::channel(); - if control_tx - .send(api::RuntimeControlRequest::Load { - spec: load_spec.clone(), - resp: resp_tx, - }) - .is_err() - { - state.record_load_failure(&action.model_ref, now_secs, policy); - let _ = emit_event(OutputEvent::Warning { - message: format!( - "Model target reconciliation could not queue '{}'", - action.model_ref - ), - context: Some("runtime control channel closed".to_string()), - }); - continue; - } - state.mark_load_started(&action.model_ref); let event_tx = runtime_event_tx.clone(); let model_ref = action.model_ref.clone(); + let control_tx = control_tx.clone(); + let replace_model_ref = action.replace_model_ref.clone(); tokio::spawn(async move { - let result = match resp_rx.await { - Ok(result) => result.map_err(|err| err.to_string()), - Err(err) => Err(format!("runtime load response channel closed: {err}")), - }; + let result = + run_model_target_reconciliation_action(control_tx, load_spec, replace_model_ref) + .await; let _ = event_tx .send(RuntimeEvent::ModelTargetReconciliationLoadFinished { model_ref, result }); }); - let _ = emit_event(OutputEvent::Info { - message: format!("Model target reconciliation loading '{}'", action.model_ref), - context: Some(format!("path={load_spec}")), - }); + emit_model_target_reconciliation_queued(&action); } } +async fn run_model_target_reconciliation_action( + control_tx: tokio::sync::mpsc::UnboundedSender, + load_spec: String, + replace_model_ref: Option, +) -> std::result::Result { + if let Some(replace_model_ref) = replace_model_ref { + run_model_target_reconciliation_unload(control_tx.clone(), replace_model_ref).await?; + } + run_model_target_reconciliation_load(control_tx, load_spec).await +} + +async fn run_model_target_reconciliation_unload( + control_tx: tokio::sync::mpsc::UnboundedSender, + model_ref: String, +) -> std::result::Result { + let (resp, response) = tokio::sync::oneshot::channel(); + control_tx + .send(api::RuntimeControlRequest::Unload { + target: UnloadTarget::Model(model_ref.clone()), + options: UnloadOptions::default(), + resp, + }) + .map_err(|_| format!("runtime unload queue closed for replacement target '{model_ref}'"))?; + response + .await + .map_err(|err| format!("runtime unload response channel closed: {err}"))? + .map_err(|err| err.to_string()) +} + +async fn run_model_target_reconciliation_load( + control_tx: tokio::sync::mpsc::UnboundedSender, + load_spec: String, +) -> std::result::Result { + let (resp, response) = tokio::sync::oneshot::channel(); + control_tx + .send(api::RuntimeControlRequest::Load { + spec: load_spec.clone(), + resp, + }) + .map_err(|_| format!("runtime load queue closed for '{load_spec}'"))?; + response + .await + .map_err(|err| format!("runtime load response channel closed: {err}"))? + .map_err(|err| err.to_string()) +} + +fn emit_model_target_reconciliation_queued(action: &ModelTargetReconciliationAction) { + let context = match action.replace_model_ref.as_deref() { + Some(replace_model_ref) => Some(format!("replace={replace_model_ref}")), + None => Some(format!("path={}", action.load_spec.display())), + }; + let verb = if action.replace_model_ref.is_some() { + "upgrading to" + } else { + "loading" + }; + let _ = emit_event(OutputEvent::Info { + message: format!("Model target reconciliation {verb} '{}'", action.model_ref), + context, + }); +} + fn runtime_loaded_model_refs( runtime_models: &HashMap, managed_models: &HashMap, @@ -3659,6 +3713,20 @@ fn model_target_reconciliation_local_fit( .is_some_and(|required| local_vram_bytes >= required) } +fn model_target_reconciliation_demand_upgrade_candidate( + policy: &ModelTargetReconciliationPolicy, + loaded_model_refs: &BTreeSet, + target: &api::status::ModelTargetPayload, +) -> bool { + policy.demand_upgrades_enabled + && !loaded_model_refs.is_empty() + && target.wanted_reason == Some("active_demand") + && target.request_count >= policy.demand_upgrade_min_request_count + && target + .last_active_secs_ago + .is_some_and(|age| age <= policy.demand_upgrade_max_age_secs) +} + fn runtime_unix_secs() -> u64 { SystemTime::now() .duration_since(UNIX_EPOCH) @@ -8707,6 +8775,51 @@ mod tests { )); } + #[tokio::test] + async fn model_target_reconciliation_replacement_unloads_before_loading() { + let (control_tx, mut control_rx) = + tokio::sync::mpsc::unbounded_channel::(); + let task = tokio::spawn(run_model_target_reconciliation_action( + control_tx, + "/models/large.gguf".to_string(), + Some("Small".to_string()), + )); + + match control_rx.recv().await { + Some(api::RuntimeControlRequest::Unload { target, resp, .. }) => { + assert_eq!(target.as_runtime_target(), "Small"); + resp.send(Ok(api::RuntimeUnloadResponse { + model: "Small".to_string(), + instance_id: "runtime-1".to_string(), + unloaded: true, + })) + .expect("replacement unload response should be received"); + } + _ => panic!("expected unload request before load"), + } + match control_rx.recv().await { + Some(api::RuntimeControlRequest::Load { spec, resp }) => { + assert_eq!(spec, "/models/large.gguf"); + resp.send(Ok(api::RuntimeLoadResponse { + model_ref: spec, + model: "Large".to_string(), + instance_id: "runtime-2".to_string(), + backend: Some("skippy".to_string()), + context_length: Some(4096), + })) + .expect("replacement load response should be received"); + } + _ => panic!("expected load request after unload"), + } + + let result = task + .await + .expect("replacement task should join") + .expect("replacement action should finish"); + assert_eq!(result.model, "Large"); + assert!(control_rx.try_recv().is_err()); + } + fn remote_catalog_layer_entry( variant_name: &str, curated_name: &str, diff --git a/crates/mesh-llm-host-runtime/src/runtime/model_target_reconciliation.rs b/crates/mesh-llm-host-runtime/src/runtime/model_target_reconciliation.rs index 7823400b3..5fe0b325f 100644 --- a/crates/mesh-llm-host-runtime/src/runtime/model_target_reconciliation.rs +++ b/crates/mesh-llm-host-runtime/src/runtime/model_target_reconciliation.rs @@ -9,6 +9,9 @@ pub(crate) struct ModelTargetReconciliationPolicy { pub(crate) max_loads_per_tick: usize, pub(crate) failure_cooldown_secs: u64, pub(crate) manual_unload_cooldown_secs: u64, + pub(crate) demand_upgrades_enabled: bool, + pub(crate) demand_upgrade_min_request_count: u64, + pub(crate) demand_upgrade_max_age_secs: u64, } impl Default for ModelTargetReconciliationPolicy { @@ -18,6 +21,11 @@ impl Default for ModelTargetReconciliationPolicy { max_loads_per_tick: 1, failure_cooldown_secs: 5 * 60, manual_unload_cooldown_secs: 5 * 60, + demand_upgrades_enabled: false, + demand_upgrade_min_request_count: + mesh_llm_config::DEFAULT_MODEL_TARGET_DEMAND_UPGRADE_MIN_REQUESTS, + demand_upgrade_max_age_secs: + mesh_llm_config::DEFAULT_MODEL_TARGET_DEMAND_UPGRADE_MAX_AGE_SECS, } } } @@ -114,9 +122,13 @@ pub(crate) struct ModelTargetReconciliationInput<'a> { #[derive(Clone, Debug, Eq, PartialEq)] pub(crate) struct ModelTargetReconciliationCandidate { + pub(crate) rank: usize, pub(crate) model_ref: String, pub(crate) model_name: Option, pub(crate) wanted: bool, + pub(crate) wanted_reason: Option<&'static str>, + pub(crate) request_count: u64, + pub(crate) last_active_secs_ago: Option, pub(crate) serving_node_count: usize, pub(crate) capacity_state: ModelTargetReconciliationCapacityState, pub(crate) local_path: Option, @@ -152,6 +164,7 @@ pub(crate) struct ModelTargetReconciliationAction { pub(crate) model_ref: String, pub(crate) model_name: Option, pub(crate) load_spec: PathBuf, + pub(crate) replace_model_ref: Option, } pub(crate) fn plan_model_target_reconciliation( @@ -175,10 +188,13 @@ pub(crate) fn plan_model_target_reconciliation( let Some(load_spec) = target.local_path.clone() else { continue; }; + let replace_model_ref = + replacement_target(policy, input.loaded_model_refs, input.targets, target); + let has_local_interest = input.local_interest_model_refs.contains(&target.model_ref); if !target.wanted || target.serving_node_count > 0 || target.capacity_state != ModelTargetReconciliationCapacityState::SingleNodeFit - || !input.local_interest_model_refs.contains(&target.model_ref) + || (!has_local_interest && replace_model_ref.is_none()) || loaded_target(input.loaded_model_refs, target) || state.suppressed( &target.model_ref, @@ -193,11 +209,58 @@ pub(crate) fn plan_model_target_reconciliation( model_ref: target.model_ref.clone(), model_name: target.model_name.clone(), load_spec, + replace_model_ref, }); } actions } +fn replacement_target( + policy: &ModelTargetReconciliationPolicy, + loaded_model_refs: &BTreeSet, + targets: &[ModelTargetReconciliationCandidate], + target: &ModelTargetReconciliationCandidate, +) -> Option { + if !demand_upgrade_candidate(policy, loaded_model_refs, target) { + return None; + } + loaded_model_refs + .iter() + .find(|loaded| replacement_improves_target_mix(loaded, targets, target)) + .cloned() +} + +fn demand_upgrade_candidate( + policy: &ModelTargetReconciliationPolicy, + loaded_model_refs: &BTreeSet, + target: &ModelTargetReconciliationCandidate, +) -> bool { + policy.demand_upgrades_enabled + && !loaded_model_refs.is_empty() + && target.wanted_reason == Some("active_demand") + && target.request_count >= policy.demand_upgrade_min_request_count + && target + .last_active_secs_ago + .is_some_and(|age| age <= policy.demand_upgrade_max_age_secs) +} + +fn replacement_improves_target_mix( + loaded_model_ref: &str, + targets: &[ModelTargetReconciliationCandidate], + target: &ModelTargetReconciliationCandidate, +) -> bool { + let Some(loaded) = targets + .iter() + .find(|candidate| model_target_matches_loaded(candidate, loaded_model_ref)) + else { + return true; + }; + if loaded.request_count >= target.request_count { + return false; + } + target.rank < loaded.rank || loaded.request_count == 0 +} + fn loaded_target( loaded_model_refs: &BTreeSet, target: &ModelTargetReconciliationCandidate, @@ -211,6 +274,17 @@ fn loaded_target( }) } +fn model_target_matches_loaded( + target: &ModelTargetReconciliationCandidate, + loaded_model_ref: &str, +) -> bool { + model_identity_matches(loaded_model_ref, &target.model_ref) + || target + .model_name + .as_deref() + .is_some_and(|name| model_identity_matches(loaded_model_ref, name)) +} + fn model_identity_matches(left: &str, right: &str) -> bool { if left == right { return true; @@ -243,11 +317,24 @@ mod tests { } } + fn demand_upgrade_policy() -> ModelTargetReconciliationPolicy { + ModelTargetReconciliationPolicy { + demand_upgrades_enabled: true, + demand_upgrade_min_request_count: 2, + demand_upgrade_max_age_secs: 60 * 60, + ..enabled_policy() + } + } + fn target(model_ref: &str) -> ModelTargetReconciliationCandidate { ModelTargetReconciliationCandidate { + rank: 1, model_ref: model_ref.to_string(), model_name: Some("Qwen3-8B-Q4_K_M".to_string()), wanted: true, + wanted_reason: Some("explicit_interest"), + request_count: 0, + last_active_secs_ago: None, serving_node_count: 0, capacity_state: ModelTargetReconciliationCapacityState::SingleNodeFit, local_path: Some(PathBuf::from("/models/qwen.gguf")), @@ -303,10 +390,140 @@ mod tests { model_ref: "org/model@main:file.gguf".to_string(), model_name: Some("Qwen3-8B-Q4_K_M".to_string()), load_spec: PathBuf::from("/models/qwen.gguf"), + replace_model_ref: None, }] ); } + #[test] + fn demand_upgrade_replaces_lower_demand_loaded_model() { + let mut wanted_large = target("org/large@main:file.gguf"); + wanted_large.rank = 1; + wanted_large.model_name = Some("Large".to_string()); + wanted_large.wanted_reason = Some("active_demand"); + wanted_large.request_count = 8; + wanted_large.last_active_secs_ago = Some(30); + wanted_large.local_path = Some(PathBuf::from("/models/large.gguf")); + let mut loaded_small = target("org/small@main:file.gguf"); + loaded_small.rank = 2; + loaded_small.model_name = Some("Small".to_string()); + loaded_small.wanted = false; + loaded_small.request_count = 1; + loaded_small.serving_node_count = 1; + loaded_small.capacity_state = ModelTargetReconciliationCapacityState::AlreadyServing; + loaded_small.local_path = None; + let targets = vec![wanted_large, loaded_small]; + let local_interests = BTreeSet::new(); + let loaded = BTreeSet::from(["Small".to_string()]); + let mut state = ModelTargetReconciliationState::default(); + + let actions = plan_model_target_reconciliation( + &demand_upgrade_policy(), + &mut state, + input(&local_interests, &loaded, &targets), + ); + + assert_eq!( + actions, + vec![ModelTargetReconciliationAction { + model_ref: "org/large@main:file.gguf".to_string(), + model_name: Some("Large".to_string()), + load_spec: PathBuf::from("/models/large.gguf"), + replace_model_ref: Some("Small".to_string()), + }] + ); + } + + #[test] + fn demand_upgrade_requires_explicit_policy_opt_in() { + let mut wanted_large = target("org/large@main:file.gguf"); + wanted_large.model_name = Some("Large".to_string()); + wanted_large.wanted_reason = Some("active_demand"); + wanted_large.request_count = 8; + wanted_large.last_active_secs_ago = Some(30); + let loaded = BTreeSet::from(["Small".to_string()]); + let targets = vec![wanted_large]; + let local_interests = BTreeSet::new(); + let mut state = ModelTargetReconciliationState::default(); + + let actions = plan_model_target_reconciliation( + &enabled_policy(), + &mut state, + input(&local_interests, &loaded, &targets), + ); + + assert!(actions.is_empty()); + } + + #[test] + fn stale_demand_does_not_replace_loaded_model() { + let mut wanted_large = target("org/large@main:file.gguf"); + wanted_large.model_name = Some("Large".to_string()); + wanted_large.wanted_reason = Some("active_demand"); + wanted_large.request_count = 8; + wanted_large.last_active_secs_ago = Some(2 * 60 * 60); + let loaded = BTreeSet::from(["Small".to_string()]); + let targets = vec![wanted_large]; + let local_interests = BTreeSet::new(); + let mut state = ModelTargetReconciliationState::default(); + + let actions = plan_model_target_reconciliation( + &demand_upgrade_policy(), + &mut state, + input(&local_interests, &loaded, &targets), + ); + + assert!(actions.is_empty()); + } + + #[test] + fn requested_only_target_does_not_replace_loaded_model_without_request_demand() { + let mut requested_only = target("org/requested@main:file.gguf"); + requested_only.request_count = 0; + let targets = vec![requested_only]; + let local_interests = BTreeSet::new(); + let loaded = BTreeSet::from(["Small".to_string()]); + let mut state = ModelTargetReconciliationState::default(); + + let actions = plan_model_target_reconciliation( + &demand_upgrade_policy(), + &mut state, + input(&local_interests, &loaded, &targets), + ); + + assert!(actions.is_empty()); + } + + #[test] + fn demand_upgrade_preserves_loaded_model_with_equal_or_higher_demand() { + let mut wanted_large = target("org/large@main:file.gguf"); + wanted_large.rank = 2; + wanted_large.model_name = Some("Large".to_string()); + wanted_large.wanted_reason = Some("active_demand"); + wanted_large.request_count = 3; + wanted_large.last_active_secs_ago = Some(30); + let mut loaded_hot = target("org/hot@main:file.gguf"); + loaded_hot.rank = 1; + loaded_hot.model_name = Some("Hot".to_string()); + loaded_hot.wanted = false; + loaded_hot.request_count = 3; + loaded_hot.serving_node_count = 1; + loaded_hot.capacity_state = ModelTargetReconciliationCapacityState::AlreadyServing; + loaded_hot.local_path = None; + let targets = vec![loaded_hot, wanted_large]; + let local_interests = BTreeSet::new(); + let loaded = BTreeSet::from(["Hot".to_string()]); + let mut state = ModelTargetReconciliationState::default(); + + let actions = plan_model_target_reconciliation( + &demand_upgrade_policy(), + &mut state, + input(&local_interests, &loaded, &targets), + ); + + assert!(actions.is_empty()); + } + #[test] fn skips_peer_only_or_requested_targets_without_local_interest() { let targets = vec![target("org/model@main:file.gguf")]; diff --git a/docs/design/DESIGN.md b/docs/design/DESIGN.md index 61d093f86..7b1a9530c 100644 --- a/docs/design/DESIGN.md +++ b/docs/design/DESIGN.md @@ -280,8 +280,14 @@ without the HTML via curl/scripts. Runtime reconciliation is opt-in. When `[runtime] reconcile_model_targets = true` is set, the local runtime may load an already-present local GGUF for a locally registered explicit interest, but only when `/api/model-targets` says the target -is wanted, unserved, and a single-node capacity fit for the current node. It -does not download models, start split serving, or act on peer-only interest. +is wanted, unserved, and a single-node capacity fit for the current node. A +host that also sets `reconcile_model_target_demand_upgrades = true` may replace +a less-demanded local model with a locally present, higher-ranked unserved +target once fresh active request demand crosses +`model_target_demand_upgrade_min_requests`. Stale demand older than +`model_target_demand_upgrade_max_age_secs` is advisory only. Runtime +reconciliation does not download models, start split serving, or act on +requested-only seed interest. `/api/model-targets` keeps raw inputs and computed hints separate. Each target reports `signals` from observed mesh state (`explicit_interest_count`, diff --git a/docs/design/TESTING.md b/docs/design/TESTING.md index f4ca94c93..14279f121 100644 --- a/docs/design/TESTING.md +++ b/docs/design/TESTING.md @@ -618,7 +618,7 @@ curl localhost:3131/api/discover # Nostr meshes (current mesh marked by mesh_id) - `/api/search` returns 200 JSON with canonical model refs for matching results - `/api/model-interests` stores and returns local explicit-interest entries keyed by canonical model refs - `/api/model-targets` returns ranked targets with explicit-interest counts, request counts, serving-node counts, `wanted` for targets not currently served, and derived `capacity_advice` without changing ranking or routing behavior -- If `[runtime] reconcile_model_targets = true` is enabled, unserved local explicit interests that are already present on disk and fit the current node may be runtime-loaded automatically. Leave it unset for read-only advisory checks. +- If `[runtime] reconcile_model_targets = true` is enabled, unserved local explicit interests that are already present on disk and fit the current node may be runtime-loaded automatically. If `reconcile_model_target_demand_upgrades = true` is also enabled, an already-serving host may replace a lower-demand local model with a locally present, higher-demand unserved target whose active demand is still within `model_target_demand_upgrade_max_age_secs`. Leave these unset for read-only advisory checks. - Discover results can be matched to current mesh by `mesh_id` ### 24. HTTP proxy single-request connection contract