use std::path::Path;
use serde::Deserialize;
use crate::binding::{
BINDING_VERSION, Binding, BuildMode, BuildOperation, CapabilityError, CoverageSemantics,
Operations, PruneConfig, validate_binding,
};
use crate::engine::Engine;
use crate::pipeline::{IngestTrigger, Source};
use crate::pipeline_store::{self, BindingConfigs};
use crate::workspace_store::StoreError;
#[derive(Debug, thiserror::Error)]
pub enum PipelineEditError {
#[error("engine has no workspace root — pipeline edits require a workspace-backed engine")]
NoWorkspaceRoot,
#[error("pipeline edit landed, but recording provenance failed: {0}")]
Provenance(String),
#[error("{primitive} '{key}' already exists")]
AlreadyExists {
primitive: &'static str,
key: String,
},
#[error("{primitive} '{key}' does not exist")]
NotFound {
primitive: &'static str,
key: String,
},
#[error("rename target {primitive} '{key}' already exists")]
RenameTargetExists {
primitive: &'static str,
key: String,
},
#[error("invalid {primitive} JSON: {message}")]
InvalidJson {
primitive: &'static str,
message: String,
},
#[error(
"binding '{key}' edit refused by validation: {}",
format_refusals(refusals)
)]
Capability {
key: String,
refusals: Vec<CapabilityError>,
},
#[error(transparent)]
Store(#[from] StoreError),
}
fn key(mem: &str, name: &str) -> String {
format!("{mem}/{name}")
}
fn format_refusals(refusals: &[CapabilityError]) -> String {
refusals
.iter()
.map(|r| r.to_string())
.collect::<Vec<_>>()
.join("; ")
}
fn binding_exists(c: &BindingConfigs, mem: &str, name: &str) -> bool {
c.bindings.iter().any(|r| r.mem == mem && r.name == name)
}
fn patch_field<'de, D, T>(deserializer: D) -> Result<Option<Option<T>>, D::Error>
where
D: serde::Deserializer<'de>,
T: serde::Deserialize<'de>,
{
Option::<T>::deserialize(deserializer).map(Some)
}
#[derive(Debug, Default, Deserialize)]
pub struct BindingPatch {
#[serde(default, deserialize_with = "patch_field")]
pub intent: Option<Option<String>>,
#[serde(default)]
pub sources: Option<Vec<Source>>,
#[serde(default)]
pub reference_mems: Option<Vec<String>>,
#[serde(default)]
pub destination_mem: Option<String>,
#[serde(default)]
pub deny_paths: Option<Vec<String>>,
#[serde(default)]
pub coverage_semantics: Option<CoverageSemantics>,
#[serde(default, deserialize_with = "patch_field")]
pub rules: Option<Option<serde_json::Value>>,
#[serde(default, deserialize_with = "patch_field")]
pub prune: Option<Option<PruneConfig>>,
#[serde(default)]
pub operations: Option<Operations>,
}
impl BindingPatch {
fn apply(self, binding: &mut Binding) {
if let Some(v) = self.intent {
binding.intent = v;
}
if let Some(v) = self.sources {
binding.sources = v;
}
if let Some(v) = self.reference_mems {
binding.reference_mems = v;
}
if let Some(v) = self.destination_mem {
binding.destination_mem = v;
}
if let Some(v) = self.deny_paths {
binding.deny_paths = v;
}
if let Some(v) = self.coverage_semantics {
binding.coverage_semantics = Some(v);
}
if let Some(v) = self.rules {
binding.rules = v;
}
if let Some(v) = self.prune {
binding.prune = v;
}
if let Some(v) = self.operations {
binding.operations = v;
}
}
}
fn default_binding_scaffold() -> Binding {
Binding {
version: BINDING_VERSION,
intent: None,
sources: Vec::new(),
reference_mems: Vec::new(),
destination_mem: String::new(),
deny_paths: Vec::new(),
coverage_semantics: None,
rules: None,
prune: None,
operations: Operations {
build: Some(BuildOperation {
mode: BuildMode::Discovery,
trigger: IngestTrigger::Loop,
batch_size: 20,
post_actions: None,
}),
sync: None,
verify: None,
},
}
}
pub fn add_binding_json(
root: &Path,
mem: &str,
name: &str,
patch_json: &str,
) -> Result<Binding, PipelineEditError> {
let configs = pipeline_store::load_pipeline_configs_strict(root)?;
if binding_exists(&configs, mem, name) {
return Err(PipelineEditError::AlreadyExists {
primitive: "projection",
key: key(mem, name),
});
}
let patch: BindingPatch = parse_json(patch_json, "projection")?;
let mut binding = default_binding_scaffold();
patch.apply(&mut binding);
if binding.destination_mem.is_empty() {
return Err(PipelineEditError::InvalidJson {
primitive: "projection",
message: "destination_mem is required".to_string(),
});
}
if let Err(refusals) = validate_binding(&binding) {
return Err(PipelineEditError::Capability {
key: key(mem, name),
refusals,
});
}
pipeline_store::write_binding(root, mem, name, &binding)?;
Ok(binding)
}
pub fn update_binding_json(
root: &Path,
mem: &str,
name: &str,
patch_json: &str,
) -> Result<Binding, PipelineEditError> {
let configs = pipeline_store::load_pipeline_configs_strict(root)?;
if !binding_exists(&configs, mem, name) {
return Err(PipelineEditError::NotFound {
primitive: "projection",
key: key(mem, name),
});
}
let patch: BindingPatch = parse_json(patch_json, "projection")?;
let existing = pipeline_store::read_binding(root, mem, name)?;
let mut patched = existing.clone();
patch.apply(&mut patched);
if let Err(refusals) = validate_binding(&patched) {
let before = validate_binding(&existing).err().unwrap_or_default();
let introduced: Vec<CapabilityError> = refusals
.into_iter()
.filter(|r| !before.contains(r))
.collect();
if !introduced.is_empty() {
return Err(PipelineEditError::Capability {
key: key(mem, name),
refusals: introduced,
});
}
}
pipeline_store::write_binding(root, mem, name, &patched)?;
Ok(patched)
}
pub fn delete_binding(root: &Path, mem: &str, name: &str) -> Result<(), PipelineEditError> {
let configs = pipeline_store::load_pipeline_configs_strict(root)?;
if !binding_exists(&configs, mem, name) {
return Err(PipelineEditError::NotFound {
primitive: "projection",
key: key(mem, name),
});
}
pipeline_store::delete_projection(root, mem, name)?;
Ok(())
}
pub fn rename_binding(
root: &Path,
mem: &str,
old: &str,
new: &str,
) -> Result<(), PipelineEditError> {
if old == new {
return Ok(());
}
let configs = pipeline_store::load_pipeline_configs_strict(root)?;
if !binding_exists(&configs, mem, old) {
return Err(PipelineEditError::NotFound {
primitive: "projection",
key: key(mem, old),
});
}
if binding_exists(&configs, mem, new) {
return Err(PipelineEditError::RenameTargetExists {
primitive: "projection",
key: key(mem, new),
});
}
pipeline_store::rename_projection(root, mem, old, new)?;
Ok(())
}
impl Engine {
fn pipeline_edit_root(&self) -> Result<std::path::PathBuf, PipelineEditError> {
self.workspace_root()
.map(Path::to_path_buf)
.ok_or(PipelineEditError::NoWorkspaceRoot)
}
fn refresh_pipeline_configs(&mut self, root: &Path) -> Result<(), PipelineEditError> {
self.set_pipeline_configs(pipeline_store::load_pipeline_configs_strict(root)?);
Ok(())
}
fn pipeline_provenance(
&self,
mem: &str,
kind: &str,
edits: &[(String, Option<Vec<u8>>)],
note: Option<&str>,
verb: &str,
) -> Result<(), PipelineEditError> {
self.record_pipeline_edit_provenance(mem, kind, edits, note, verb)
.map_err(|e| PipelineEditError::Provenance(e.to_string()))
}
pub fn add_projection_json(
&mut self,
mem: &str,
name: &str,
projection_json: &str,
note: Option<&str>,
) -> Result<(), PipelineEditError> {
let root = self.pipeline_edit_root()?;
let binding = add_binding_json(&root, mem, name, projection_json)?;
self.record_binding_edit(mem, name, &binding, &root, note, "add")
}
pub fn update_projection_json(
&mut self,
mem: &str,
name: &str,
projection_json: &str,
note: Option<&str>,
) -> Result<(), PipelineEditError> {
let root = self.pipeline_edit_root()?;
let binding = update_binding_json(&root, mem, name, projection_json)?;
self.record_binding_edit(mem, name, &binding, &root, note, "update")
}
pub fn delete_projection(
&mut self,
mem: &str,
name: &str,
note: Option<&str>,
) -> Result<(), PipelineEditError> {
let root = self.pipeline_edit_root()?;
delete_binding(&root, mem, name)?;
self.pipeline_provenance(
mem,
"projections",
&[(name.to_string(), None)],
note,
"delete",
)?;
self.refresh_pipeline_configs(&root)
}
pub fn rename_projection(
&mut self,
mem: &str,
old: &str,
new: &str,
note: Option<&str>,
) -> Result<(), PipelineEditError> {
let root = self.pipeline_edit_root()?;
rename_binding(&root, mem, old, new)?;
let bytes = self
.pipeline_configs()
.bindings
.iter()
.find(|r| r.mem == mem && r.name == old)
.map(|r| serde_json::to_vec_pretty(&r.config))
.transpose()
.map_err(|e| PipelineEditError::InvalidJson {
primitive: "config",
message: e.to_string(),
})?;
self.pipeline_provenance(
mem,
"projections",
&[(old.to_string(), None), (new.to_string(), bytes)],
note,
"rename",
)?;
self.refresh_pipeline_configs(&root)
}
fn record_binding_edit(
&mut self,
mem: &str,
name: &str,
binding: &Binding,
root: &Path,
note: Option<&str>,
verb: &str,
) -> Result<(), PipelineEditError> {
let bytes =
serde_json::to_vec_pretty(binding).map_err(|e| PipelineEditError::InvalidJson {
primitive: "config",
message: e.to_string(),
})?;
self.pipeline_provenance(
mem,
"projections",
&[(name.to_string(), Some(bytes))],
note,
verb,
)?;
self.refresh_pipeline_configs(root)
}
}
fn parse_json<T: serde::de::DeserializeOwned>(
json: &str,
primitive: &'static str,
) -> Result<T, PipelineEditError> {
serde_json::from_str(json).map_err(|e| PipelineEditError::InvalidJson {
primitive,
message: e.to_string(),
})
}
#[cfg(test)]
mod tests {
use super::*;
use crate::binding::{PruneGuarantee, SyncOperation};
use tempfile::TempDir;
const BASE_PAYLOAD: &str = r#"{
"intent": "i",
"sources": [{
"name": "f",
"type": "codebase",
"pointer": "../src",
"scope": [{ "path": "**/*.rs", "mode": "allow" }]
}],
"reference_mems": [],
"destination_mem": "v",
"rules": { "routing": "r" }
}"#;
fn full_binding_payload() -> &'static str {
r#"{
"intent": "i",
"sources": [{
"name": "f",
"type": "codebase",
"pointer": "../src",
"scope": [{ "path": "**/*.rs", "mode": "allow" }]
}],
"reference_mems": ["r"],
"destination_mem": "v",
"deny_paths": ["dev/**"],
"coverage_semantics": "curated",
"rules": { "routing": "r" },
"prune": { "guarantee": "never-clobber" },
"operations": {
"build": { "mode": "discovery", "trigger": "loop", "batch_size": 20 },
"sync": { "trigger": "manual", "batch_size": 20 },
"verify": { "trigger": "manual", "batch_size": 20 }
}
}"#
}
#[test]
fn add_binding_json_scaffolds_default_build() {
let tmp = TempDir::new().unwrap();
let root = tmp.path();
let b = add_binding_json(root, "v", "p", BASE_PAYLOAD).unwrap();
assert_eq!(b.version, BINDING_VERSION);
let build = b.operations.build.as_ref().unwrap();
assert_eq!(build.mode, BuildMode::Discovery);
assert_eq!(build.batch_size, 20);
assert!(b.operations.sync.is_none() && b.operations.verify.is_none());
assert_eq!(b.rules, Some(serde_json::json!({ "routing": "r" })));
assert_eq!(b.sources.len(), 1);
assert_eq!(b.sources[0].name, "f");
assert_eq!(pipeline_store::read_binding(root, "v", "p").unwrap(), b);
}
#[test]
fn add_binding_json_accepts_the_full_record() {
let tmp = TempDir::new().unwrap();
let root = tmp.path();
let b = add_binding_json(root, "v", "p", full_binding_payload()).unwrap();
assert_eq!(
b.operations.build.as_ref().unwrap().mode,
BuildMode::Discovery
);
assert_eq!(b.operations.sync.as_ref().unwrap().batch_size, 20);
assert!(b.operations.verify.is_some());
assert_eq!(b.deny_paths, vec!["dev/**"]);
assert_eq!(b.coverage_semantics, Some(CoverageSemantics::Curated));
assert_eq!(
b.prune.as_ref().unwrap().guarantee,
PruneGuarantee::NeverClobber
);
}
#[test]
fn add_binding_json_refuses_duplicate() {
let tmp = TempDir::new().unwrap();
let root = tmp.path();
add_binding_json(root, "v", "p", BASE_PAYLOAD).unwrap();
let err = add_binding_json(root, "v", "p", BASE_PAYLOAD).unwrap_err();
assert!(
matches!(err, PipelineEditError::AlreadyExists { .. }),
"got {err:?}"
);
}
#[test]
fn add_binding_json_requires_destination_mem() {
let tmp = TempDir::new().unwrap();
let err = add_binding_json(tmp.path(), "v", "p", r#"{"intent":"i"}"#).unwrap_err();
match err {
PipelineEditError::InvalidJson { message, .. } => {
assert!(message.contains("destination_mem"), "got: {message}")
}
other => panic!("expected InvalidJson, got {other:?}"),
}
}
#[test]
fn add_binding_json_refuses_duplicate_source_names() {
let tmp = TempDir::new().unwrap();
let root = tmp.path();
let err = add_binding_json(
root,
"v",
"p",
r#"{
"sources": [
{ "name": "dup", "type": "codebase", "pointer": "../a" },
{ "name": "dup", "type": "codebase", "pointer": "../b" }
],
"destination_mem": "v"
}"#,
)
.unwrap_err();
match err {
PipelineEditError::Capability { refusals, .. } => {
assert!(
refusals.iter().any(|r| matches!(
r,
CapabilityError::DuplicateSourceName { name } if name == "dup"
)),
"expected DuplicateSourceName, got {refusals:?}"
);
}
other => panic!("expected Capability, got {other:?}"),
}
assert!(!root.join(".memstead/projections/v/p.json").exists());
}
#[test]
fn add_binding_json_refuses_capability_violation() {
let tmp = TempDir::new().unwrap();
let root = tmp.path();
let err = add_binding_json(
root,
"v",
"p",
r#"{
"sources": [{ "name": "wf", "type": "web", "pointer": "https://example.com" }],
"destination_mem": "v",
"operations": {
"build": { "mode": "discovery", "trigger": "loop", "batch_size": 20 },
"sync": { "trigger": "manual", "batch_size": 20 }
}
}"#,
)
.unwrap_err();
match &err {
PipelineEditError::Capability { refusals, .. } => {
assert!(
refusals.iter().any(|r| matches!(
r,
CapabilityError::OperationOutOfScope { operation, .. } if *operation == "sync"
)),
"expected an OperationOutOfScope(sync) refusal, got {refusals:?}"
);
}
other => panic!("expected Capability, got {other:?}"),
}
assert!(pipeline_store::read_binding(root, "v", "p").is_err());
}
#[test]
fn update_binding_json_patch_preserves_untouched_fields() {
let tmp = TempDir::new().unwrap();
let root = tmp.path();
let before = add_binding_json(root, "v", "p", full_binding_payload()).unwrap();
let after = update_binding_json(root, "v", "p", r#"{"intent":"new"}"#).unwrap();
assert_eq!(after.intent.as_deref(), Some("new"));
assert_eq!(after.version, before.version);
assert_eq!(after.sources, before.sources);
assert_eq!(after.reference_mems, before.reference_mems);
assert_eq!(after.destination_mem, before.destination_mem);
assert_eq!(after.deny_paths, before.deny_paths);
assert_eq!(after.coverage_semantics, before.coverage_semantics);
assert_eq!(after.rules, before.rules);
assert_eq!(after.prune, before.prune);
assert_eq!(after.operations, before.operations);
}
#[test]
fn update_binding_json_null_clears_and_absence_preserves() {
let tmp = TempDir::new().unwrap();
let root = tmp.path();
add_binding_json(root, "v", "p", full_binding_payload()).unwrap();
let untouched = update_binding_json(root, "v", "p", r#"{"deny_paths":[]}"#).unwrap();
assert!(untouched.rules.is_some() && untouched.prune.is_some());
assert_eq!(untouched.intent.as_deref(), Some("i"));
assert!(untouched.deny_paths.is_empty());
let cleared =
update_binding_json(root, "v", "p", r#"{"rules": null, "prune": null}"#).unwrap();
assert!(cleared.rules.is_none() && cleared.prune.is_none());
assert_eq!(cleared.intent.as_deref(), Some("i"), "intent untouched");
assert_eq!(
pipeline_store::read_binding(root, "v", "p").unwrap(),
cleared
);
}
#[test]
fn update_binding_json_replaces_whole_blocks() {
let tmp = TempDir::new().unwrap();
let root = tmp.path();
add_binding_json(root, "v", "p", full_binding_payload()).unwrap();
let after = update_binding_json(
root,
"v",
"p",
r#"{"operations": { "build": { "mode": "discovery", "trigger": "manual", "batch_size": 9 } }}"#,
)
.unwrap();
assert_eq!(after.operations.build.as_ref().unwrap().batch_size, 9);
assert!(
after.operations.sync.is_none(),
"sync removed with the block"
);
assert!(
after.operations.verify.is_none(),
"verify removed with the block"
);
assert_eq!(after.rules, Some(serde_json::json!({ "routing": "r" })));
let after = update_binding_json(
root,
"v",
"p",
r#"{"sources": [{ "name": "g", "type": "filesystem", "pointer": "../docs" }]}"#,
)
.unwrap();
assert_eq!(after.sources.len(), 1);
assert_eq!(after.sources[0].name, "g");
}
#[test]
fn update_binding_json_refuses_introduced_capability_violation() {
let tmp = TempDir::new().unwrap();
let root = tmp.path();
let before = add_binding_json(
root,
"v",
"p",
r#"{
"sources": [{ "name": "wf", "type": "web", "pointer": "https://example.com" }],
"destination_mem": "v"
}"#,
)
.unwrap();
let err = update_binding_json(
root,
"v",
"p",
r#"{"operations": {
"build": { "mode": "discovery", "trigger": "loop", "batch_size": 20 },
"sync": { "trigger": "manual", "batch_size": 20 }
}}"#,
)
.unwrap_err();
assert!(
matches!(err, PipelineEditError::Capability { .. }),
"got {err:?}"
);
assert_eq!(
pipeline_store::read_binding(root, "v", "p").unwrap(),
before,
"record unchanged on refusal"
);
}
#[test]
fn update_binding_json_allows_edit_despite_preexisting_refusal() {
let tmp = TempDir::new().unwrap();
let root = tmp.path();
let broken = Binding {
version: BINDING_VERSION,
intent: None,
sources: vec![Source {
name: "wf".to_string(),
medium_type: crate::pipeline::MediumType::Web,
pointer: "https://example.com".to_string(),
change_detection: None,
scope: vec![],
engagement: None,
preparation: None,
}],
reference_mems: vec![],
destination_mem: "v".to_string(),
deny_paths: vec![],
coverage_semantics: None,
rules: None,
prune: None,
operations: Operations {
build: Some(BuildOperation {
mode: BuildMode::Discovery,
trigger: IngestTrigger::Loop,
batch_size: 20,
post_actions: None,
}),
sync: Some(SyncOperation {
trigger: IngestTrigger::Manual,
batch_size: 20,
}),
verify: None,
},
};
pipeline_store::write_binding(root, "v", "p", &broken).unwrap();
let after = update_binding_json(root, "v", "p", r#"{"intent":"fixed"}"#).unwrap();
assert_eq!(after.intent.as_deref(), Some("fixed"));
assert!(
after.operations.sync.is_some(),
"pre-existing sync survives"
);
let err = update_binding_json(
root,
"v",
"p",
r#"{"sources": [
{ "name": "dup", "type": "web", "pointer": "https://a" },
{ "name": "dup", "type": "web", "pointer": "https://b" }
]}"#,
)
.unwrap_err();
assert!(
matches!(err, PipelineEditError::Capability { .. }),
"got {err:?}"
);
}
#[test]
fn update_binding_json_refuses_missing_record() {
let tmp = TempDir::new().unwrap();
let err = update_binding_json(tmp.path(), "v", "p", r#"{"intent":"x"}"#).unwrap_err();
assert!(
matches!(err, PipelineEditError::NotFound { .. }),
"got {err:?}"
);
}
#[test]
fn update_binding_json_ignores_unknown_keys_and_version() {
let tmp = TempDir::new().unwrap();
let root = tmp.path();
add_binding_json(root, "v", "p", BASE_PAYLOAD).unwrap();
let after = update_binding_json(
root,
"v",
"p",
r#"{"version": 99, "future_key": { "x": 1 }, "intent": "i2"}"#,
)
.unwrap();
assert_eq!(
after.version, BINDING_VERSION,
"version stays engine-managed"
);
assert_eq!(after.intent.as_deref(), Some("i2"));
}
#[test]
fn delete_binding_removes_the_record() {
let tmp = TempDir::new().unwrap();
let root = tmp.path();
add_binding_json(root, "v", "p", BASE_PAYLOAD).unwrap();
delete_binding(root, "v", "p").unwrap();
assert!(!root.join(".memstead/projections/v/p.json").exists());
let err = delete_binding(root, "v", "p").unwrap_err();
assert!(
matches!(err, PipelineEditError::NotFound { .. }),
"got {err:?}"
);
}
#[test]
fn rename_binding_moves_the_record() {
let tmp = TempDir::new().unwrap();
let root = tmp.path();
let created = add_binding_json(root, "v", "old", BASE_PAYLOAD).unwrap();
rename_binding(root, "v", "old", "new").unwrap();
assert!(!root.join(".memstead/projections/v/old.json").exists());
assert_eq!(
pipeline_store::read_binding(root, "v", "new").unwrap(),
created
);
}
#[test]
fn rename_binding_refuses_existing_target_and_missing_source() {
let tmp = TempDir::new().unwrap();
let root = tmp.path();
add_binding_json(root, "v", "a", BASE_PAYLOAD).unwrap();
add_binding_json(root, "v", "b", BASE_PAYLOAD).unwrap();
let err = rename_binding(root, "v", "a", "b").unwrap_err();
assert!(
matches!(err, PipelineEditError::RenameTargetExists { .. }),
"got {err:?}"
);
let err = rename_binding(root, "v", "missing", "c").unwrap_err();
assert!(
matches!(err, PipelineEditError::NotFound { .. }),
"got {err:?}"
);
rename_binding(root, "v", "a", "a").unwrap();
}
#[test]
fn editing_a_pre_v2_store_refuses_with_migrate_pointer() {
let tmp = TempDir::new().unwrap();
let root = tmp.path();
let dir = root.join(".memstead/projections/v");
std::fs::create_dir_all(&dir).unwrap();
std::fs::write(
dir.join("p.json"),
br#"{"version": 1, "source_facets": ["f"], "destination_mem": "v", "operations": {}}"#,
)
.unwrap();
let err = add_binding_json(root, "v", "q", BASE_PAYLOAD).unwrap_err();
match err {
PipelineEditError::Store(StoreError::LegacyProjectionStore { .. }) => {}
other => panic!("expected LegacyProjectionStore, got {other:?}"),
}
}
}