chore: cargo fmt — fix lint 'Format' CI job (red since v0.2.3)

Whole-repo rustfmt drift; behaviour unchanged (whitespace/layout only).
All test platforms already passed; this greens the lint job.
This commit is contained in:
infosave2007
2026-07-08 19:32:35 +03:00
parent 3c90dad9b9
commit b4f78bd192
7 changed files with 272 additions and 95 deletions
+13 -18
View File
@@ -169,7 +169,10 @@ pub fn api_routes(admin_token: String) -> Router<SharedState> {
.route("/admin/api/shadow", get(get_shadow)) .route("/admin/api/shadow", get(get_shadow))
.route("/admin/api/hf/search", get(hf_search)) .route("/admin/api/hf/search", get(hf_search))
.route("/admin/api/import", get(list_imports).post(start_import)) .route("/admin/api/import", get(list_imports).post(start_import))
.route("/admin/api/import/:job", get(import_status).delete(delete_import)) .route(
"/admin/api/import/:job",
get(import_status).delete(delete_import),
)
.route("/admin/api/import/:job/cancel", post(cancel_import)) .route("/admin/api/import/:job/cancel", post(cancel_import))
.route("/admin/api/import/:job/register", post(register_import)) .route("/admin/api/import/:job/register", post(register_import))
.route("/admin/api/cmf", get(cmf_status)) .route("/admin/api/cmf", get(cmf_status))
@@ -736,7 +739,11 @@ async fn cmf_port_check(State(state): State<SharedState>, Query(q): Query<PortQu
if port_is_free(&host, port) { if port_is_free(&host, port) {
(true, "free", None) (true, "free", None)
} else { } else {
(false, "occupied by another process", next_free_port(&host, port)) (
false,
"occupied by another process",
next_free_port(&host, port),
)
} }
}) })
.await .await
@@ -974,10 +981,7 @@ async fn start_import(
ok(json!({ "job": id })) ok(json!({ "job": id }))
} }
async fn import_status( async fn import_status(State(state): State<SharedState>, Path(job): Path<String>) -> ApiResult {
State(state): State<SharedState>,
Path(job): Path<String>,
) -> ApiResult {
match state.imports.get(&job) { match state.imports.get(&job) {
Some(j) => ok(serde_json::to_value(j).unwrap_or_else(|_| json!({}))), Some(j) => ok(serde_json::to_value(j).unwrap_or_else(|_| json!({}))),
None => Err(ApiError::not_found(format!("job '{job}' not found"))), None => Err(ApiError::not_found(format!("job '{job}' not found"))),
@@ -985,10 +989,7 @@ async fn import_status(
} }
/// Cancel a running conversion: kills the converter and removes partial output. /// Cancel a running conversion: kills the converter and removes partial output.
async fn cancel_import( async fn cancel_import(State(state): State<SharedState>, Path(job): Path<String>) -> ApiResult {
State(state): State<SharedState>,
Path(job): Path<String>,
) -> ApiResult {
if state.imports.get(&job).is_none() { if state.imports.get(&job).is_none() {
return Err(ApiError::not_found(format!("job '{job}' not found"))); return Err(ApiError::not_found(format!("job '{job}' not found")));
} }
@@ -997,10 +998,7 @@ async fn cancel_import(
} }
/// Delete a finished conversion and its converted `.cmf` file(s) from disk. /// Delete a finished conversion and its converted `.cmf` file(s) from disk.
async fn delete_import( async fn delete_import(State(state): State<SharedState>, Path(job): Path<String>) -> ApiResult {
State(state): State<SharedState>,
Path(job): Path<String>,
) -> ApiResult {
if state.imports.get(&job).is_none() { if state.imports.get(&job).is_none() {
return Err(ApiError::not_found(format!("job '{job}' not found"))); return Err(ApiError::not_found(format!("job '{job}' not found")));
} }
@@ -1009,10 +1007,7 @@ async fn delete_import(
} }
/// Register a finished `.cmf` as a local model (OpenAI provider → cortiq-server). /// Register a finished `.cmf` as a local model (OpenAI provider → cortiq-server).
async fn register_import( async fn register_import(State(state): State<SharedState>, Path(job): Path<String>) -> ApiResult {
State(state): State<SharedState>,
Path(job): Path<String>,
) -> ApiResult {
let j = state let j = state
.imports .imports
.get(&job) .get(&job)
+26 -9
View File
@@ -113,14 +113,15 @@ async fn latest_crates_version() -> Option<String> {
.await .await
.ok()?; .ok()?;
let v: serde_json::Value = serde_json::from_str(&body).ok()?; let v: serde_json::Value = serde_json::from_str(&body).ok()?;
v["crate"]["newest_version"] v["crate"]["newest_version"].as_str().map(|s| s.to_string())
.as_str()
.map(|s| s.to_string())
} }
/// `<bin> --version` → the trailing version token (e.g. "cortiq 0.1.2" → "0.1.2"). /// `<bin> --version` → the trailing version token (e.g. "cortiq 0.1.2" → "0.1.2").
pub(crate) fn installed_version(bin: &str) -> Option<String> { pub(crate) fn installed_version(bin: &str) -> Option<String> {
let out = std::process::Command::new(bin).arg("--version").output().ok()?; let out = std::process::Command::new(bin)
.arg("--version")
.output()
.ok()?;
if !out.status.success() { if !out.status.success() {
return None; return None;
} }
@@ -130,7 +131,11 @@ pub(crate) fn installed_version(bin: &str) -> Option<String> {
fn parse_ver(s: &str) -> (u32, u32, u32) { fn parse_ver(s: &str) -> (u32, u32, u32) {
let mut it = s.split('.').map(|x| x.trim().parse::<u32>().unwrap_or(0)); let mut it = s.split('.').map(|x| x.trim().parse::<u32>().unwrap_or(0));
(it.next().unwrap_or(0), it.next().unwrap_or(0), it.next().unwrap_or(0)) (
it.next().unwrap_or(0),
it.next().unwrap_or(0),
it.next().unwrap_or(0),
)
} }
fn version_lt(a: &str, b: &str) -> bool { fn version_lt(a: &str, b: &str) -> bool {
@@ -161,7 +166,10 @@ pub async fn install_now(rt: Arc<CmfRuntime>, bin: String) {
Ok(()) => { Ok(()) => {
let v = installed_version(&bin); let v = installed_version(&bin);
rt.status.lock().unwrap().installed_version = v.clone(); rt.status.lock().unwrap().installed_version = v.clone();
rt.log(format!("installed cortiq-cli {}", v.as_deref().unwrap_or("?"))); rt.log(format!(
"installed cortiq-cli {}",
v.as_deref().unwrap_or("?")
));
} }
Err(e) => rt.fail(format!("cargo install cortiq-cli failed: {e}")), Err(e) => rt.fail(format!("cargo install cortiq-cli failed: {e}")),
} }
@@ -245,7 +253,9 @@ pub async fn manage(rt: Arc<CmfRuntime>, cfg: CmfCfg) {
rt.log("installed cortiq-cli"); rt.log("installed cortiq-cli");
} }
} else { } else {
rt.fail("cortiq binary not found (set [cmf].auto_install = true, or install cortiq-cli)"); rt.fail(
"cortiq binary not found (set [cmf].auto_install = true, or install cortiq-cli)",
);
return; return;
} }
} }
@@ -255,8 +265,15 @@ pub async fn manage(rt: Arc<CmfRuntime>, cfg: CmfCfg) {
if cfg.auto_update { if cfg.auto_update {
if let Some(latest) = latest_crates_version().await { if let Some(latest) = latest_crates_version().await {
rt.status.lock().unwrap().latest_version = Some(latest.clone()); rt.status.lock().unwrap().latest_version = Some(latest.clone());
if ver.as_deref().map(|c| version_lt(c, &latest)).unwrap_or(false) { if ver
rt.log(format!("updating cortiq-cli {} → {latest}…", ver.as_deref().unwrap_or("?"))); .as_deref()
.map(|c| version_lt(c, &latest))
.unwrap_or(false)
{
rt.log(format!(
"updating cortiq-cli {} → {latest}…",
ver.as_deref().unwrap_or("?")
));
if let Err(e) = cargo_install(true).await { if let Err(e) = cargo_install(true).await {
rt.log(format!("update failed (keeping current): {e}")); rt.log(format!("update failed (keeping current): {e}"));
} else { } else {
+10 -2
View File
@@ -10,10 +10,18 @@ use std::path::PathBuf;
/// `~/.cortiq-gateway` — so defaults work regardless of the process CWD /// `~/.cortiq-gateway` — so defaults work regardless of the process CWD
/// (important once the gateway is installed rather than run from its repo). /// (important once the gateway is installed rather than run from its repo).
pub fn data_dir() -> PathBuf { pub fn data_dir() -> PathBuf {
for (var, sub) in [("CORTIQ_GATEWAY_HOME", ""), ("XDG_DATA_HOME", "cortiq-gateway"), ("HOME", ".cortiq-gateway")] { for (var, sub) in [
("CORTIQ_GATEWAY_HOME", ""),
("XDG_DATA_HOME", "cortiq-gateway"),
("HOME", ".cortiq-gateway"),
] {
if let Ok(v) = std::env::var(var) { if let Ok(v) = std::env::var(var) {
if !v.is_empty() { if !v.is_empty() {
return if sub.is_empty() { PathBuf::from(v) } else { PathBuf::from(v).join(sub) }; return if sub.is_empty() {
PathBuf::from(v)
} else {
PathBuf::from(v).join(sub)
};
} }
} }
} }
+95 -28
View File
@@ -31,13 +31,19 @@ fn cleanup_output(path: &str) {
} }
fn now() -> u64 { fn now() -> u64 {
SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs() SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_secs()
} }
/// Job id from a hash of (repo, nanos) — unique without a uuid dependency. /// Job id from a hash of (repo, nanos) — unique without a uuid dependency.
fn gen_id(repo: &str) -> String { fn gen_id(repo: &str) -> String {
use std::hash::{Hash, Hasher}; use std::hash::{Hash, Hasher};
let nanos = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_nanos(); let nanos = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_nanos();
let mut h = std::collections::hash_map::DefaultHasher::new(); let mut h = std::collections::hash_map::DefaultHasher::new();
repo.hash(&mut h); repo.hash(&mut h);
nanos.hash(&mut h); nanos.hash(&mut h);
@@ -82,8 +88,8 @@ pub struct Job {
pub started: u64, pub started: u64,
pub finished: Option<u64>, pub finished: Option<u64>,
pub size_bytes: Option<u64>, pub size_bytes: Option<u64>,
pub progress: Option<f32>, // 0..1, parsed from converter @PROGRESS pub progress: Option<f32>, // 0..1, parsed from converter @PROGRESS
pub phase: Option<String>, // current phase label pub phase: Option<String>, // current phase label
} }
#[derive(Default)] #[derive(Default)]
@@ -185,9 +191,17 @@ impl JobStore {
} }
/// Proxy HuggingFace model search (avoids browser CORS + adds our token). /// Proxy HuggingFace model search (avoids browser CORS + adds our token).
pub async fn hf_search(query: &str, limit: usize, token: Option<&str>) -> Result<serde_json::Value, String> { pub async fn hf_search(
query: &str,
limit: usize,
token: Option<&str>,
) -> Result<serde_json::Value, String> {
let q: String = url_escape(query); let q: String = url_escape(query);
let sort = if query.trim().is_empty() { "trendingScore" } else { "downloads" }; let sort = if query.trim().is_empty() {
"trendingScore"
} else {
"downloads"
};
let url = format!( let url = format!(
"https://huggingface.co/api/models?search={q}&sort={sort}&direction=-1&limit={limit}&full=false" "https://huggingface.co/api/models?search={q}&sort={sort}&direction=-1&limit={limit}&full=false"
); );
@@ -218,7 +232,13 @@ fn url_escape(s: &str) -> String {
fn sanitize(s: &str) -> String { fn sanitize(s: &str) -> String {
s.chars() s.chars()
.map(|c| if c.is_ascii_alphanumeric() || c == '-' || c == '_' { c } else { '-' }) .map(|c| {
if c.is_ascii_alphanumeric() || c == '-' || c == '_' {
c
} else {
'-'
}
})
.collect() .collect()
} }
@@ -244,7 +264,10 @@ pub fn start_import(store: Arc<JobStore>, cfg: &CmfCfg, p: ImportParams) -> Resu
// doesn't support yet (linear-attention folding, v-bit shaping). // doesn't support yet (linear-attention folding, v-bit shaping).
let use_python = p.linear_core.is_some() || p.vbit_shape.is_some() || p.mean_bits.is_some(); let use_python = p.linear_core.is_some() || p.vbit_shape.is_some() || p.mean_bits.is_some();
if use_python && !std::path::Path::new(&cfg.converter).exists() { if use_python && !std::path::Path::new(&cfg.converter).exists() {
return Err(format!("advanced options need the Python converter, not found: {}", cfg.converter)); return Err(format!(
"advanced options need the Python converter, not found: {}",
cfg.converter
));
} }
let _ = std::fs::create_dir_all(&cfg.models_dir); let _ = std::fs::create_dir_all(&cfg.models_dir);
let base = if p.name.trim().is_empty() { let base = if p.name.trim().is_empty() {
@@ -288,18 +311,43 @@ pub fn start_import(store: Arc<JobStore>, cfg: &CmfCfg, p: ImportParams) -> Resu
let (program, args, workdir): (String, Vec<String>, std::path::PathBuf) = if use_python { let (program, args, workdir): (String, Vec<String>, std::path::PathBuf) = if use_python {
let conv = std::path::Path::new(&cfg.converter); let conv = std::path::Path::new(&cfg.converter);
let mut a = vec![ let mut a = vec![
std::fs::canonicalize(conv).map(|p| p.to_string_lossy().to_string()).unwrap_or_else(|_| cfg.converter.clone()), std::fs::canonicalize(conv)
"--model".into(), p.repo.clone(), .map(|p| p.to_string_lossy().to_string())
"--quant".into(), p.quant.clone(), .unwrap_or_else(|_| cfg.converter.clone()),
"--output".into(), output_abs.clone(), "--model".into(),
p.repo.clone(),
"--quant".into(),
p.quant.clone(),
"--output".into(),
output_abs.clone(),
]; ];
if let Some(lc) = &p.linear_core { a.push("--linear-core".into()); a.push(lc.clone()); } if let Some(lc) = &p.linear_core {
if let Some(n) = p.nphase { a.push("--nphase".into()); a.push(n.to_string()); } a.push("--linear-core".into());
if let Some(vs) = &p.vbit_shape { a.push("--vbit-shape".into()); a.push(vs.clone()); } a.push(lc.clone());
if let Some(mb) = p.mean_bits { a.push("--mean-bits".into()); a.push(mb.to_string()); } }
if let Some(g) = p.shard_max_gb { a.push("--shard-max-gb".into()); a.push(g.to_string()); } if let Some(n) = p.nphase {
if p.skip_mtp { a.push("--skip-mtp".into()); } a.push("--nphase".into());
let wd = conv.parent().map(|d| d.to_path_buf()).unwrap_or_else(|| std::path::PathBuf::from(".")); a.push(n.to_string());
}
if let Some(vs) = &p.vbit_shape {
a.push("--vbit-shape".into());
a.push(vs.clone());
}
if let Some(mb) = p.mean_bits {
a.push("--mean-bits".into());
a.push(mb.to_string());
}
if let Some(g) = p.shard_max_gb {
a.push("--shard-max-gb".into());
a.push(g.to_string());
}
if p.skip_mtp {
a.push("--skip-mtp".into());
}
let wd = conv
.parent()
.map(|d| d.to_path_buf())
.unwrap_or_else(|| std::path::PathBuf::from("."));
(cfg.python_bin.clone(), a, wd) (cfg.python_bin.clone(), a, wd)
} else if p.repo.to_ascii_lowercase().contains("gguf") { } else if p.repo.to_ascii_lowercase().contains("gguf") {
// GGUF repo → native `cortiq import-gguf` (downloads + dequantizes any // GGUF repo → native `cortiq import-gguf` (downloads + dequantizes any
@@ -307,21 +355,40 @@ pub fn start_import(store: Arc<JobStore>, cfg: &CmfCfg, p: ImportParams) -> Resu
let mut a = vec![ let mut a = vec![
"import-gguf".into(), "import-gguf".into(),
p.repo.clone(), p.repo.clone(),
"--quant".into(), map_quant(&p.quant).into(), "--quant".into(),
"--output".into(), output_abs.clone(), map_quant(&p.quant).into(),
"--output".into(),
output_abs.clone(),
]; ];
if let Some(t) = &hf_token { a.push("--hf-token".into()); a.push(t.clone()); } if let Some(t) = &hf_token {
(cfg.cortiq_bin.clone(), a, std::env::current_dir().unwrap_or_else(|_| std::path::PathBuf::from("."))) a.push("--hf-token".into());
a.push(t.clone());
}
(
cfg.cortiq_bin.clone(),
a,
std::env::current_dir().unwrap_or_else(|_| std::path::PathBuf::from(".")),
)
} else { } else {
// safetensors → native `cortiq convert` (dense / MoE / GatedDeltaNet). // safetensors → native `cortiq convert` (dense / MoE / GatedDeltaNet).
let mut a = vec![ let mut a = vec![
"convert".into(), "convert".into(),
"--model".into(), p.repo.clone(), "--model".into(),
"--quant".into(), map_quant(&p.quant).into(), p.repo.clone(),
"--output".into(), output_abs.clone(), "--quant".into(),
map_quant(&p.quant).into(),
"--output".into(),
output_abs.clone(),
]; ];
if let Some(t) = &hf_token { a.push("--hf-token".into()); a.push(t.clone()); } if let Some(t) = &hf_token {
(cfg.cortiq_bin.clone(), a, std::env::current_dir().unwrap_or_else(|_| std::path::PathBuf::from("."))) a.push("--hf-token".into());
a.push(t.clone());
}
(
cfg.cortiq_bin.clone(),
a,
std::env::current_dir().unwrap_or_else(|_| std::path::PathBuf::from(".")),
)
}; };
let ret_id = id.clone(); let ret_id = id.clone();
+28 -8
View File
@@ -208,15 +208,25 @@ impl Pipeline {
}; };
let (decision, route_source, candidates) = if live.cfg.cmf.local_only { let (decision, route_source, candidates) = if live.cfg.cmf.local_only {
(mk("local-only", "local"), "local".to_string(), live.local_candidates()) (
mk("local-only", "local"),
"local".to_string(),
live.local_candidates(),
)
} else { } else {
match &req.routing { match &req.routing {
RoutingDirective::Pinned { model_id } => { RoutingDirective::Pinned { model_id } => (
(mk("pinned", "pinned"), "pinned".to_string(), vec![model_id.clone()]) mk("pinned", "pinned"),
} "pinned".to_string(),
vec![model_id.clone()],
),
RoutingDirective::Auto { profile } => { RoutingDirective::Auto { profile } => {
if !live.cfg.router.enabled { if !live.cfg.router.enabled {
(mk("router-disabled", "local"), "local".to_string(), live.local_candidates()) (
mk("router-disabled", "local"),
"local".to_string(),
live.local_candidates(),
)
} else { } else {
let prof = profile.as_deref().unwrap_or(&live.cfg.route.profile); let prof = profile.as_deref().unwrap_or(&live.cfg.route.profile);
match live.router.route(&text, prof).await { match live.router.route(&text, prof).await {
@@ -229,7 +239,11 @@ impl Pipeline {
tracing::warn!( tracing::warn!(
"Router is unavailable. Gracefully degrading to the local/default model." "Router is unavailable. Gracefully degrading to the local/default model."
); );
(mk("degraded", "fallback"), "fallback".to_string(), live.local_candidates()) (
mk("degraded", "fallback"),
"fallback".to_string(),
live.local_candidates(),
)
} }
} }
} }
@@ -289,7 +303,10 @@ impl Pipeline {
&& !sh.local_model_id.is_empty() && !sh.local_model_id.is_empty()
&& matches!(req.routing, RoutingDirective::Auto { .. }) && matches!(req.routing, RoutingDirective::Auto { .. })
&& state.promotion.serves_local(&decision.task_label) && state.promotion.serves_local(&decision.task_label)
&& !matches!(decision.complexity_tier.as_str(), "high" | "hard" | "complex") && !matches!(
decision.complexity_tier.as_str(),
"high" | "hard" | "complex"
)
&& !final_candidates.iter().any(|m| m == &sh.local_model_id) && !final_candidates.iter().any(|m| m == &sh.local_model_id)
{ {
final_candidates.insert(0, sh.local_model_id.clone()); final_candidates.insert(0, sh.local_model_id.clone());
@@ -747,7 +764,10 @@ async fn shadow_capture(
}; };
// 1. Local answer (free, CPU). // 1. Local answer (free, CPU).
let a_local = match local.chat(shadow_req(&cfg.local_model_id, clip(&prompt, 4000), 512)).await { let a_local = match local
.chat(shadow_req(&cfg.local_model_id, clip(&prompt, 4000), 512))
.await
{
Ok(r) => r Ok(r) => r
.choices .choices
.into_iter() .into_iter()
+95 -29
View File
@@ -18,7 +18,10 @@ use std::sync::Mutex;
use std::time::{SystemTime, UNIX_EPOCH}; use std::time::{SystemTime, UNIX_EPOCH};
fn now_secs() -> u64 { fn now_secs() -> u64 {
SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs() SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_secs()
} }
/// One judgment of a local answer (JSONL line, replayed on startup). /// One judgment of a local answer (JSONL line, replayed on startup).
@@ -62,14 +65,14 @@ pub enum PromotionState {
#[derive(Clone, Debug)] #[derive(Clone, Debug)]
pub struct PromotionCfg { pub struct PromotionCfg {
pub enabled: bool, pub enabled: bool,
pub window: usize, // rolling judgments per label (W) pub window: usize, // rolling judgments per label (W)
pub n_min: usize, // min judgments before promotion is even considered pub n_min: usize, // min judgments before promotion is even considered
pub promote_lb: f64, // Wilson-95 LB of pass must clear this pub promote_lb: f64, // Wilson-95 LB of pass must clear this
pub demote_pass: f64, // point pass-rate below this → demote a served label pub demote_pass: f64, // point pass-rate below this → demote a served label
pub hardfail_ub_max: f64,// Wilson-95 UPPER bound of hard_fail must stay under this pub hardfail_ub_max: f64, // Wilson-95 UPPER bound of hard_fail must stay under this
pub soak: usize, // extra Canary judgments before LocalServed pub soak: usize, // extra Canary judgments before LocalServed
pub z: f64, // 1.96 = 95% pub z: f64, // 1.96 = 95%
pub file: Option<String>,// judge.jsonl path (append + replay) pub file: Option<String>, // judge.jsonl path (append + replay)
} }
impl Default for PromotionCfg { impl Default for PromotionCfg {
@@ -92,7 +95,7 @@ struct LabelState {
passes: VecDeque<bool>, // rolling window of pass/fail passes: VecDeque<bool>, // rolling window of pass/fail
hard_fails: VecDeque<bool>, // rolling window of hard_fail flags hard_fails: VecDeque<bool>, // rolling window of hard_fail flags
state: PromotionState, state: PromotionState,
soaked: usize, // judgments accumulated while in Canary soaked: usize, // judgments accumulated while in Canary
} }
impl LabelState { impl LabelState {
@@ -124,20 +127,29 @@ pub fn wilson_bound(s: usize, n: usize, z: f64, upper: bool) -> f64 {
let denom = 1.0 + z2 / n; let denom = 1.0 + z2 / n;
let centre = p + z2 / (2.0 * n); let centre = p + z2 / (2.0 * n);
let margin = z * ((p * (1.0 - p) / n) + z2 / (4.0 * n * n)).sqrt(); let margin = z * ((p * (1.0 - p) / n) + z2 / (4.0 * n * n)).sqrt();
let v = if upper { centre + margin } else { centre - margin }; let v = if upper {
centre + margin
} else {
centre - margin
};
(v / denom).clamp(0.0, 1.0) (v / denom).clamp(0.0, 1.0)
} }
impl Promotion { impl Promotion {
pub fn new(cfg: PromotionCfg) -> std::sync::Arc<Self> { pub fn new(cfg: PromotionCfg) -> std::sync::Arc<Self> {
let p = std::sync::Arc::new(Self { cfg, labels: Mutex::new(HashMap::new()) }); let p = std::sync::Arc::new(Self {
cfg,
labels: Mutex::new(HashMap::new()),
});
p.replay(); p.replay();
p p
} }
fn replay(&self) { fn replay(&self) {
let Some(path) = &self.cfg.file else { return }; let Some(path) = &self.cfg.file else { return };
let Ok(content) = std::fs::read_to_string(path) else { return }; let Ok(content) = std::fs::read_to_string(path) else {
return;
};
for line in content.lines() { for line in content.lines() {
if let Ok(rec) = serde_json::from_str::<JudgeRecord>(line) { if let Ok(rec) = serde_json::from_str::<JudgeRecord>(line) {
self.apply(&rec); self.apply(&rec);
@@ -151,8 +163,10 @@ impl Promotion {
let st = self.apply(&rec); let st = self.apply(&rec);
if let Some(path) = &self.cfg.file { if let Some(path) = &self.cfg.file {
if let Ok(line) = serde_json::to_string(&rec) { if let Ok(line) = serde_json::to_string(&rec) {
if let Ok(mut f) = if let Ok(mut f) = std::fs::OpenOptions::new()
std::fs::OpenOptions::new().create(true).append(true).open(path) .create(true)
.append(true)
.open(path)
{ {
let _ = writeln!(f, "{line}"); let _ = writeln!(f, "{line}");
} }
@@ -163,7 +177,9 @@ impl Promotion {
fn apply(&self, rec: &JudgeRecord) -> PromotionState { fn apply(&self, rec: &JudgeRecord) -> PromotionState {
let mut map = self.labels.lock().unwrap(); let mut map = self.labels.lock().unwrap();
let ls = map.entry(rec.task_label.clone()).or_insert_with(LabelState::new); let ls = map
.entry(rec.task_label.clone())
.or_insert_with(LabelState::new);
push_cap(&mut ls.passes, rec.pass, self.cfg.window); push_cap(&mut ls.passes, rec.pass, self.cfg.window);
push_cap(&mut ls.hard_fails, rec.hard_fail, self.cfg.window); push_cap(&mut ls.hard_fails, rec.hard_fail, self.cfg.window);
self.advance(ls); self.advance(ls);
@@ -188,7 +204,12 @@ impl Promotion {
ls.state = match ls.state { ls.state = match ls.state {
PromotionState::CloudOnly | PromotionState::ShadowJudged => { PromotionState::CloudOnly | PromotionState::ShadowJudged => {
if bar { ls.soaked = 0; PromotionState::Canary } else { PromotionState::ShadowJudged } if bar {
ls.soaked = 0;
PromotionState::Canary
} else {
PromotionState::ShadowJudged
}
} }
PromotionState::Canary => { PromotionState::Canary => {
if regressed { if regressed {
@@ -203,7 +224,11 @@ impl Promotion {
} }
} }
PromotionState::LocalServed => { PromotionState::LocalServed => {
if regressed { PromotionState::Demoted } else { PromotionState::LocalServed } if regressed {
PromotionState::Demoted
} else {
PromotionState::LocalServed
}
} }
// Demoted goes back to SHADOW to re-earn promotion (skill stays in .cmf). // Demoted goes back to SHADOW to re-earn promotion (skill stays in .cmf).
PromotionState::Demoted => PromotionState::ShadowJudged, PromotionState::Demoted => PromotionState::ShadowJudged,
@@ -223,7 +248,10 @@ impl Promotion {
/// Should a request of this label be SERVED by the local model? /// Should a request of this label be SERVED by the local model?
/// (Canary/LocalServed; caller still applies the per-request veto.) /// (Canary/LocalServed; caller still applies the per-request veto.)
pub fn serves_local(&self, label: &str) -> bool { pub fn serves_local(&self, label: &str) -> bool {
matches!(self.state(label), PromotionState::Canary | PromotionState::LocalServed) matches!(
self.state(label),
PromotionState::Canary | PromotionState::LocalServed
)
} }
/// Snapshot for /admin: (state, n, pass-rate, Wilson-LB) per label. /// Snapshot for /admin: (state, n, pass-rate, Wilson-LB) per label.
@@ -235,7 +263,13 @@ impl Promotion {
let n = ls.passes.len(); let n = ls.passes.len();
let s = ls.passes.iter().filter(|&&p| p).count(); let s = ls.passes.iter().filter(|&&p| p).count();
let p = if n > 0 { s as f64 / n as f64 } else { 0.0 }; let p = if n > 0 { s as f64 / n as f64 } else { 0.0 };
(k.clone(), ls.state, n, p, wilson_bound(s, n, self.cfg.z, false)) (
k.clone(),
ls.state,
n,
p,
wilson_bound(s, n, self.cfg.z, false),
)
}) })
.collect(); .collect();
out.sort_by(|a, b| b.2.cmp(&a.2)); out.sort_by(|a, b| b.2.cmp(&a.2));
@@ -259,12 +293,24 @@ mod tests {
use super::*; use super::*;
fn cfg() -> PromotionCfg { fn cfg() -> PromotionCfg {
PromotionCfg { enabled: true, n_min: 200, soak: 0, ..Default::default() } PromotionCfg {
enabled: true,
n_min: 200,
soak: 0,
..Default::default()
}
} }
fn rec(label: &str, pass: bool, hard_fail: bool) -> JudgeRecord { fn rec(label: &str, pass: bool, hard_fail: bool) -> JudgeRecord {
JudgeRecord { JudgeRecord {
ts: 0, task_label: label.into(), pass, hard_fail, score: if pass { 4 } else { 0 }, ts: 0,
source: "checker".into(), recon_e: None, born_conf: None, judge_tokens: 0, task_label: label.into(),
pass,
hard_fail,
score: if pass { 4 } else { 0 },
source: "checker".into(),
recon_e: None,
born_conf: None,
judge_tokens: 0,
} }
} }
@@ -316,7 +362,10 @@ mod tests {
match row { match row {
Some((_, st, n, pr, lb)) => println!( Some((_, st, n, pr, lb)) => println!(
" {label:<12} {:<13} n={n:<4} pass={:.0}% LB={:.3} serves_local={}", " {label:<12} {:<13} n={n:<4} pass={:.0}% LB={:.3} serves_local={}",
format!("{st:?}"), pr * 100.0, lb, p.serves_local(label) format!("{st:?}"),
pr * 100.0,
lb,
p.serves_local(label)
), ),
None => println!(" {label:<12} (cloud-only, no judgments)"), None => println!(" {label:<12} (cloud-only, no judgments)"),
} }
@@ -324,19 +373,29 @@ mod tests {
println!("\n── self-warming loop, live ──"); println!("\n── self-warming loop, live ──");
// A mature skill (e.g. ru-tech): local answers pass the judge ~99%. // A mature skill (e.g. ru-tech): local answers pass the judge ~99%.
println!(" 1) SHADOW→CANARY: судим локаль 'ru_tech' (сильный скилл ~99% pass), 210 судейств:"); println!(
" 1) SHADOW→CANARY: судим локаль 'ru_tech' (сильный скилл ~99% pass), 210 судейств:"
);
for i in 0..210 { for i in 0..210 {
p.record(rec("ru_tech", i % 100 != 0, false)); p.record(rec("ru_tech", i % 100 != 0, false));
} }
show("ru_tech"); show("ru_tech");
assert_eq!(p.state("ru_tech"), PromotionState::Canary, "n≥200 & LB≥0.95 → CANARY"); assert_eq!(
p.state("ru_tech"),
PromotionState::Canary,
"n≥200 & LB≥0.95 → CANARY"
);
println!(" 2) CANARY soak (ещё 60 судейств держат планку) → LOCAL_SERVED:"); println!(" 2) CANARY soak (ещё 60 судейств держат планку) → LOCAL_SERVED:");
for i in 0..60 { for i in 0..60 {
p.record(rec("ru_tech", i % 100 != 0, false)); p.record(rec("ru_tech", i % 100 != 0, false));
} }
show("ru_tech"); show("ru_tech");
assert_eq!(p.state("ru_tech"), PromotionState::LocalServed, "soak → LOCAL_SERVED"); assert_eq!(
p.state("ru_tech"),
PromotionState::LocalServed,
"soak → LOCAL_SERVED"
);
assert!(p.serves_local("ru_tech"), "теперь обслуживается локально"); assert!(p.serves_local("ru_tech"), "теперь обслуживается локально");
// The code dataset: local fails the checker → never promotes. // The code dataset: local fails the checker → never promotes.
@@ -345,7 +404,11 @@ mod tests {
p.record(rec("code", false, false)); p.record(rec("code", false, false));
} }
show("code"); show("code");
assert_eq!(p.state("code"), PromotionState::ShadowJudged, "остаётся на облаке — безопасно"); assert_eq!(
p.state("code"),
PromotionState::ShadowJudged,
"остаётся на облаке — безопасно"
);
assert!(!p.serves_local("code")); assert!(!p.serves_local("code"));
// Drift on the promoted label → auto-demote back to cloud. // Drift on the promoted label → auto-demote back to cloud.
@@ -376,6 +439,9 @@ mod tests {
for _ in 0..300 { for _ in 0..300 {
p2.record(rec("t", false, false)); // quality collapses p2.record(rec("t", false, false)); // quality collapses
} }
assert!(matches!(p2.state("t"), PromotionState::Demoted | PromotionState::ShadowJudged)); assert!(matches!(
p2.state("t"),
PromotionState::Demoted | PromotionState::ShadowJudged
));
} }
} }
+5 -1
View File
@@ -90,7 +90,11 @@ fn promotion_cfg(s: &crate::config::ShadowCfg) -> crate::promotion::PromotionCfg
n_min: s.n_min, n_min: s.n_min,
promote_lb: s.promote_lb, promote_lb: s.promote_lb,
soak: s.soak, soak: s.soak,
file: if s.file.is_empty() { None } else { Some(s.file.clone()) }, file: if s.file.is_empty() {
None
} else {
Some(s.file.clone())
},
..Default::default() ..Default::default()
} }
} }