Skip to content

Commit e564c63

Browse files
committed
Merge commit '101b847b9c5fd4c1c96ac6cefbc70877a61a89c0' into land/owned-metal-pooling
2 parents bc99d55 + 101b847 commit e564c63

4 files changed

Lines changed: 252 additions & 1 deletion

File tree

‎crates/synapse-module/src/lib.rs‎

Lines changed: 79 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2814,6 +2814,13 @@ fn sync_and_load_catalog_models(
28142814
cache: &ModelCache,
28152815
) -> Result<Vec<StoredModelConfig>, ModuleError> {
28162816
let now = now_ms();
2817+
// Only rows present before startup are legacy registrations. A duplicate
2818+
// preload id in this startup must not bypass new-registration validation.
2819+
let persisted_ids: BTreeSet<_> = store
2820+
.catalog_models()?
2821+
.into_iter()
2822+
.map(|model| model.model_id)
2823+
.collect();
28172824
for (index, preload) in config.preload_models.clone().into_iter().enumerate() {
28182825
let id = preload
28192826
.model_id
@@ -2833,6 +2840,9 @@ fn sync_and_load_catalog_models(
28332840
continue;
28342841
}
28352842
let spec = build_preload_catalog_model(index, preload, &config.inline, &config.jobs)?;
2843+
if !persisted_ids.contains(&id) {
2844+
validate_new_owned_metal_readout(&spec).map_err(ModuleError::Config)?;
2845+
}
28362846
store.upsert_model(&spec, now)?;
28372847
}
28382848

@@ -3646,6 +3656,46 @@ fn build_stored_model_config(
36463656
})
36473657
}
36483658

