use std::collections::{BTreeMap, BTreeSet};
use std::path::{Path, PathBuf};
use std::time::{SystemTime, UNIX_EPOCH};
use serde::{Deserialize, Serialize};
use crate::Engine;
use crate::anchor::{Anchor, AnchorState};
use crate::binding::{
BindingV1, DEFAULT_ADJUDICATION_CAP, DEFAULT_FULL_RESYNC_EVERY, ResolvedBinding, hash_binding,
medium_capabilities,
};
use crate::workspace_store::{StoreError, WORKSPACE_STORE_DIR};
use super::advance::is_single_component;
use super::cursor::{compute_source_cursor, enumerate_facet_files};
use super::refinement::{
ROTATION_ANCHOR_ADJUDICATION, bump_verify_runs, next_batch, next_rotation_batch,
};
use super::resolve::{ResolvedIngest, ResolvedSource};
const STATE_DIR: &str = "state";
const FINDINGS_DIR: &str = "findings";
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
pub struct FindingKey {
pub binding_hash: String,
pub source_head: String,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "kebab-case")]
pub enum FindingClass {
Drifted,
Wrong,
Uncovered,
UnresolvableAnchor,
QueuedForAdjudication,
}
impl FindingClass {
pub const WIRE_VALUES: &'static [&'static str] = &[
"drifted",
"wrong",
"uncovered",
"unresolvable-anchor",
"queued-for-adjudication",
];
pub fn as_wire(&self) -> &'static str {
match self {
FindingClass::Drifted => "drifted",
FindingClass::Wrong => "wrong",
FindingClass::Uncovered => "uncovered",
FindingClass::UnresolvableAnchor => "unresolvable-anchor",
FindingClass::QueuedForAdjudication => "queued-for-adjudication",
}
}
pub fn from_wire(s: &str) -> Option<Self> {
match s {
"drifted" => Some(FindingClass::Drifted),
"wrong" => Some(FindingClass::Wrong),
"uncovered" => Some(FindingClass::Uncovered),
"unresolvable-anchor" => Some(FindingClass::UnresolvableAnchor),
"queued-for-adjudication" => Some(FindingClass::QueuedForAdjudication),
_ => None,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "kebab-case")]
pub enum FindingTarget {
Anchor {
entity: String,
artifact: String,
},
Artifact {
artifact: String,
},
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct Finding {
pub key: FindingKey,
pub facet: String,
pub target: FindingTarget,
pub class: FindingClass,
pub detail: String,
pub created_at: String,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct FindingsBatch {
pub key: FindingKey,
pub recorded_at: String,
pub findings: Vec<Finding>,
}
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct FindingsStore {
pub binding: String,
#[serde(default)]
pub batches: Vec<FindingsBatch>,
}
impl FindingsStore {
pub fn record(&mut self, key: FindingKey, recorded_at: String, findings: Vec<Finding>) {
if let Some(batch) = self.batches.iter_mut().find(|b| b.key == key) {
batch.recorded_at = recorded_at;
batch.findings = findings;
} else {
self.batches.push(FindingsBatch {
key,
recorded_at,
findings,
});
}
}
pub fn current(&self, key: &FindingKey) -> &[Finding] {
self.batches
.iter()
.find(|b| &b.key == key)
.map(|b| b.findings.as_slice())
.unwrap_or(&[])
}
pub fn superseded(&self, key: &FindingKey) -> Vec<&Finding> {
self.batches
.iter()
.filter(|b| &b.key != key)
.flat_map(|b| b.findings.iter())
.collect()
}
}
pub fn findings_store_path(workspace_root: &Path, mem: &str, name: &str) -> PathBuf {
workspace_root
.join(WORKSPACE_STORE_DIR)
.join(STATE_DIR)
.join(FINDINGS_DIR)
.join(mem)
.join(format!("{name}.json"))
}
pub fn read_findings_store(
workspace_root: &Path,
mem: &str,
name: &str,
) -> Result<Option<FindingsStore>, StoreError> {
let path = findings_store_path(workspace_root, mem, name);
match std::fs::read(&path) {
Ok(bytes) => serde_json::from_slice(&bytes)
.map(Some)
.map_err(|e| StoreError::Parse {
path,
message: e.to_string(),
}),
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None),
Err(e) => Err(StoreError::Io { path, source: e }),
}
}
pub fn write_findings_store(
workspace_root: &Path,
mem: &str,
name: &str,
store: &FindingsStore,
) -> Result<(), StoreError> {
let path = findings_store_path(workspace_root, mem, name);
if let Some(parent) = path.parent() {
std::fs::create_dir_all(parent).map_err(|e| StoreError::Io {
path: parent.to_path_buf(),
source: e,
})?;
}
let bytes = serde_json::to_vec_pretty(store).map_err(|e| StoreError::Parse {
path: path.clone(),
message: e.to_string(),
})?;
std::fs::write(&path, bytes).map_err(|e| StoreError::Io { path, source: e })
}
pub fn delete_findings_store(
workspace_root: &Path,
mem: &str,
name: &str,
) -> Result<(), StoreError> {
let path = findings_store_path(workspace_root, mem, name);
match std::fs::remove_file(&path) {
Ok(()) => Ok(()),
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
Err(e) => Err(StoreError::Io { path, source: e }),
}
}
#[derive(Debug, thiserror::Error)]
pub enum FindingsError {
#[error("malformed binding id '{0}': expected `<mem>/<stem>`")]
MalformedId(String),
#[error("findings store error: {0}")]
Store(#[source] StoreError),
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct VerifyOutcome {
pub binding: String,
pub key: FindingKey,
pub recorded: usize,
pub superseded: usize,
pub backlog: usize,
pub full_resync: FullResyncDecision,
}
fn split_binding_id(binding_id: &str) -> Result<(String, String), FindingsError> {
binding_id
.split_once('/')
.filter(|(m, n)| is_single_component(m) && is_single_component(n))
.map(|(m, n)| (m.to_string(), n.to_string()))
.ok_or_else(|| FindingsError::MalformedId(binding_id.to_string()))
}
fn source_facet_label(resolved: &ResolvedIngest) -> String {
let facets: Vec<&str> = resolved
.sources
.iter()
.filter_map(|s| match s {
ResolvedSource::Primary(p) => Some(p.facet_ref.as_str()),
ResolvedSource::Reference { .. } => None,
})
.collect();
facets.join(",")
}
fn now_seconds() -> String {
let secs = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0);
secs.to_string()
}
fn current_source_head(
engine: &Engine,
workspace_root: &Path,
resolved: &ResolvedIngest,
) -> String {
let binding_id = &resolved.name;
let prefix = format!("{binding_id}/");
let mut tokens: BTreeMap<String, String> = BTreeMap::new();
if let Some(cfg) = engine.mem_config_for(&resolved.destination_mem) {
for (k, v) in &cfg.sync_state {
if let Some(rest) = k.strip_prefix(&prefix)
&& let Some(facet) = rest.strip_suffix("#synced")
{
tokens.insert(facet.to_string(), v.clone());
}
}
}
let cursor = compute_source_cursor(engine, resolved, workspace_root);
for c in cursor.write_commands.iter().chain(cursor.reseed.iter()) {
if let Some(rest) = c.key.strip_prefix(&prefix)
&& let Some(facet) = rest.strip_suffix("#synced")
{
tokens.insert(facet.to_string(), c.token.clone());
}
}
tokens
.iter()
.map(|(facet, token)| format!("{facet}={token}"))
.collect::<Vec<_>>()
.join(";")
}
fn current_key(
engine: &Engine,
workspace_root: &Path,
binding: &BindingV1,
resolved: &ResolvedIngest,
) -> FindingKey {
let primary_sources = resolved
.sources
.iter()
.filter_map(|s| match s {
ResolvedSource::Primary(p) => Some(p.clone()),
ResolvedSource::Reference { .. } => None,
})
.collect();
let rb = ResolvedBinding {
binding: binding.clone(),
primary_sources,
};
FindingKey {
binding_hash: hash_binding(&rb),
source_head: current_source_head(engine, workspace_root, resolved),
}
}
pub fn current_findings(
engine: &Engine,
workspace_root: &Path,
binding: &BindingV1,
resolved: &ResolvedIngest,
) -> Result<(FindingKey, Vec<Finding>), FindingsError> {
let (mem, name) = split_binding_id(&resolved.name)?;
let key = current_key(engine, workspace_root, binding, resolved);
let findings = read_findings_store(workspace_root, &mem, &name)
.map_err(FindingsError::Store)?
.map(|s| s.current(&key).to_vec())
.unwrap_or_default();
Ok((key, findings))
}
pub fn adjudicate_anchor(
key: &FindingKey,
facet: &str,
entity: &str,
anchor: &Anchor,
state: AnchorState,
created_at: &str,
) -> Option<Finding> {
let (class, detail) = match state {
AnchorState::Resolves => return None,
AnchorState::Orphaned => (
FindingClass::UnresolvableAnchor,
format!(
"artifact '{}' the anchor references is no longer present in the medium",
anchor.artifact
),
),
AnchorState::Drifted | AnchorState::Recheck => {
if !anchor.class.is_hash_bearing() {
return None;
}
match state {
AnchorState::Drifted => (
FindingClass::Drifted,
format!(
"prepared-content hash of '{}' drifted from the anchored hash",
anchor.artifact
),
),
_ => (
FindingClass::QueuedForAdjudication,
format!(
"hash adjudication of '{}' deferred (recheck); queued",
anchor.artifact
),
),
}
}
};
Some(Finding {
key: key.clone(),
facet: facet.to_string(),
target: FindingTarget::Anchor {
entity: entity.to_string(),
artifact: anchor.artifact.clone(),
},
class,
detail,
created_at: created_at.to_string(),
})
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct FacetEnumerability {
pub facet: String,
pub medium_type: String,
pub enumerable: bool,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct FullResyncRefusal {
pub facet: String,
pub medium_type: String,
pub reason: String,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "state", rename_all = "kebab-case")]
pub enum FullResyncDecision {
Disabled,
NotDue {
run_count: u64,
every: u32,
runs_until_due: u32,
},
Due {
run_count: u64,
every: u32,
walked_facets: Vec<String>,
refused: Vec<FullResyncRefusal>,
},
}
impl FullResyncDecision {
pub fn is_full_walk(&self) -> bool {
matches!(self, FullResyncDecision::Due { .. })
}
}
pub fn schedule_full_resync(
every: u32,
run_count: u64,
facets: &[FacetEnumerability],
) -> FullResyncDecision {
if every == 0 {
return FullResyncDecision::Disabled;
}
let modulo = run_count % u64::from(every);
if modulo != 0 {
return FullResyncDecision::NotDue {
run_count,
every,
runs_until_due: (u64::from(every) - modulo) as u32,
};
}
let mut walked_facets = Vec::new();
let mut refused = Vec::new();
for f in facets {
if f.enumerable {
walked_facets.push(f.facet.clone());
} else {
refused.push(FullResyncRefusal {
facet: f.facet.clone(),
medium_type: f.medium_type.clone(),
reason: format!(
"medium type '{}' is non-enumerable — a full-enumeration walk cannot cover \
it; the scheduled full resync refuses rather than claim full coverage",
f.medium_type
),
});
}
}
FullResyncDecision::Due {
run_count,
every,
walked_facets,
refused,
}
}
fn candidate_key(entity: &str, anchor: &Anchor) -> String {
format!("{entity}\u{1f}{}", anchor.artifact)
}
fn adjudicate_candidates(
key: &FindingKey,
facet: &str,
candidates: &[(String, Anchor, AnchorState)],
window: Option<&BTreeSet<String>>,
created_at: &str,
) -> Vec<Finding> {
let mut out = Vec::new();
for (entity, anchor, state) in candidates {
let ck = candidate_key(entity, anchor);
let adjudicate_now = window.is_none_or(|w| w.contains(&ck));
if adjudicate_now {
if let Some(f) = adjudicate_anchor(key, facet, entity, anchor, *state, created_at) {
out.push(f);
}
} else {
out.push(Finding {
key: key.clone(),
facet: facet.to_string(),
target: FindingTarget::Anchor {
entity: entity.clone(),
artifact: anchor.artifact.clone(),
},
class: FindingClass::QueuedForAdjudication,
detail: format!(
"adjudication of '{}' deferred (per-run adjudication cap reached); queued",
anchor.artifact
),
created_at: created_at.to_string(),
});
}
}
out
}
pub fn verify_binding(
engine: &Engine,
workspace_root: &Path,
binding: &BindingV1,
resolved: &ResolvedIngest,
) -> Result<VerifyOutcome, FindingsError> {
let binding_id = resolved.name.clone();
let (mem, name) = split_binding_id(&binding_id)?;
let key = current_key(engine, workspace_root, binding, resolved);
let now = now_seconds();
let facet = source_facet_label(resolved);
let cache_root = workspace_root.join(".memstead.cache").join("ingest");
let verify_op = binding.operations.verify.as_ref();
let cap = verify_op.map_or(DEFAULT_ADJUDICATION_CAP, |v| v.adjudication_cap);
let full_resync_every = verify_op.map_or(DEFAULT_FULL_RESYNC_EVERY, |v| v.full_resync_every);
let sample_batch = verify_op
.map_or(resolved.batch_size, |v| v.batch_size)
.max(1) as usize;
let run_count = bump_verify_runs(&cache_root, &binding_id);
let facet_enum: Vec<FacetEnumerability> = resolved
.sources
.iter()
.filter_map(|s| match s {
ResolvedSource::Primary(p) => Some(FacetEnumerability {
facet: p.facet_ref.clone(),
medium_type: medium_type_wire(p.medium_type),
enumerable: medium_capabilities(p.medium_type).enumerable,
}),
ResolvedSource::Reference { .. } => None,
})
.collect();
let full_resync = schedule_full_resync(full_resync_every, run_count, &facet_enum);
let mut findings: Vec<Finding> = Vec::new();
let mut existence: Vec<(String, Anchor, AnchorState)> = Vec::new();
let mut candidates: Vec<(String, Anchor, AnchorState)> = Vec::new();
for (eid, resolved_anchor) in engine.mem_anchors_resolved(&resolved.destination_mem) {
let Some(state) = resolved_anchor.state else {
continue;
};
let anchor = resolved_anchor.anchor;
match state {
AnchorState::Resolves => {}
AnchorState::Orphaned => existence.push((eid.as_ref().to_string(), anchor, state)),
AnchorState::Drifted | AnchorState::Recheck => {
if anchor.class.is_hash_bearing() {
candidates.push((eid.as_ref().to_string(), anchor, state));
}
}
}
}
for (entity, anchor, state) in &existence {
if let Some(f) = adjudicate_anchor(&key, &facet, entity, anchor, *state, &now) {
findings.push(f);
}
}
let window: Option<BTreeSet<String>> = if cap == 0 {
None
} else {
let mut keys: Vec<String> = candidates
.iter()
.map(|(e, a, _)| candidate_key(e, a))
.collect();
keys.sort();
keys.dedup();
next_rotation_batch(
&cache_root,
&binding_id,
ROTATION_ANCHOR_ADJUDICATION,
keys,
cap as usize,
)
.map(|b| b.files.into_iter().collect())
};
findings.extend(adjudicate_candidates(
&key,
&facet,
&candidates,
window.as_ref(),
&now,
));
let sample_files: Vec<String> = if full_resync.is_full_walk() {
let mut all: Vec<String> = Vec::new();
for source in &resolved.sources {
if let ResolvedSource::Primary(p) = source
&& medium_capabilities(p.medium_type).enumerable
{
all.extend(enumerate_facet_files(
p,
&resolved.deny_paths,
workspace_root,
));
}
}
all.sort();
all.dedup();
all
} else {
next_batch(resolved, workspace_root, &cache_root, sample_batch)
.map(|b| b.files)
.unwrap_or_default()
};
for file in sample_files {
let covered = engine
.anchors_referencing_artifact(&file)
.iter()
.any(|(eid, _)| eid.mem() == resolved.destination_mem.as_str());
if !covered {
findings.push(Finding {
key: key.clone(),
facet: facet.clone(),
target: FindingTarget::Artifact { artifact: file },
class: FindingClass::Uncovered,
detail: "source artifact in scope has no anchor in the destination mem".to_string(),
created_at: now.clone(),
});
}
}
let backlog = findings
.iter()
.filter(|f| f.class == FindingClass::QueuedForAdjudication)
.count();
let mut store = read_findings_store(workspace_root, &mem, &name)
.map_err(FindingsError::Store)?
.unwrap_or_else(|| FindingsStore {
binding: binding_id.clone(),
..Default::default()
});
let recorded = findings.len();
store.record(key.clone(), now, findings);
let superseded = store.superseded(&key).len();
write_findings_store(workspace_root, &mem, &name, &store).map_err(FindingsError::Store)?;
Ok(VerifyOutcome {
binding: binding_id,
key,
recorded,
superseded,
backlog,
full_resync,
})
}
fn medium_type_wire(t: crate::pipeline::MediumType) -> String {
serde_json::to_value(t)
.ok()
.and_then(|v| v.as_str().map(str::to_string))
.unwrap_or_default()
}
#[cfg(test)]
mod tests {
use super::*;
use crate::anchor::{Anchor, AnchorGrain, AnchorHashStability, AnchorProvenanceClass};
fn key(hash: &str, head: &str) -> FindingKey {
FindingKey {
binding_hash: hash.to_string(),
source_head: head.to_string(),
}
}
fn anchor(class: AnchorProvenanceClass) -> Anchor {
Anchor {
artifact: "src/lib.rs".to_string(),
grain: AnchorGrain::File,
class,
at_version: None,
hash: if class.is_hash_bearing() {
Some("h1".to_string())
} else {
None
},
hash_stability: AnchorHashStability::Stable,
derived_from: Vec::new(),
binding: None,
}
}
#[test]
fn store_round_trips_on_disk_and_delete_is_idempotent() {
let tmp = tempfile::tempdir().unwrap();
let root = tmp.path();
assert!(
read_findings_store(root, "engine", "graph")
.unwrap()
.is_none()
);
let mut store = FindingsStore {
binding: "engine/graph".to_string(),
..Default::default()
};
let k = key("hashA", "head1");
store.record(
k.clone(),
"1".to_string(),
vec![Finding {
key: k.clone(),
facet: "src".to_string(),
target: FindingTarget::Artifact {
artifact: "src/a.rs".to_string(),
},
class: FindingClass::Uncovered,
detail: "d".to_string(),
created_at: "1".to_string(),
}],
);
write_findings_store(root, "engine", "graph", &store).unwrap();
assert!(findings_store_path(root, "engine", "graph").exists());
let back = read_findings_store(root, "engine", "graph")
.unwrap()
.unwrap();
assert_eq!(back, store);
assert_eq!(back.current(&k).len(), 1);
delete_findings_store(root, "engine", "graph").unwrap();
assert!(
read_findings_store(root, "engine", "graph")
.unwrap()
.is_none()
);
delete_findings_store(root, "engine", "graph").unwrap();
}
#[test]
fn changed_binding_hash_supersedes_prior_findings() {
let mut store = FindingsStore::default();
let old = key("hashOLD", "head1");
let new = key("hashNEW", "head1");
let f_old = Finding {
key: old.clone(),
facet: "src".to_string(),
target: FindingTarget::Artifact {
artifact: "src/old.rs".to_string(),
},
class: FindingClass::Uncovered,
detail: "old".to_string(),
created_at: "1".to_string(),
};
store.record(old.clone(), "1".to_string(), vec![f_old.clone()]);
store.record(new.clone(), "2".to_string(), Vec::new());
assert!(store.current(&new).is_empty(), "new key has its own view");
let superseded = store.superseded(&new);
assert_eq!(superseded.len(), 1, "old batch is segregated as superseded");
assert_eq!(superseded[0], &f_old);
assert!(!store.current(&new).contains(&f_old));
}
#[test]
fn moved_source_head_supersedes_prior_findings() {
let mut store = FindingsStore::default();
let before = key("hashA", "head1");
let after = key("hashA", "head2");
let f = Finding {
key: before.clone(),
facet: "src".to_string(),
target: FindingTarget::Anchor {
entity: "engine--e".to_string(),
artifact: "src/x.rs".to_string(),
},
class: FindingClass::UnresolvableAnchor,
detail: "gone".to_string(),
created_at: "1".to_string(),
};
store.record(before.clone(), "1".to_string(), vec![f.clone()]);
store.record(after.clone(), "2".to_string(), Vec::new());
assert!(store.current(&after).is_empty());
assert_eq!(store.superseded(&after), vec![&f]);
store.record(after.clone(), "3".to_string(), Vec::new());
assert_eq!(store.batches.len(), 2, "one batch per distinct key");
}
#[test]
fn informed_by_anchor_never_drifts() {
let k = key("h", "s");
for class in [
AnchorProvenanceClass::InformedBy,
AnchorProvenanceClass::Authored,
] {
let a = anchor(class);
assert!(
adjudicate_anchor(&k, "f", "engine--e", &a, AnchorState::Drifted, "1").is_none(),
"{class:?} must not produce a drift finding"
);
assert!(
adjudicate_anchor(&k, "f", "engine--e", &a, AnchorState::Recheck, "1").is_none(),
"{class:?} must not produce a queued finding"
);
}
}
#[test]
fn hash_bearing_drifts_and_orphan_is_class_independent() {
let k = key("h", "s");
let anchored = anchor(AnchorProvenanceClass::Anchored);
let drifted =
adjudicate_anchor(&k, "f", "engine--e", &anchored, AnchorState::Drifted, "1").unwrap();
assert_eq!(drifted.class, FindingClass::Drifted);
assert_eq!(drifted.key, k, "the finding carries its recording key (A2)");
let queued =
adjudicate_anchor(&k, "f", "engine--e", &anchored, AnchorState::Recheck, "1").unwrap();
assert_eq!(queued.class, FindingClass::QueuedForAdjudication);
let informed = anchor(AnchorProvenanceClass::InformedBy);
let orphan =
adjudicate_anchor(&k, "f", "engine--e", &informed, AnchorState::Orphaned, "1").unwrap();
assert_eq!(orphan.class, FindingClass::UnresolvableAnchor);
assert!(
adjudicate_anchor(&k, "f", "engine--e", &anchored, AnchorState::Resolves, "1")
.is_none()
);
}
#[test]
fn finding_class_wire_round_trips() {
for w in FindingClass::WIRE_VALUES {
let c = FindingClass::from_wire(w).expect("known wire value");
assert_eq!(c.as_wire(), *w);
}
assert!(FindingClass::from_wire("nonsense").is_none());
}
#[test]
fn malformed_binding_id_refuses() {
assert!(matches!(
split_binding_id("../escape"),
Err(FindingsError::MalformedId(_))
));
assert!(matches!(
split_binding_id("no-slash"),
Err(FindingsError::MalformedId(_))
));
assert_eq!(
split_binding_id("engine/graph").unwrap(),
("engine".to_string(), "graph".to_string())
);
}
use crate::anchor::AnchorSidecar;
use crate::binding::{
BINDING_VERSION, BuildMode, BuildOperation, CoverageSemantics, DEFAULT_ADJUDICATION_CAP,
DEFAULT_FULL_RESYNC_EVERY, Operations, VerifyOperation,
};
use crate::ingest::resolve::resolve_binding_run;
use crate::pipeline::{Facet, IngestTrigger, Medium, MediumType, PatternEntry, PatternMode};
use crate::pipeline_store::{load_pipeline_configs, write_binding, write_facet, write_medium};
use crate::workspace::{
Mount, MountCapability, MountLifecycle, MountStorage, Workspace, WorkspaceSettings,
};
use crate::workspace_store::WorkspaceStoreAdapter;
#[test]
fn verify_persists_findings_readable_fresh() {
let tmp = tempfile::tempdir().unwrap();
let root = tmp.path();
let mem_dir = root.join("mem");
std::fs::create_dir_all(mem_dir.join(".memstead")).unwrap();
std::fs::write(
mem_dir.join(".memstead").join("config.json"),
r#"{"format":1,"schema":"default@1.0.0","version":"1.0.0"}"#,
)
.unwrap();
std::fs::create_dir_all(root.join(".memstead")).unwrap();
std::fs::write(
root.join(".memstead").join("workspace.toml"),
"format = \"memstead-git-branch-2\"\n\n[persistence_adapter]\nname = \"file-two-layer\"\n",
)
.unwrap();
let mount = Mount {
mem: "engine".to_string(),
schema: Some("default@1.0.0".parse().unwrap()),
storage: MountStorage::Folder {
path: mem_dir.clone(),
},
capability: MountCapability::Write,
lifecycle: MountLifecycle::Eager,
cross_linkable: false,
migration_target: None,
};
crate::FileWorkspaceStore::new()
.save_state(
root,
&Workspace {
mounts: vec![mount],
settings: WorkspaceSettings::default(),
},
)
.unwrap();
let out = std::process::Command::new("git")
.args(["init", "-q"])
.current_dir(root)
.output()
.unwrap();
assert!(out.status.success());
std::fs::create_dir_all(root.join("src")).unwrap();
std::fs::write(root.join("src").join("present.rs"), "fn a() {}\n").unwrap();
std::fs::write(root.join("src").join("uncovered.rs"), "fn b() {}\n").unwrap();
let mk = |artifact: &str, class: AnchorProvenanceClass| Anchor {
artifact: artifact.to_string(),
grain: AnchorGrain::File,
class,
at_version: None,
hash: class.is_hash_bearing().then(|| "recorded".to_string()),
hash_stability: AnchorHashStability::Stable,
derived_from: Vec::new(),
binding: None,
};
let mut sidecar = AnchorSidecar::default();
sidecar.set(
"engine--e",
vec![
mk("src/present.rs", AnchorProvenanceClass::Anchored), mk("src/gone.rs", AnchorProvenanceClass::Anchored), mk("src/present.rs", AnchorProvenanceClass::InformedBy), ],
);
std::fs::write(
mem_dir.join(crate::anchor::ANCHOR_SIDECAR_PATH),
sidecar.to_bytes(),
)
.unwrap();
write_medium(
root,
"engine",
"graph",
&Medium {
name: "graph".to_string(),
medium_type: MediumType::Codebase,
pointer: String::new(),
change_detection: Some("git".to_string()),
},
)
.unwrap();
write_facet(
root,
"engine",
"graph",
&Facet {
name: "graph".to_string(),
medium: "graph".to_string(),
scope: vec![PatternEntry {
path: "src/**/*.rs".to_string(),
mode: PatternMode::Allow,
}],
engagement: None,
preparation: None,
},
)
.unwrap();
write_binding(
root,
"engine",
"graph",
&BindingV1 {
version: BINDING_VERSION,
intent: None,
source_facets: vec!["graph".to_string()],
reference_mems: Vec::new(),
destination_mem: "engine".to_string(),
deny_paths: Vec::new(),
coverage_semantics: CoverageSemantics::Exhaustive,
rules: None,
prune: None,
operations: Operations {
build: Some(BuildOperation {
mode: BuildMode::Discovery,
trigger: IngestTrigger::Loop,
batch_size: 20,
post_actions: None,
}),
sync: None,
verify: Some(VerifyOperation {
trigger: IngestTrigger::Manual,
batch_size: 20,
adjudication_cap: DEFAULT_ADJUDICATION_CAP,
full_resync_every: DEFAULT_FULL_RESYNC_EVERY,
}),
},
},
)
.unwrap();
let engine = Engine::from_workspace_root(root).unwrap();
let configs = load_pipeline_configs(root).unwrap();
let binding = &configs.bindings[0].config;
let resolved = resolve_binding_run(&configs, "engine/graph", binding).unwrap();
let outcome = verify_binding(&engine, root, binding, &resolved).unwrap();
assert!(
outcome.recorded >= 3,
"orphan + queued + uncovered at least"
);
assert_eq!(outcome.superseded, 0, "no prior key yet");
assert_eq!(outcome.backlog, 1, "the present hash-bearing anchor queued");
let store = read_findings_store(root, "engine", "graph")
.unwrap()
.unwrap();
let current = store.current(&outcome.key);
assert_eq!(current.len(), outcome.recorded);
let has = |c: FindingClass, art: &str| {
current.iter().any(|f| {
f.class == c
&& match &f.target {
FindingTarget::Anchor { artifact, .. } => artifact == art,
FindingTarget::Artifact { artifact } => artifact == art,
}
})
};
assert!(has(FindingClass::UnresolvableAnchor, "src/gone.rs"));
assert!(has(FindingClass::QueuedForAdjudication, "src/present.rs"));
assert!(has(FindingClass::Uncovered, "src/uncovered.rs"));
assert!(
!current
.iter()
.any(|f| f.class == FindingClass::Drifted || f.class == FindingClass::Wrong),
"no drift finding from a non-hash / present-clean anchor"
);
assert!(!has(FindingClass::Uncovered, "src/present.rs"));
}
#[test]
fn adjudication_cap_queues_the_remainder() {
let k = key("h", "s");
let mk = |art: &str| {
let mut a = anchor(AnchorProvenanceClass::Anchored);
a.artifact = art.to_string();
a
};
let candidates = vec![
(
"engine--a".to_string(),
mk("src/a.rs"),
AnchorState::Drifted,
),
(
"engine--b".to_string(),
mk("src/b.rs"),
AnchorState::Drifted,
),
(
"engine--c".to_string(),
mk("src/c.rs"),
AnchorState::Drifted,
),
];
let window: BTreeSet<String> = [candidate_key("engine--a", &mk("src/a.rs"))]
.into_iter()
.collect();
let out = adjudicate_candidates(&k, "f", &candidates, Some(&window), "1");
let drifted = out
.iter()
.filter(|f| f.class == FindingClass::Drifted)
.count();
let queued = out
.iter()
.filter(|f| f.class == FindingClass::QueuedForAdjudication)
.count();
assert_eq!(drifted, 1, "only the in-window candidate is adjudicated");
assert_eq!(queued, 2, "the remainder is queued as the tier-3 backlog");
assert!(
out.iter()
.any(|f| f.class == FindingClass::QueuedForAdjudication
&& f.detail.contains("cap reached")),
"capped remainder states it was deferred by the cap"
);
let uncapped = adjudicate_candidates(&k, "f", &candidates, None, "1");
assert_eq!(
uncapped
.iter()
.filter(|f| f.class == FindingClass::Drifted)
.count(),
3,
"uncapped adjudicates every candidate"
);
assert_eq!(
uncapped
.iter()
.filter(|f| f.class == FindingClass::QueuedForAdjudication)
.count(),
0
);
}
#[test]
fn full_resync_schedule_disabled_notdue_due() {
let codebase = FacetEnumerability {
facet: "src".to_string(),
medium_type: "codebase".to_string(),
enumerable: true,
};
assert_eq!(
schedule_full_resync(0, 5, std::slice::from_ref(&codebase)),
FullResyncDecision::Disabled
);
match schedule_full_resync(3, 2, std::slice::from_ref(&codebase)) {
FullResyncDecision::NotDue { runs_until_due, .. } => assert_eq!(runs_until_due, 1),
other => panic!("expected NotDue, got {other:?}"),
}
match schedule_full_resync(3, 3, std::slice::from_ref(&codebase)) {
FullResyncDecision::Due {
walked_facets,
refused,
..
} => {
assert_eq!(walked_facets, vec!["src".to_string()]);
assert!(refused.is_empty(), "enumerable facet is not refused");
}
other => panic!("expected Due, got {other:?}"),
}
}
#[test]
fn full_resync_refuses_non_enumerable_medium() {
let web = FacetEnumerability {
facet: "manual".to_string(),
medium_type: "web".to_string(),
enumerable: false,
};
let d = schedule_full_resync(1, 1, &[web]);
assert!(
d.is_full_walk(),
"a due sweep is a full walk even when refused"
);
match d {
FullResyncDecision::Due {
walked_facets,
refused,
..
} => {
assert!(walked_facets.is_empty(), "nothing enumerable to walk");
assert_eq!(refused.len(), 1, "the non-enumerable facet is refused");
assert_eq!(refused[0].facet, "manual");
assert_eq!(refused[0].medium_type, "web");
assert!(
refused[0].reason.contains("non-enumerable"),
"the refusal is typed and states why"
);
}
other => panic!("expected Due with a refusal, got {other:?}"),
}
}
#[test]
fn full_resync_full_walk_covers_whole_source() {
let tmp = tempfile::tempdir().unwrap();
let root = tmp.path();
let mem_dir = root.join("mem");
std::fs::create_dir_all(mem_dir.join(".memstead")).unwrap();
std::fs::write(
mem_dir.join(".memstead").join("config.json"),
r#"{"format":1,"schema":"default@1.0.0","version":"1.0.0"}"#,
)
.unwrap();
std::fs::create_dir_all(root.join(".memstead")).unwrap();
std::fs::write(
root.join(".memstead").join("workspace.toml"),
"format = \"memstead-git-branch-2\"\n\n[persistence_adapter]\nname = \"file-two-layer\"\n",
)
.unwrap();
let mount = Mount {
mem: "engine".to_string(),
schema: Some("default@1.0.0".parse().unwrap()),
storage: MountStorage::Folder {
path: mem_dir.clone(),
},
capability: MountCapability::Write,
lifecycle: MountLifecycle::Eager,
cross_linkable: false,
migration_target: None,
};
crate::FileWorkspaceStore::new()
.save_state(
root,
&Workspace {
mounts: vec![mount],
settings: WorkspaceSettings::default(),
},
)
.unwrap();
let out = std::process::Command::new("git")
.args(["init", "-q"])
.current_dir(root)
.output()
.unwrap();
assert!(out.status.success());
std::fs::create_dir_all(root.join("src")).unwrap();
for f in ["a.rs", "b.rs", "c.rs"] {
std::fs::write(root.join("src").join(f), "fn x() {}\n").unwrap();
}
write_medium(
root,
"engine",
"graph",
&Medium {
name: "graph".to_string(),
medium_type: MediumType::Codebase,
pointer: String::new(),
change_detection: Some("git".to_string()),
},
)
.unwrap();
write_facet(
root,
"engine",
"graph",
&Facet {
name: "graph".to_string(),
medium: "graph".to_string(),
scope: vec![PatternEntry {
path: "src/**/*.rs".to_string(),
mode: PatternMode::Allow,
}],
engagement: None,
preparation: None,
},
)
.unwrap();
write_binding(
root,
"engine",
"graph",
&BindingV1 {
version: BINDING_VERSION,
intent: None,
source_facets: vec!["graph".to_string()],
reference_mems: Vec::new(),
destination_mem: "engine".to_string(),
deny_paths: Vec::new(),
coverage_semantics: CoverageSemantics::Exhaustive,
rules: None,
prune: None,
operations: Operations {
build: Some(BuildOperation {
mode: BuildMode::Discovery,
trigger: IngestTrigger::Loop,
batch_size: 20,
post_actions: None,
}),
sync: None,
verify: Some(VerifyOperation {
trigger: IngestTrigger::Manual,
batch_size: 1, adjudication_cap: DEFAULT_ADJUDICATION_CAP,
full_resync_every: 1, }),
},
},
)
.unwrap();
let engine = Engine::from_workspace_root(root).unwrap();
let configs = load_pipeline_configs(root).unwrap();
let binding = &configs.bindings[0].config;
let resolved = resolve_binding_run(&configs, "engine/graph", binding).unwrap();
let outcome = verify_binding(&engine, root, binding, &resolved).unwrap();
match &outcome.full_resync {
FullResyncDecision::Due {
walked_facets,
refused,
run_count,
..
} => {
assert_eq!(*run_count, 1);
assert_eq!(walked_facets, &vec!["graph".to_string()]);
assert!(refused.is_empty());
}
other => panic!("expected a due full walk, got {other:?}"),
}
let store = read_findings_store(root, "engine", "graph")
.unwrap()
.unwrap();
let uncovered = store
.current(&outcome.key)
.iter()
.filter(|f| f.class == FindingClass::Uncovered)
.count();
assert_eq!(
uncovered, 3,
"the scheduled full walk covers the whole source, not a batch of one"
);
}
}