use std::collections::BTreeSet;
use crate::model::{BindingsDef, ParsedOp, Workload, WorkloadPhase};
pub fn bind_implementation(
blueprint: &mut Workload,
implementation: Workload,
) -> Result<(), String> {
let provided_global: BTreeSet<String> = blueprint
.declared_params
.iter()
.cloned()
.chain(bindings_wire_names(&blueprint.bindings))
.collect();
let mut unbound: BTreeSet<String> = BTreeSet::new();
for (phase_name, phase) in &blueprint.phases {
for op in &phase.ops {
if op.abstract_interface.is_some() {
unbound.insert(format!("{phase_name}.{}", op.name));
}
}
}
for (phase_name, impl_phase) in implementation.phases {
let offending = scaffolding_fields_set(&impl_phase);
if !offending.is_empty() {
return Err(format!(
"implementation phase '{phase_name}' declares scaffolding \
fields [{}] — scaffolding belongs to the blueprint; an \
implementation phase carries only `ops:`",
offending.join(", ")
));
}
let Some(blueprint_phase) = blueprint.phases.get_mut(&phase_name) else {
return Err(format!(
"implementation provides phase '{phase_name}', but the \
blueprint declares no such phase"
));
};
for impl_op in impl_phase.ops {
let slot_key = format!("{phase_name}.{}", impl_op.name);
let Some(slot) = blueprint_phase
.ops
.iter_mut()
.find(|op| op.name == impl_op.name)
else {
return Err(format!(
"implementation op '{slot_key}' names no op in the \
blueprint phase"
));
};
if slot.abstract_interface.is_none() {
return Err(format!(
"implementation op '{slot_key}' targets a CONCRETE \
blueprint op — only abstract slots accept bindings"
));
}
if !unbound.remove(&slot_key) {
return Err(format!("implementation binds slot '{slot_key}' twice"));
}
bind_slot(slot, impl_op, &slot_key)?;
let mut provided = provided_global.clone();
provided.extend(bindings_wire_names(&blueprint_phase.bindings));
if let Some(iface) = slot.abstract_interface.as_ref() {
for need in iface.needs.keys() {
if !provided.contains(need) {
return Err(format!(
"slot '{slot_key}' declares need '{need}', but \
the blueprint provides no param or binding \
by that name"
));
}
}
}
}
}
if !unbound.is_empty() {
return Err(format!(
"abstract slot(s) [{}] remain unbound — the implementation \
must cover every abstract op",
unbound.into_iter().collect::<Vec<_>>().join(", ")
));
}
let blueprint_declared: BTreeSet<String> = blueprint.declared_params.iter().cloned().collect();
let impl_declared: Vec<String> = implementation.declared_params;
for k in impl_declared {
if let Some(v) = implementation.params.get(&k) {
blueprint.params.insert(k.clone(), v.clone());
}
if !blueprint_declared.contains(&k) {
blueprint.declared_params.push(k);
}
}
concat_bindings(&mut blueprint.bindings, implementation.bindings)
.map_err(|e| format!("workload-level bindings: {e}"))?;
Ok(())
}
fn bind_slot(slot: &mut ParsedOp, impl_op: ParsedOp, slot_key: &str) -> Result<(), String> {
if impl_op.condition.is_some()
|| impl_op.delay.is_some()
|| !impl_op.metrics.is_empty()
|| impl_op.traverse.is_some()
|| impl_op.wrappers.is_some()
|| impl_op.while_cond.is_some()
|| impl_op.rate.is_some()
|| !matches!(impl_op.daemon, crate::model::DaemonSpec::Disabled)
{
return Err(format!(
"implementation op '{slot_key}' sets op semantics \
(if/delay/metrics/traverse/wrappers/while/rate/\
daemon) — those belong to the blueprint slot"
));
}
if impl_op.abstract_interface.is_some() {
return Err(format!(
"implementation op '{slot_key}' declares `abstract:` — an \
implementation provides bodies, not interfaces"
));
}
for (k, v) in impl_op.op {
if slot.op.contains_key(&k) {
return Err(format!(
"slot '{slot_key}': op field '{k}' declared by BOTH the \
blueprint slot and the implementation — the blueprint \
is authoritative; remove one"
));
}
slot.op.insert(k, v);
}
for (k, v) in impl_op.params {
slot.params.insert(k, v);
}
for (k, v) in impl_op.tags {
match slot.tags.get(&k) {
Some(existing) if existing != &v => {
return Err(format!(
"slot '{slot_key}': tag '{k}' has conflicting values \
('{existing}' vs '{v}')"
));
}
_ => {
slot.tags.insert(k, v);
}
}
}
if !impl_op.captures.is_empty() {
let existing: BTreeSet<&str> = slot.captures.iter().map(|c| c.as_name.as_str()).collect();
for cap in &impl_op.captures {
if existing.contains(cap.as_name.as_str()) {
return Err(format!(
"slot '{slot_key}': capture '{}' declared by both \
sides",
cap.as_name
));
}
}
slot.captures.extend(impl_op.captures);
}
concat_bindings(&mut slot.bindings, impl_op.bindings)
.map_err(|e| format!("slot '{slot_key}' bindings: {e}"))?;
if let Some(iface) = slot.abstract_interface.as_ref() {
let blueprint_declared = result_binding_names(slot.result.as_ref());
for results_name in iface.results.keys() {
if blueprint_declared.contains(results_name.as_str()) {
return Err(format!(
"slot '{slot_key}': `results:` wire '{results_name}' \
also has a blueprint-side `result:` binding — the \
projection path is protocol matter; the \
implementation declares it"
));
}
}
}
if let Some(impl_result) = impl_op.result {
let blueprint_declared = result_binding_names(slot.result.as_ref());
for name in result_binding_names(Some(&impl_result)) {
if blueprint_declared.contains(name.as_str()) {
return Err(format!(
"slot '{slot_key}': result binding '{name}' declared \
by BOTH the blueprint slot and the implementation — \
remove one"
));
}
}
slot.result = Some(match slot.result.take() {
None => impl_result,
Some(existing) => crate::model::ResultSpec::List(vec![existing, impl_result]),
});
}
if let Some(iface) = slot.abstract_interface.as_ref() {
let delivered: BTreeSet<&str> = slot.captures.iter().map(|c| c.as_name.as_str()).collect();
for yield_name in iface.yields.keys() {
if !delivered.contains(yield_name.as_str()) {
return Err(format!(
"slot '{slot_key}' promises yield '{yield_name}', \
but the bound implementation captures [{}] — add a \
capture (`[{yield_name}]` or `... as {yield_name}`)",
delivered.into_iter().collect::<Vec<_>>().join(", ")
));
}
}
let projected = result_binding_names(slot.result.as_ref());
for results_name in iface.results.keys() {
if !projected.contains(results_name.as_str()) {
return Err(format!(
"slot '{slot_key}' promises results wire \
'{results_name}', but the bound implementation \
declares result bindings [{}] — add a `result:` \
entry (`{results_name}: <path-expr>`, e.g. \
`{results_name}: rows[*].column`)",
projected.iter().cloned().collect::<Vec<_>>().join(", ")
));
}
}
}
slot.interface_bound = true;
Ok(())
}
fn result_binding_names(spec: Option<&crate::model::ResultSpec>) -> BTreeSet<String> {
let mut out = BTreeSet::new();
if let Some(spec) = spec {
spec.walk_fragments(|frag| match frag {
crate::model::ResultFragment::Named { name, .. } => {
out.insert(name.to_string());
}
crate::model::ResultFragment::Source(source) => {
for line in source.lines() {
if let Some((lhs, _)) = line.trim().split_once(":=") {
out.insert(lhs.trim().to_string());
}
}
}
});
}
out
}
fn scaffolding_fields_set(phase: &WorkloadPhase) -> Vec<&'static str> {
let mut out = Vec::new();
if phase.cycles.is_some() {
out.push("cycles");
}
if phase.concurrency.is_some() {
out.push("concurrency");
}
if phase.rate.is_some() {
out.push("rate");
}
if phase.daemon {
out.push("daemon");
}
if phase.adapter.is_some() {
out.push("adapter");
}
if phase.errors.is_some() {
out.push("errors");
}
if phase.tries.is_some() {
out.push("tries");
}
if phase.tries_backoff.is_some() {
out.push("tries_backoff");
}
if phase.interval.is_some() {
out.push("interval");
}
if phase.repeat.is_some() {
out.push("repeat");
}
if phase.error_rate_max.is_some() {
out.push("error_rate_max");
}
if phase.timeout.is_some() {
out.push("timeout");
}
if !phase.stop_when.is_empty() {
out.push("stop_when");
}
if phase.tags.is_some() {
out.push("tags");
}
if phase.for_each.is_some() {
out.push("for_each");
}
if phase.continue_if.is_some() {
out.push("continue_if");
}
if phase.loop_scope.is_some() {
out.push("loop_scope");
}
if phase.iter_scope.is_some() {
out.push("iter_scope");
}
if phase.checkpoint.is_some() {
out.push("checkpoint");
}
if !phase.status_metrics.is_empty() {
out.push("status_metrics");
}
if !phase.bindings.is_empty() {
out.push("bindings");
}
if !phase.metrics.is_empty() {
out.push("metrics");
}
if !phase.dimensions.is_empty() {
out.push("dimensions");
}
out
}
fn bindings_wire_names(bindings: &BindingsDef) -> Vec<String> {
match bindings {
BindingsDef::PolydatSource(s) => crate::inline::binding_wire_names(s),
BindingsDef::Map(m) => m.keys().cloned().collect(),
}
}
fn concat_bindings(base: &mut BindingsDef, extra: BindingsDef) -> Result<(), String> {
if extra.is_empty() {
return Ok(());
}
if base.is_empty() {
*base = extra;
return Ok(());
}
match (&mut *base, extra) {
(BindingsDef::PolydatSource(b), BindingsDef::PolydatSource(e)) => {
if !b.ends_with('\n') {
b.push('\n');
}
b.push_str(&e);
Ok(())
}
(BindingsDef::Map(b), BindingsDef::Map(e)) => {
for (k, v) in e {
if b.contains_key(&k) {
return Err(format!("binding '{k}' declared by both sides"));
}
b.insert(k, v);
}
Ok(())
}
_ => Err("mixed bindings forms (map vs polydat source) cannot \
be concatenated — use one form on both sides"
.into()),
}
}
pub fn unbound_abstract_slots(workload: &Workload) -> Vec<String> {
let mut out = Vec::new();
for (phase_name, phase) in &workload.phases {
for op in &phase.ops {
if op.abstract_interface.is_some() && !op.interface_bound {
out.push(format!("{phase_name}.{}", op.name));
}
}
}
out.sort();
out
}
#[cfg(test)]
mod tests {
use super::*;
use std::collections::HashMap;
fn parse(yaml: &str) -> Workload {
crate::parse::parse_workload(yaml, &HashMap::new()).expect("parse test workload")
}
const BLUEPRINT: &str = r#"
params:
suite_k: "10"
bindings: |
query_vector := "0.1,0.2"
phases:
probe:
cycles: 4
concurrency: 2
ops:
search:
abstract:
needs:
query_vector: String
suite_k: u64
yields:
key: String
"#;
const IMPL_OK: &str = r#"
implements: blueprint
phases:
probe:
ops:
search:
stmt: "SEARCH {query_vector} LIMIT {suite_k}"
captures: "[key]"
"#;
#[test]
fn binds_and_marks_the_slot() {
let mut blueprint = parse(BLUEPRINT);
let implementation = parse(IMPL_OK);
bind_implementation(&mut blueprint, implementation).unwrap();
let op = &blueprint.phases["probe"].ops[0];
assert!(op.interface_bound);
assert!(op.op.contains_key("stmt"));
assert_eq!(op.captures.len(), 1);
assert!(unbound_abstract_slots(&blueprint).is_empty());
}
#[test]
fn impl_params_add_knobs_and_redefault_blueprint_params() {
let mut blueprint = parse(BLUEPRINT);
let implementation = parse(
r#"
implements: blueprint
params:
suite_k: "25"
knob: "x"
phases:
probe:
ops:
search:
stmt: "SEARCH {query_vector} LIMIT {suite_k}"
captures: "[key]"
"#,
);
bind_implementation(&mut blueprint, implementation).unwrap();
assert_eq!(blueprint.params["suite_k"], "25");
assert_eq!(blueprint.params["knob"], "x");
assert_eq!(
blueprint
.declared_params
.iter()
.filter(|k| k.as_str() == "suite_k")
.count(),
1,
"re-defaulting must not re-declare"
);
assert!(blueprint.declared_params.iter().any(|k| k == "knob"));
}
#[test]
fn missing_yield_capture_is_named() {
let mut blueprint = parse(BLUEPRINT);
let implementation = parse(
r#"
implements: blueprint
phases:
probe:
ops:
search:
stmt: "SEARCH"
"#,
);
let err = bind_implementation(&mut blueprint, implementation).unwrap_err();
assert!(err.contains("yield 'key'"), "err: {err}");
}
#[test]
fn unknown_slot_and_uncovered_slot_are_errors() {
let mut blueprint = parse(BLUEPRINT);
let implementation = parse(
r#"
implements: blueprint
phases:
probe:
ops:
wrong_name:
stmt: "X"
"#,
);
let err = bind_implementation(&mut blueprint, implementation).unwrap_err();
assert!(err.contains("wrong_name"), "err: {err}");
let mut blueprint = parse(BLUEPRINT);
let err =
bind_implementation(&mut blueprint, parse("implements: blueprint\n")).unwrap_err();
assert!(err.contains("remain unbound"), "err: {err}");
}
#[test]
fn scaffolding_on_implementation_phase_is_rejected() {
let mut blueprint = parse(BLUEPRINT);
let implementation = parse(
r#"
implements: blueprint
phases:
probe:
cycles: 99
ops:
search:
stmt: "X"
captures: "[key]"
"#,
);
let err = bind_implementation(&mut blueprint, implementation).unwrap_err();
assert!(err.contains("scaffolding"), "err: {err}");
}
#[test]
fn undeclared_need_is_named() {
let mut blueprint = parse(
r#"
phases:
probe:
cycles: 1
ops:
search:
abstract:
needs:
not_provided: u64
"#,
);
let err = bind_implementation(
&mut blueprint,
parse(
r#"
implements: blueprint
phases:
probe:
ops:
search:
stmt: "X"
"#,
),
)
.unwrap_err();
assert!(err.contains("need 'not_provided'"), "err: {err}");
}
#[test]
fn op_field_collision_is_an_error() {
let mut blueprint = parse(
r#"
phases:
probe:
cycles: 1
ops:
search:
stmt: "BLUEPRINT SIDE"
abstract:
yields:
key: String
"#,
);
let err = bind_implementation(
&mut blueprint,
parse(
r#"
implements: blueprint
phases:
probe:
ops:
search:
stmt: "IMPL SIDE"
captures: "[key]"
"#,
),
)
.unwrap_err();
assert!(err.contains("'stmt'") && err.contains("BOTH"), "err: {err}");
}
#[test]
fn unbound_slots_are_reported() {
let blueprint = parse(BLUEPRINT);
assert_eq!(unbound_abstract_slots(&blueprint), vec!["probe.search"]);
}
const RESULTS_BLUEPRINT: &str = r#"
phases:
probe:
cycles: 1
ops:
search:
abstract:
results:
keys: vec_i64
"#;
#[test]
fn results_wire_delivered_by_impl_result_binding() {
let mut blueprint = parse(RESULTS_BLUEPRINT);
bind_implementation(
&mut blueprint,
parse(
r#"
implements: blueprint
phases:
probe:
ops:
search:
stmt: "SEARCH"
result:
keys: "rows[*].key"
"#,
),
)
.unwrap();
let op = &blueprint.phases["probe"].ops[0];
assert!(op.interface_bound);
assert!(result_binding_names(op.result.as_ref()).contains("keys"));
}
#[test]
fn missing_results_binding_is_named_with_remedy() {
let mut blueprint = parse(RESULTS_BLUEPRINT);
let err = bind_implementation(
&mut blueprint,
parse(
r#"
implements: blueprint
phases:
probe:
ops:
search:
stmt: "SEARCH"
"#,
),
)
.unwrap_err();
assert!(
err.contains("results wire") && err.contains("keys") && err.contains("path-expr"),
"err: {err}"
);
}
#[test]
fn blueprint_side_path_for_results_wire_is_a_collision() {
let mut blueprint = parse(
r#"
phases:
probe:
cycles: 1
ops:
search:
result:
keys: "rows[*].key"
abstract:
results:
keys: vec_i64
"#,
);
let err = bind_implementation(
&mut blueprint,
parse(
r#"
implements: blueprint
phases:
probe:
ops:
search:
stmt: "SEARCH"
result:
keys: "other[*].id"
"#,
),
)
.unwrap_err();
assert!(err.contains("protocol matter"), "err: {err}");
}
#[test]
fn result_binding_name_collision_across_sides_is_an_error() {
let mut blueprint = parse(
r#"
phases:
probe:
cycles: 1
ops:
search:
result:
row_count: count
abstract:
results:
keys: vec_i64
"#,
);
let err = bind_implementation(
&mut blueprint,
parse(
r#"
implements: blueprint
phases:
probe:
ops:
search:
stmt: "SEARCH"
result:
row_count: count
keys: "rows[*].key"
"#,
),
)
.unwrap_err();
assert!(
err.contains("row_count") && err.contains("BOTH"),
"err: {err}"
);
}
}