use std::collections::BTreeMap;
use std::path::Path;
use serde::Serialize;
use crate::Engine;
use crate::binding::CoverageSemantics;
use crate::ingest::advance::read_advance_store;
use crate::ingest::cursor::source_moved;
use crate::ingest::findings::{FindingClass, current_findings};
use crate::ingest::render::mem_predates_binding;
use crate::ingest::resolve::{
ChangeStrategy, ResolvedIngest, ResolvedSource, resolve_binding_run, resolve_change_strategy,
};
use crate::pipeline_store::load_pipeline_configs;
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct FacetState {
pub synced: Option<String>,
pub verified: Option<String>,
pub signal: String,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct AdvanceCounts {
pub pending: usize,
pub disposed: usize,
}
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize)]
pub struct FindingCounts {
pub unresolvable: usize,
pub drifted: usize,
pub uncovered: usize,
pub queued: usize,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct ProjectionStatus {
pub binding: String,
pub destination_mem: String,
pub operations: Vec<String>,
pub state: BTreeMap<String, FacetState>,
pub advance: AdvanceCounts,
pub verdict: RollupVerdict,
pub source_moved: bool,
pub findings: FindingCounts,
}
struct BindingResolution {
onboarding: bool,
source_moved: bool,
findings: FindingCounts,
has_action: bool,
}
impl BindingResolution {
fn verdict(&self) -> RollupVerdict {
if self.onboarding {
RollupVerdict::Onboarding
} else if self.has_action {
RollupVerdict::ActionNeeded
} else {
RollupVerdict::Clean
}
}
}
fn resolve_binding_status(
engine: &Engine,
workspace_root: &Path,
binding: &crate::binding::Binding,
resolved: &ResolvedIngest,
) -> BindingResolution {
if mem_predates_binding(engine, resolved) {
return BindingResolution {
onboarding: true,
source_moved: false,
findings: FindingCounts::default(),
has_action: false,
};
}
let source_moved = source_moved(engine, resolved, workspace_root);
let mut findings = FindingCounts::default();
if let Ok((_key, list)) = current_findings(engine, workspace_root, binding, resolved) {
for f in &list {
match f.class {
FindingClass::UnresolvableAnchor => findings.unresolvable += 1,
FindingClass::Drifted | FindingClass::Wrong => findings.drifted += 1,
FindingClass::Uncovered => findings.uncovered += 1,
FindingClass::QueuedForAdjudication => findings.queued += 1,
}
}
}
let uncovered_counts = findings.uncovered > 0
&& matches!(
crate::binding::effective_coverage_semantics(binding).value,
CoverageSemantics::Exhaustive
);
let has_action = source_moved
|| findings.unresolvable > 0
|| findings.drifted > 0
|| uncovered_counts
|| findings.queued > 0;
BindingResolution {
onboarding: false,
source_moved,
findings,
has_action,
}
}
fn signal_of(strategy: ChangeStrategy) -> &'static str {
match strategy {
ChangeStrategy::None => "none",
ChangeStrategy::Git => "git",
ChangeStrategy::Mtime => "mtime",
ChangeStrategy::Graph => "graph",
}
}
pub fn projection_status(engine: &Engine, workspace_root: &Path) -> Vec<ProjectionStatus> {
let Ok(configs) = load_pipeline_configs(workspace_root) else {
return Vec::new();
};
let mut out = Vec::with_capacity(configs.bindings.len());
for record in &configs.bindings {
let binding_id = format!("{}/{}", record.mem, record.name);
let binding = &record.config;
let mut operations = Vec::new();
if binding.operations.build.is_some() {
operations.push("build".to_string());
}
if binding.operations.sync.is_some() {
operations.push("sync".to_string());
}
if binding.operations.verify.is_some() {
operations.push("verify".to_string());
}
let sync_state = engine
.mem_config_for(&binding.destination_mem)
.map(|c| c.sync_state.clone())
.unwrap_or_default();
let mut state = BTreeMap::new();
let mut resolution: Option<BindingResolution> = None;
if let Ok(resolved) = resolve_binding_run(&binding_id, binding) {
resolution = Some(resolve_binding_status(
engine,
workspace_root,
binding,
&resolved,
));
for source in &resolved.sources {
let (facet, signal) = match source {
ResolvedSource::Primary(p) => (
p.name.clone(),
signal_of(resolve_change_strategy(p, workspace_root)).to_string(),
),
ResolvedSource::Reference { mem } => (mem.clone(), "graph".to_string()),
};
let synced = sync_state
.get(&format!("{binding_id}/{facet}#synced"))
.cloned();
let verified = sync_state
.get(&format!("{binding_id}/{facet}#verified"))
.cloned();
state.insert(
facet,
FacetState {
synced,
verified,
signal,
},
);
}
}
let advance = match read_advance_store(workspace_root, &record.mem, &record.name) {
Ok(Some(s)) => AdvanceCounts {
pending: s.pending(),
disposed: s.disposed(),
},
_ => AdvanceCounts {
pending: 0,
disposed: 0,
},
};
let (verdict, source_moved, findings) = match &resolution {
Some(r) => (r.verdict(), r.source_moved, r.findings),
None => (RollupVerdict::Clean, false, FindingCounts::default()),
};
out.push(ProjectionStatus {
binding: binding_id,
destination_mem: binding.destination_mem.clone(),
operations,
state,
advance,
verdict,
source_moved,
findings,
});
}
out
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "kebab-case")]
pub enum RollupVerdict {
Clean,
Onboarding,
ActionNeeded,
}
impl RollupVerdict {
pub fn as_wire(&self) -> &'static str {
match self {
RollupVerdict::Clean => "clean",
RollupVerdict::Onboarding => "onboarding",
RollupVerdict::ActionNeeded => "action-needed",
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct Rollup {
pub verdict: RollupVerdict,
pub headline: String,
pub actions: Vec<String>,
}
impl Default for Rollup {
fn default() -> Self {
Rollup {
verdict: RollupVerdict::Clean,
headline: "No projection bindings declared.".to_string(),
actions: Vec::new(),
}
}
}
struct Candidate {
severity: u8,
text: String,
}
pub fn projection_rollup(engine: &Engine, workspace_root: &Path) -> Rollup {
let Ok(configs) = load_pipeline_configs(workspace_root) else {
return Rollup::default();
};
if configs.bindings.is_empty() {
return Rollup::default();
}
let total = configs.bindings.len();
let mut candidates: Vec<Candidate> = Vec::new();
let mut action_bindings = 0usize;
let mut onboarding_bindings = 0usize;
for record in &configs.bindings {
let binding_id = format!("{}/{}", record.mem, record.name);
let binding = &record.config;
let Ok(resolved) = resolve_binding_run(&binding_id, binding) else {
continue;
};
let resolution = resolve_binding_status(engine, workspace_root, binding, &resolved);
if resolution.onboarding {
onboarding_bindings += 1;
candidates.push(Candidate {
severity: 1,
text: format!(
"`{binding_id}` predates its binding — 0% anchored is expected; run \
`memstead projection sync {binding_id}` for a first-sync backfill"
),
});
continue;
}
if resolution.source_moved {
candidates.push(Candidate {
severity: 4,
text: format!(
"`{binding_id}` source moved since the last sync — run `memstead projection \
sync {binding_id}`"
),
});
}
let FindingCounts {
unresolvable,
drifted,
uncovered,
queued,
} = resolution.findings;
if unresolvable > 0 {
candidates.push(Candidate {
severity: 6,
text: format!(
"{unresolvable} entit{} in `{binding_id}` describe source that no longer \
exists — run `memstead projection sync {binding_id}`",
if unresolvable == 1 { "y" } else { "ies" }
),
});
}
if drifted > 0 {
candidates.push(Candidate {
severity: 5,
text: format!(
"{drifted} anchor(s) in `{binding_id}` drifted from their source — run \
`memstead projection sync {binding_id}`"
),
});
}
if uncovered > 0
&& matches!(
crate::binding::effective_coverage_semantics(binding).value,
CoverageSemantics::Exhaustive
)
{
candidates.push(Candidate {
severity: 3,
text: format!(
"{uncovered} in-scope source artifact(s) in `{binding_id}` carry no entity \
— run `memstead projection verify {binding_id}`, then sync"
),
});
}
if queued > 0 {
candidates.push(Candidate {
severity: 2,
text: format!(
"{queued} finding(s) in `{binding_id}` queued for adjudication — run \
`memstead projection verify {binding_id}`"
),
});
}
if resolution.has_action {
action_bindings += 1;
}
}
candidates.sort_by_key(|c| std::cmp::Reverse(c.severity));
let actions: Vec<String> = candidates.into_iter().take(3).map(|c| c.text).collect();
let verdict = if action_bindings > 0 {
RollupVerdict::ActionNeeded
} else if onboarding_bindings > 0 {
RollupVerdict::Onboarding
} else {
RollupVerdict::Clean
};
let headline = match verdict {
RollupVerdict::ActionNeeded => format!(
"Action needed — {action_bindings} of {total} projection(s) have open findings or a \
moved source."
),
RollupVerdict::Onboarding => format!(
"Onboarding — {onboarding_bindings} of {total} projection(s) predate their binding; a \
first-sync backfill is expected, not a defect."
),
RollupVerdict::Clean => {
format!("All {total} projection(s) are in sync — no open findings, no moved sources.")
}
};
Rollup {
verdict,
headline,
actions,
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::binding::{
BINDING_VERSION, Binding, BuildMode, BuildOperation, Operations, SyncOperation,
};
use crate::pipeline::{IngestTrigger, MediumType, PatternEntry, PatternMode};
use crate::pipeline_store::write_binding;
use crate::storage::FilesystemMemWriter;
use crate::workspace::{Mount, MountCapability, MountLifecycle, MountStorage};
use tempfile::TempDir;
#[test]
fn projection_status_reports_operations_signal_and_baseline() {
let tmp = TempDir::new().unwrap();
let root = tmp.path();
std::fs::create_dir_all(root.join(".memstead")).unwrap();
std::fs::write(
root.join(".memstead").join("config.json"),
br#"{"format":1,"schema":"default@1.0.0"}"#,
)
.unwrap();
std::fs::write(
root.join(".memstead").join("workspace.toml"),
"[workspace]\n",
)
.unwrap();
let out = std::process::Command::new("git")
.args(["init", "-q"])
.current_dir(root)
.output()
.unwrap();
assert!(out.status.success());
write_binding(
root,
"engine",
"graph",
&Binding {
version: BINDING_VERSION,
intent: None,
sources: vec![crate::pipeline::Source {
name: "graph".to_string(),
medium_type: MediumType::Codebase,
pointer: String::new(),
change_detection: Some("git".to_string()),
scope: vec![PatternEntry {
path: "**/*.rs".to_string(),
mode: PatternMode::Allow,
}],
engagement: None,
preparation: None,
}],
reference_mems: Vec::new(),
destination_mem: "engine".to_string(),
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: Some(SyncOperation {
trigger: IngestTrigger::Manual,
batch_size: 20,
}),
verify: None,
},
},
)
.unwrap();
let mount = Mount {
mem: "engine".to_string(),
schema: Some("default@1.0.0".parse().unwrap()),
storage: MountStorage::Folder {
path: root.to_path_buf(),
},
capability: MountCapability::Write,
lifecycle: MountLifecycle::Eager,
cross_linkable: false,
migration_target: None,
};
let mut engine = Engine::from_mounts(vec![(
mount,
Box::new(FilesystemMemWriter::new(root.to_path_buf()))
as Box<dyn crate::backend::MemBackend>,
)])
.unwrap();
engine
.set_mem_sync_state("engine", "engine/graph/graph#synced", "deadbeef", None)
.unwrap();
let ps = projection_status(&engine, root);
assert_eq!(ps.len(), 1);
let p = &ps[0];
assert_eq!(p.binding, "engine/graph");
assert_eq!(p.destination_mem, "engine");
assert_eq!(p.operations, vec!["build".to_string(), "sync".to_string()]);
let facet = p.state.get("graph").expect("the source facet's state");
assert_eq!(facet.signal, "git");
assert_eq!(facet.synced.as_deref(), Some("deadbeef"));
assert_eq!(facet.verified, None);
assert_eq!(
p.advance,
AdvanceCounts {
pending: 0,
disposed: 0
}
);
}
#[test]
fn projection_status_empty_without_bindings() {
let tmp = TempDir::new().unwrap();
let root = tmp.path();
std::fs::create_dir_all(root.join(".memstead")).unwrap();
std::fs::write(
root.join(".memstead").join("config.json"),
br#"{"format":1,"schema":"default@1.0.0"}"#,
)
.unwrap();
let mount = Mount {
mem: "engine".to_string(),
schema: Some("default@1.0.0".parse().unwrap()),
storage: MountStorage::Folder {
path: root.to_path_buf(),
},
capability: MountCapability::Write,
lifecycle: MountLifecycle::Eager,
cross_linkable: false,
migration_target: None,
};
let engine = Engine::from_mounts(vec![(
mount,
Box::new(FilesystemMemWriter::new(root.to_path_buf()))
as Box<dyn crate::backend::MemBackend>,
)])
.unwrap();
assert!(projection_status(&engine, root).is_empty());
}
fn one_binding_workspace(tmp: &TempDir) -> Engine {
let root = tmp.path();
std::fs::create_dir_all(root.join(".memstead")).unwrap();
std::fs::write(
root.join(".memstead").join("config.json"),
br#"{"format":1,"schema":"default@1.0.0"}"#,
)
.unwrap();
std::fs::write(
root.join(".memstead").join("workspace.toml"),
"[workspace]\n",
)
.unwrap();
let out = std::process::Command::new("git")
.args(["init", "-q"])
.current_dir(root)
.output()
.unwrap();
assert!(out.status.success());
write_binding(
root,
"engine",
"graph",
&Binding {
version: BINDING_VERSION,
intent: None,
sources: vec![crate::pipeline::Source {
name: "graph".to_string(),
medium_type: MediumType::Codebase,
pointer: String::new(),
change_detection: Some("git".to_string()),
scope: vec![PatternEntry {
path: "**/*.rs".to_string(),
mode: PatternMode::Allow,
}],
engagement: None,
preparation: None,
}],
reference_mems: Vec::new(),
destination_mem: "engine".to_string(),
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: Some(SyncOperation {
trigger: IngestTrigger::Manual,
batch_size: 20,
}),
verify: None,
},
},
)
.unwrap();
let mount = Mount {
mem: "engine".to_string(),
schema: Some("default@1.0.0".parse().unwrap()),
storage: MountStorage::Folder {
path: root.to_path_buf(),
},
capability: MountCapability::Write,
lifecycle: MountLifecycle::Eager,
cross_linkable: false,
migration_target: None,
};
Engine::from_mounts(vec![(
mount,
Box::new(FilesystemMemWriter::new(root.to_path_buf()))
as Box<dyn crate::backend::MemBackend>,
)])
.unwrap()
}
#[test]
fn projection_status_carries_the_per_binding_verdict() {
let tmp = TempDir::new().unwrap();
let engine = one_binding_workspace(&tmp);
let statuses = projection_status(&engine, tmp.path());
assert_eq!(statuses.len(), 1);
let s = &statuses[0];
assert_eq!(s.verdict, RollupVerdict::Onboarding);
assert!(!s.source_moved, "onboarding skips the freshness scan");
assert_eq!(s.findings, FindingCounts::default());
let json = serde_json::to_value(s).unwrap();
assert_eq!(json["verdict"], "onboarding");
assert_eq!(json["source_moved"], false);
assert_eq!(json["findings"]["unresolvable"], 0);
let rollup = projection_rollup(&engine, tmp.path());
assert_eq!(rollup.verdict, RollupVerdict::Onboarding);
}
#[test]
fn rollup_adopt_binding_is_onboarding_not_action_needed() {
let tmp = TempDir::new().unwrap();
let engine = one_binding_workspace(&tmp);
let rollup = projection_rollup(&engine, tmp.path());
assert_eq!(rollup.verdict, RollupVerdict::Onboarding);
assert_ne!(
rollup.verdict,
RollupVerdict::ActionNeeded,
"pre-binding history alone must never be a red verdict"
);
assert!(
rollup
.actions
.iter()
.any(|a| a.contains("predates its binding")),
"the onboarding action is surfaced: {:?}",
rollup.actions
);
assert!(rollup.headline.contains("Onboarding"));
}
#[test]
fn rollup_empty_without_bindings_is_clean() {
let tmp = TempDir::new().unwrap();
let root = tmp.path();
std::fs::create_dir_all(root.join(".memstead")).unwrap();
std::fs::write(
root.join(".memstead").join("config.json"),
br#"{"format":1,"schema":"default@1.0.0"}"#,
)
.unwrap();
let mount = Mount {
mem: "engine".to_string(),
schema: Some("default@1.0.0".parse().unwrap()),
storage: MountStorage::Folder {
path: root.to_path_buf(),
},
capability: MountCapability::Write,
lifecycle: MountLifecycle::Eager,
cross_linkable: false,
migration_target: None,
};
let engine = Engine::from_mounts(vec![(
mount,
Box::new(FilesystemMemWriter::new(root.to_path_buf()))
as Box<dyn crate::backend::MemBackend>,
)])
.unwrap();
let rollup = projection_rollup(&engine, root);
assert_eq!(rollup.verdict, RollupVerdict::Clean);
assert!(rollup.actions.is_empty());
assert!(rollup.headline.contains("No projection bindings"));
}
}