3659+
fn owned_metal_readout_pooling(family: OwnedFamily, task: ModelTask) -> Option<WorkerPooling> {
3660+
match (family, task) {
3661+
(OwnedFamily::GteModernBert, ModelTask::Embed) => Some(WorkerPooling::Cls),
3662+
(OwnedFamily::GteModernBert, ModelTask::Rerank) => Some(WorkerPooling::Mean),
3663+
(OwnedFamily::Qwen3, ModelTask::Embed | ModelTask::Rerank) => Some(WorkerPooling::Last),
3664+
_ => None,
3665+
}
3666+
}
3667+
3668+
// Call only at new-registration boundaries, never while rebuilding stored
3669+
// configs: legacy declarations are part of persisted vector fingerprints.
3670+
fn validate_new_owned_metal_readout(spec: &StoredModelConfig) -> Result<(), String> {
3671+
if spec.engine != "owned-metal" || spec.engine_identity.build_flags.contains_key("profile") {
3672+
return Ok(());
3673+
}
3674+
let family = spec
3675+
.owned_family
3676+
.as_deref()
3677+
.ok_or_else(|| "owned-metal registration is missing family".to_string())?;
3678+
let family = OwnedFamily::parse(family).map_err(|error| error.to_string())?;
3679+
let task = parse_model_task(Some(&spec.task), &spec.engine, &spec.model_id)
3680+
.map_err(|error| error.to_string())?;
3681+
if let Some(expected) = owned_metal_readout_pooling(family, task) {
3682+
if spec.pooling != expected.as_str() {
3683+
let operation = match task {
3684+
ModelTask::Embed => "embedding",
3685+
ModelTask::Rerank => "reranking",
3686+
ModelTask::Generate => "generation",
3687+
};
3688+
return Err(format!(
3689+
"owned-metal {} {operation} computes {}; declare \"pooling\": \"{}\"",
3690+
family.as_str(),
3691+
expected.as_str(),
3692+
expected.as_str()
3693+
));
3694+
}
3695+
}
3696+
Ok(())
3697+
}
3698+
36493699
fn owned_tokenizer_max_tokens(max_tokens: usize, owned: Option<&OwnedCatalogConfig>) -> usize {
36503700
if owned.is_some_and(|profile| profile.family == OwnedFamily::Qwen3) {
36513701
max_tokens.saturating_sub(1).max(1)
@@ -8117,6 +8167,13 @@ async fn execute_model_load_job(state: Arc<ModuleState>, job_id: String, params:
81178167
&state.runtime.inline,
81188168
&state.runtime.jobs,
81198169
)?;
8170+
validate_new_owned_metal_readout(&spec).map_err(|message| {
8171+
catalog_wire_error(
8172+
"invalid_request",
8173+
json!({"model_id": spec.model_id}),
8174+
message,
8175+
)
8176+
})?;
81208177
register_runtime_catalog_model(&state.runtime, spec.clone())?;
81218178
state.store.upsert_model(&spec, now_ms()).map_err(|error| {
81228179
transient_model_load_error(format!(
@@ -17059,6 +17116,9 @@ fn now_ms() -> u64 {
1705917116

1706017117
#[cfg(test)]
1706117118
mod tests {
17119+
mod owned_metal_readout_tests {
17120+
include!("owned_metal_readout_tests.rs");
17121+
}
1706217122
#[cfg(unix)]
1706317123
mod coreml_load_tests {
1706417124
include!("coreml_load_tests.rs");
@@ -23836,6 +23896,25 @@ mod tests {
2383623896
.contains_key(PROJECTION_DTYPE_CONFIG_KEY));
2383723897
assert!(!spec.engine_identity.build_flags.contains_key("profile"));
2383823898
assert_eq!(spec.fingerprint.0, pinned);
23899+
let (root, descriptor) = test_storage_descriptor("persisted-metal-pin");
23900+
let store = SynapseStore::open(&descriptor).unwrap();
23901+
store.upsert_model(&spec, 1).unwrap();
23902+
let config = ModuleConfig {
23903+
inline,
23904+
jobs,
23905+
..ModuleConfig::default()
23906+
};
23907+
let restored = sync_and_load_catalog_models(
23908+
&store,
23909+
&config,
23910+
&runtime_release_catalog().unwrap(),
23911+
&ModelCache::new(root.join("cache")),
23912+
)
23913+
.unwrap();
23914+
assert_eq!(restored, vec![spec]);
23915+
assert_eq!(restored[0].fingerprint.0, pinned);
23916+
drop(store);
23917+
std::fs::remove_dir_all(root).unwrap();
2383923918
}
2384023919
}
2384123920

Lines changed: 166 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,166 @@
1+
use super::*;
2+
3+
fn fixture(label: &str) -> (PathBuf, Arc<ModuleState>, PreloadModelConfig) {
4+
let (root, descriptor) = test_storage_descriptor(label);
5+
let store = Arc::new(SynapseStore::open(&descriptor).unwrap());
6+
let profile = test_machine_profile(label);
7+
store.activate_profile(&profile, 1, 1000).unwrap();
8+
let mut state = test_module_state(store, profile);
9+
Arc::get_mut(&mut state).unwrap().model_cache = Arc::new(ModelCache::new(root.join("cache")));
10+
make_test_tokenizer(&root, 512);
11+
fs::write(root.join("config.json"), br#"{"model_type":"modernbert"}"#).unwrap();
12+
let header = br#"{"embedding.weight":{"dtype":"F16","shape":[2],"data_offsets":[0,4]}}"#;
13+
let mut weights = (header.len() as u64).to_le_bytes().to_vec();
14+
weights.extend_from_slice(header);
15+
weights.extend_from_slice(&[0; 4]);
16+
fs::write(root.join("model.safetensors"), weights).unwrap();
17+
let preload = serde_json::from_value(json!({
18+
"model_id": "legacy-gte-embed", "engine": "owned-metal", "task": "embed",
19+
"family": "gte-modernbert", "dtype": "f16", "pooling": "mean",
20+
"model_path": root.join("model.safetensors"),
21+
"tokenizer_path": root.join("tokenizer.json")
22+
}))
23+
.unwrap();
24+
(root, state, preload)
25+
}
26+
27+
#[test]
28+
fn readout_table_is_operation_specific() {
29+
for (family, task, expected) in [
30+
(
31+
OwnedFamily::GteModernBert,
32+
ModelTask::Embed,
33+
WorkerPooling::Cls,
34+
),
35+
(
36+
OwnedFamily::GteModernBert,
37+
ModelTask::Rerank,
38+
WorkerPooling::Mean,
39+
),
40+
(OwnedFamily::Qwen3, ModelTask::Embed, WorkerPooling::Last),
41+
(OwnedFamily::Qwen3, ModelTask::Rerank, WorkerPooling::Last),
42+
] {
43+
assert_eq!(owned_metal_readout_pooling(family, task), Some(expected));
44+
}
45+
}
46+
47+
#[tokio::test]
48+
async fn new_owned_metal_gte_embed_load_refuses_mean_and_omitted_pooling() {
49+
let (root, state, _) = fixture("metal-readout-load");
50+
for pooling in [Some("mean"), None] {
51+
let mut params = json!({
52+
"source": "file", "path": root, "engine": "owned-metal",
53+
"family": "gte-modernbert", "task": "embed",
54+
"model_id": "new-gte-embed",
55+
"files": {"model": "model.safetensors", "tokenizer": "tokenizer.json", "config": "config.json"},
56+
"request_key": format!("readout-{pooling:?}")
57+
});
58+
if let Some(pooling) = pooling {
59+
params["pooling"] = json!(pooling);
60+
}
61+
let accepted = response_result(model_load(state.clone(), params).await, "model.load");
62+
let job_id = accepted["job_id"].as_str().expect("load job admitted");
63+
let record = tokio::time::timeout(Duration::from_secs(10), async {
64+
loop {
65+
let record = state.store.get_job(job_id).unwrap().unwrap();
66+
if record.state == JOB_STATE_FAILED_PERMANENT
67+
|| record.state == JOB_STATE_FAILED_TRANSIENT
68+
|| record.state == JOB_STATE_DONE
69+
{
70+
break record;
71+
}
72+
tokio::task::yield_now().await;
73+
}
74+
})
75+
.await
76+
.expect("load job completed");
77+
let error = record.error_json.unwrap_or(Value::Null);
78+
assert_eq!(error["code"], "invalid_request", "{error}");
79+
assert_eq!(
80+
error["message"],
81+
"owned-metal gte-modernbert embedding computes cls; declare \"pooling\": \"cls\""
82+
);
83+
assert!(state.store.catalog_models().unwrap().is_empty());
84+
assert!(model_slot_snapshot(&state.runtime, "new-gte-embed").is_none());
85+
}
86+
drop(state);
87+
fs::remove_dir_all(root).unwrap();
88+
}
89+
90+
#[test]
91+
fn new_owned_metal_preload_refuses_mean_but_accepts_cls() {
92+
let (root, state, mut preload) = fixture("metal-readout-preload");
93+
for pooling in [Some("mean"), None] {
94+
preload.pooling = pooling.map(str::to_string);
95+
let config = ModuleConfig {
96+
preload_models: vec![preload.clone()],
97+
..ModuleConfig::default()
98+
};
99+
let error = sync_and_load_catalog_models(
100+
&state.store,
101+
&config,
102+
&state.runtime.release_catalog,
103+
&state.model_cache,
104+
)
105+
.unwrap_err();
106+
assert!(error
107+
.to_string()
108+
.contains("computes cls; declare \"pooling\": \"cls\""));
109+
assert!(state.store.catalog_models().unwrap().is_empty());
110+
}
111+
preload.pooling = Some("cls".into());
112+
let config = ModuleConfig {
113+
preload_models: vec![preload],
114+
..ModuleConfig::default()
115+
};
116+
let loaded = sync_and_load_catalog_models(
117+
&state.store,
118+
&config,
119+
&state.runtime.release_catalog,
120+
&state.model_cache,
121+
)
122+
.unwrap();
123+
assert_eq!(loaded.len(), 1);
124+
assert_eq!(loaded[0].pooling, "cls");
125+
drop(state);
126+
fs::remove_dir_all(root).unwrap();
127+
}
128+
129+
#[test]
130+
fn persisted_owned_metal_mean_survives_startup_with_and_without_preload() {
131+
let (root, state, preload) = fixture("metal-readout-persisted");
132+
// Capture the legacy declaration through the unchanged builder, before
133+
// simulating a restart with a row that already exists in the store.
134+
let legacy = build_preload_catalog_model(
135+
0,
136+
preload.clone(),
137+
&state.runtime.inline,
138+
&state.runtime.jobs,
139+
)
140+
.unwrap();
141+
state.store.upsert_model(&legacy, 1).unwrap();
142+
let legacy_bytes = serde_json::to_vec(&legacy).unwrap();
143+
for preloads in [Vec::new(), vec![preload]] {
144+
let config = ModuleConfig {
145+
preload_models: preloads,
146+
..ModuleConfig::default()
147+
};
148+
let loaded = sync_and_load_catalog_models(
149+
&state.store,
150+
&config,
151+
&state.runtime.release_catalog,
152+
&state.model_cache,
153+
)
154+
.unwrap();
155+
assert_eq!(loaded.len(), 1);
156+
assert_eq!(loaded[0].pooling, "mean");
157+
assert_eq!(loaded[0].fingerprint, legacy.fingerprint);
158+
assert_eq!(serde_json::to_vec(&loaded[0]).unwrap(), legacy_bytes);
159+
assert_eq!(state.store.catalog_models().unwrap(), vec![legacy.clone()]);
160+
let runtime = RuntimeState::from_catalog(config, loaded).unwrap();
161+
let slot = model_slot_snapshot(&runtime, &legacy.model_id).unwrap();
162+
assert_eq!(slot.spec, legacy);
163+
}
164+
drop(state);
165+
fs::remove_dir_all(root).unwrap();
166+
}

‎crates/synapse-module/tests/skeleton_e2e.rs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -906,7 +906,7 @@ async fn probe_owned_gte_modernbert_certifies_against_family_reference() {
906906
"execution": "explicit",
907907
"model_id": "gte-modernbert-base-f16",
908908
"task": "embed",
909-
"pooling": "mean",
909+
"pooling": "cls",
910910
"normalize": true,
911911
"max_tokens": 512,
912912
"pin": true,

‎docs/wire-contract-v1.md‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -462,6 +462,12 @@ Operator examples: `ck-synapse-opctl lanes pause --minutes 45 --reason measureme
462462
(`not_certified`/`probe_required`) until the (machine, fingerprint) pair is
463463
certified. Lane-1 worker-backed `microllm.oneshot` routes retain their existing
464464
dispatch path and do not acquire the owned-lane certification gate.
465+
New free-form `owned-metal` registrations reject contradictory pooling with
466+
`invalid_request`: GTE ModernBERT embed requires `cls`, GTE rerank requires
467+
`mean`, and Qwen3 embed/rerank require `last` (omitted pooling still defaults
468+
to `mean` and is rejected when it contradicts the readout). Existing persisted
469+
registrations, including matching preload ids, retain their declarations and
470+
fingerprints.
465471

466472
### Catalog serving and migration
467473

0 commit comments

Comments
 (0)