1use std::collections::{BTreeMap, BTreeSet};
36use std::path::{Path, PathBuf};
37use std::time::{SystemTime, UNIX_EPOCH};
38
39use serde::{Deserialize, Serialize};
40
41use crate::Engine;
42use crate::anchor::{Anchor, AnchorState};
43use crate::binding::{
44 BindingV1, DEFAULT_ADJUDICATION_CAP, DEFAULT_FULL_RESYNC_EVERY, ResolvedBinding, hash_binding,
45 medium_capabilities,
46};
47use crate::workspace_store::{StoreError, WORKSPACE_STORE_DIR};
48
49use super::advance::is_single_component;
50use super::cursor::{compute_source_cursor, enumerate_facet_files};
51use super::refinement::{
52 ROTATION_ANCHOR_ADJUDICATION, bump_verify_runs, next_batch, next_rotation_batch,
53};
54use super::resolve::{ResolvedIngest, ResolvedSource};
55
56const STATE_DIR: &str = "state";
59const FINDINGS_DIR: &str = "findings";
61
62#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
75pub struct FindingKey {
76 pub binding_hash: String,
79 pub source_head: String,
82}
83
84#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
95#[serde(rename_all = "kebab-case")]
96pub enum FindingClass {
97 Drifted,
100 Wrong,
102 Uncovered,
104 UnresolvableAnchor,
106 QueuedForAdjudication,
109}
110
111impl FindingClass {
112 pub const WIRE_VALUES: &'static [&'static str] = &[
114 "drifted",
115 "wrong",
116 "uncovered",
117 "unresolvable-anchor",
118 "queued-for-adjudication",
119 ];
120
121 pub fn as_wire(&self) -> &'static str {
123 match self {
124 FindingClass::Drifted => "drifted",
125 FindingClass::Wrong => "wrong",
126 FindingClass::Uncovered => "uncovered",
127 FindingClass::UnresolvableAnchor => "unresolvable-anchor",
128 FindingClass::QueuedForAdjudication => "queued-for-adjudication",
129 }
130 }
131
132 pub fn from_wire(s: &str) -> Option<Self> {
134 match s {
135 "drifted" => Some(FindingClass::Drifted),
136 "wrong" => Some(FindingClass::Wrong),
137 "uncovered" => Some(FindingClass::Uncovered),
138 "unresolvable-anchor" => Some(FindingClass::UnresolvableAnchor),
139 "queued-for-adjudication" => Some(FindingClass::QueuedForAdjudication),
140 _ => None,
141 }
142 }
143}
144
145#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
148#[serde(tag = "kind", rename_all = "kebab-case")]
149pub enum FindingTarget {
150 Anchor {
153 entity: String,
155 artifact: String,
157 },
158 Artifact {
161 artifact: String,
163 },
164}
165
166#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
171pub struct Finding {
172 pub key: FindingKey,
176 pub facet: String,
179 pub target: FindingTarget,
181 pub class: FindingClass,
183 pub detail: String,
185 pub created_at: String,
187}
188
189#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
198pub struct FindingsBatch {
199 pub key: FindingKey,
201 pub recorded_at: String,
203 pub findings: Vec<Finding>,
205}
206
207#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
213pub struct FindingsStore {
214 pub binding: String,
216 #[serde(default)]
219 pub batches: Vec<FindingsBatch>,
220}
221
222impl FindingsStore {
223 pub fn record(&mut self, key: FindingKey, recorded_at: String, findings: Vec<Finding>) {
227 if let Some(batch) = self.batches.iter_mut().find(|b| b.key == key) {
228 batch.recorded_at = recorded_at;
229 batch.findings = findings;
230 } else {
231 self.batches.push(FindingsBatch {
232 key,
233 recorded_at,
234 findings,
235 });
236 }
237 }
238
239 pub fn current(&self, key: &FindingKey) -> &[Finding] {
242 self.batches
243 .iter()
244 .find(|b| &b.key == key)
245 .map(|b| b.findings.as_slice())
246 .unwrap_or(&[])
247 }
248
249 pub fn superseded(&self, key: &FindingKey) -> Vec<&Finding> {
253 self.batches
254 .iter()
255 .filter(|b| &b.key != key)
256 .flat_map(|b| b.findings.iter())
257 .collect()
258 }
259}
260
261pub fn findings_store_path(workspace_root: &Path, mem: &str, name: &str) -> PathBuf {
268 workspace_root
269 .join(WORKSPACE_STORE_DIR)
270 .join(STATE_DIR)
271 .join(FINDINGS_DIR)
272 .join(mem)
273 .join(format!("{name}.json"))
274}
275
276pub fn read_findings_store(
279 workspace_root: &Path,
280 mem: &str,
281 name: &str,
282) -> Result<Option<FindingsStore>, StoreError> {
283 let path = findings_store_path(workspace_root, mem, name);
284 match std::fs::read(&path) {
285 Ok(bytes) => serde_json::from_slice(&bytes)
286 .map(Some)
287 .map_err(|e| StoreError::Parse {
288 path,
289 message: e.to_string(),
290 }),
291 Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None),
292 Err(e) => Err(StoreError::Io { path, source: e }),
293 }
294}
295
296pub fn write_findings_store(
299 workspace_root: &Path,
300 mem: &str,
301 name: &str,
302 store: &FindingsStore,
303) -> Result<(), StoreError> {
304 let path = findings_store_path(workspace_root, mem, name);
305 if let Some(parent) = path.parent() {
306 std::fs::create_dir_all(parent).map_err(|e| StoreError::Io {
307 path: parent.to_path_buf(),
308 source: e,
309 })?;
310 }
311 let bytes = serde_json::to_vec_pretty(store).map_err(|e| StoreError::Parse {
312 path: path.clone(),
313 message: e.to_string(),
314 })?;
315 std::fs::write(&path, bytes).map_err(|e| StoreError::Io { path, source: e })
316}
317
318pub fn delete_findings_store(
321 workspace_root: &Path,
322 mem: &str,
323 name: &str,
324) -> Result<(), StoreError> {
325 let path = findings_store_path(workspace_root, mem, name);
326 match std::fs::remove_file(&path) {
327 Ok(()) => Ok(()),
328 Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
329 Err(e) => Err(StoreError::Io { path, source: e }),
330 }
331}
332
333#[derive(Debug, thiserror::Error)]
339pub enum FindingsError {
340 #[error("malformed binding id '{0}': expected `<mem>/<stem>`")]
342 MalformedId(String),
343 #[error("findings store error: {0}")]
345 Store(#[source] StoreError),
346}
347
348#[derive(Debug, Clone, PartialEq, Eq)]
350pub struct VerifyOutcome {
351 pub binding: String,
353 pub key: FindingKey,
355 pub recorded: usize,
357 pub superseded: usize,
359 pub backlog: usize,
361 pub full_resync: FullResyncDecision,
365}
366
367fn split_binding_id(binding_id: &str) -> Result<(String, String), FindingsError> {
371 binding_id
372 .split_once('/')
373 .filter(|(m, n)| is_single_component(m) && is_single_component(n))
374 .map(|(m, n)| (m.to_string(), n.to_string()))
375 .ok_or_else(|| FindingsError::MalformedId(binding_id.to_string()))
376}
377
378fn source_facet_label(resolved: &ResolvedIngest) -> String {
382 let facets: Vec<&str> = resolved
383 .sources
384 .iter()
385 .filter_map(|s| match s {
386 ResolvedSource::Primary(p) => Some(p.facet_ref.as_str()),
387 ResolvedSource::Reference { .. } => None,
388 })
389 .collect();
390 facets.join(",")
391}
392
393fn now_seconds() -> String {
395 let secs = SystemTime::now()
396 .duration_since(UNIX_EPOCH)
397 .map(|d| d.as_secs())
398 .unwrap_or(0);
399 secs.to_string()
400}
401
402fn current_source_head(
409 engine: &Engine,
410 workspace_root: &Path,
411 resolved: &ResolvedIngest,
412) -> String {
413 let binding_id = &resolved.name;
414 let prefix = format!("{binding_id}/");
415 let mut tokens: BTreeMap<String, String> = BTreeMap::new();
416
417 if let Some(cfg) = engine.mem_config_for(&resolved.destination_mem) {
419 for (k, v) in &cfg.sync_state {
420 if let Some(rest) = k.strip_prefix(&prefix)
421 && let Some(facet) = rest.strip_suffix("#synced")
422 {
423 tokens.insert(facet.to_string(), v.clone());
424 }
425 }
426 }
427
428 let cursor = compute_source_cursor(engine, resolved, workspace_root);
430 for c in cursor.write_commands.iter().chain(cursor.reseed.iter()) {
431 if let Some(rest) = c.key.strip_prefix(&prefix)
432 && let Some(facet) = rest.strip_suffix("#synced")
433 {
434 tokens.insert(facet.to_string(), c.token.clone());
435 }
436 }
437
438 tokens
439 .iter()
440 .map(|(facet, token)| format!("{facet}={token}"))
441 .collect::<Vec<_>>()
442 .join(";")
443}
444
445fn current_key(
447 engine: &Engine,
448 workspace_root: &Path,
449 binding: &BindingV1,
450 resolved: &ResolvedIngest,
451) -> FindingKey {
452 let primary_sources = resolved
453 .sources
454 .iter()
455 .filter_map(|s| match s {
456 ResolvedSource::Primary(p) => Some(p.clone()),
457 ResolvedSource::Reference { .. } => None,
458 })
459 .collect();
460 let rb = ResolvedBinding {
461 binding: binding.clone(),
462 primary_sources,
463 };
464 FindingKey {
465 binding_hash: hash_binding(&rb),
466 source_head: current_source_head(engine, workspace_root, resolved),
467 }
468}
469
470pub fn current_findings(
477 engine: &Engine,
478 workspace_root: &Path,
479 binding: &BindingV1,
480 resolved: &ResolvedIngest,
481) -> Result<(FindingKey, Vec<Finding>), FindingsError> {
482 let (mem, name) = split_binding_id(&resolved.name)?;
483 let key = current_key(engine, workspace_root, binding, resolved);
484 let findings = read_findings_store(workspace_root, &mem, &name)
485 .map_err(FindingsError::Store)?
486 .map(|s| s.current(&key).to_vec())
487 .unwrap_or_default();
488 Ok((key, findings))
489}
490
491pub fn adjudicate_anchor(
502 key: &FindingKey,
503 facet: &str,
504 entity: &str,
505 anchor: &Anchor,
506 state: AnchorState,
507 created_at: &str,
508) -> Option<Finding> {
509 let (class, detail) = match state {
510 AnchorState::Resolves => return None,
511 AnchorState::Orphaned => (
512 FindingClass::UnresolvableAnchor,
513 format!(
514 "artifact '{}' the anchor references is no longer present in the medium",
515 anchor.artifact
516 ),
517 ),
518 AnchorState::Drifted | AnchorState::Recheck => {
519 if !anchor.class.is_hash_bearing() {
521 return None;
522 }
523 match state {
524 AnchorState::Drifted => (
525 FindingClass::Drifted,
526 format!(
527 "prepared-content hash of '{}' drifted from the anchored hash",
528 anchor.artifact
529 ),
530 ),
531 _ => (
532 FindingClass::QueuedForAdjudication,
533 format!(
534 "hash adjudication of '{}' deferred (recheck); queued",
535 anchor.artifact
536 ),
537 ),
538 }
539 }
540 };
541 Some(Finding {
542 key: key.clone(),
543 facet: facet.to_string(),
544 target: FindingTarget::Anchor {
545 entity: entity.to_string(),
546 artifact: anchor.artifact.clone(),
547 },
548 class,
549 detail,
550 created_at: created_at.to_string(),
551 })
552}
553
554#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
561pub struct FacetEnumerability {
562 pub facet: String,
564 pub medium_type: String,
566 pub enumerable: bool,
568}
569
570#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
575pub struct FullResyncRefusal {
576 pub facet: String,
578 pub medium_type: String,
580 pub reason: String,
582}
583
584#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
588#[serde(tag = "state", rename_all = "kebab-case")]
589pub enum FullResyncDecision {
590 Disabled,
593 NotDue {
596 run_count: u64,
598 every: u32,
600 runs_until_due: u32,
602 },
603 Due {
608 run_count: u64,
610 every: u32,
612 walked_facets: Vec<String>,
614 refused: Vec<FullResyncRefusal>,
616 },
617}
618
619impl FullResyncDecision {
620 pub fn is_full_walk(&self) -> bool {
623 matches!(self, FullResyncDecision::Due { .. })
624 }
625}
626
627pub fn schedule_full_resync(
633 every: u32,
634 run_count: u64,
635 facets: &[FacetEnumerability],
636) -> FullResyncDecision {
637 if every == 0 {
638 return FullResyncDecision::Disabled;
639 }
640 let modulo = run_count % u64::from(every);
641 if modulo != 0 {
642 return FullResyncDecision::NotDue {
643 run_count,
644 every,
645 runs_until_due: (u64::from(every) - modulo) as u32,
646 };
647 }
648 let mut walked_facets = Vec::new();
649 let mut refused = Vec::new();
650 for f in facets {
651 if f.enumerable {
652 walked_facets.push(f.facet.clone());
653 } else {
654 refused.push(FullResyncRefusal {
655 facet: f.facet.clone(),
656 medium_type: f.medium_type.clone(),
657 reason: format!(
658 "medium type '{}' is non-enumerable — a full-enumeration walk cannot cover \
659 it; the scheduled full resync refuses rather than claim full coverage",
660 f.medium_type
661 ),
662 });
663 }
664 }
665 FullResyncDecision::Due {
666 run_count,
667 every,
668 walked_facets,
669 refused,
670 }
671}
672
673fn candidate_key(entity: &str, anchor: &Anchor) -> String {
677 format!("{entity}\u{1f}{}", anchor.artifact)
678}
679
680fn adjudicate_candidates(
692 key: &FindingKey,
693 facet: &str,
694 candidates: &[(String, Anchor, AnchorState)],
695 window: Option<&BTreeSet<String>>,
696 created_at: &str,
697) -> Vec<Finding> {
698 let mut out = Vec::new();
699 for (entity, anchor, state) in candidates {
700 let ck = candidate_key(entity, anchor);
701 let adjudicate_now = window.is_none_or(|w| w.contains(&ck));
702 if adjudicate_now {
703 if let Some(f) = adjudicate_anchor(key, facet, entity, anchor, *state, created_at) {
704 out.push(f);
705 }
706 } else {
707 out.push(Finding {
711 key: key.clone(),
712 facet: facet.to_string(),
713 target: FindingTarget::Anchor {
714 entity: entity.clone(),
715 artifact: anchor.artifact.clone(),
716 },
717 class: FindingClass::QueuedForAdjudication,
718 detail: format!(
719 "adjudication of '{}' deferred (per-run adjudication cap reached); queued",
720 anchor.artifact
721 ),
722 created_at: created_at.to_string(),
723 });
724 }
725 }
726 out
727}
728
729pub fn verify_binding(
743 engine: &Engine,
744 workspace_root: &Path,
745 binding: &BindingV1,
746 resolved: &ResolvedIngest,
747) -> Result<VerifyOutcome, FindingsError> {
748 let binding_id = resolved.name.clone();
749 let (mem, name) = split_binding_id(&binding_id)?;
750
751 let key = current_key(engine, workspace_root, binding, resolved);
752 let now = now_seconds();
753 let facet = source_facet_label(resolved);
754 let cache_root = workspace_root.join(".memstead.cache").join("ingest");
755
756 let verify_op = binding.operations.verify.as_ref();
762 let cap = verify_op.map_or(DEFAULT_ADJUDICATION_CAP, |v| v.adjudication_cap);
763 let full_resync_every = verify_op.map_or(DEFAULT_FULL_RESYNC_EVERY, |v| v.full_resync_every);
764 let sample_batch = verify_op
765 .map_or(resolved.batch_size, |v| v.batch_size)
766 .max(1) as usize;
767
768 let run_count = bump_verify_runs(&cache_root, &binding_id);
771 let facet_enum: Vec<FacetEnumerability> = resolved
772 .sources
773 .iter()
774 .filter_map(|s| match s {
775 ResolvedSource::Primary(p) => Some(FacetEnumerability {
776 facet: p.facet_ref.clone(),
777 medium_type: medium_type_wire(p.medium_type),
778 enumerable: medium_capabilities(p.medium_type).enumerable,
779 }),
780 ResolvedSource::Reference { .. } => None,
781 })
782 .collect();
783 let full_resync = schedule_full_resync(full_resync_every, run_count, &facet_enum);
784
785 let mut findings: Vec<Finding> = Vec::new();
786
787 let mut existence: Vec<(String, Anchor, AnchorState)> = Vec::new();
794 let mut candidates: Vec<(String, Anchor, AnchorState)> = Vec::new();
795 for (eid, resolved_anchor) in engine.mem_anchors_resolved(&resolved.destination_mem) {
796 let Some(state) = resolved_anchor.state else {
797 continue;
798 };
799 let anchor = resolved_anchor.anchor;
800 match state {
801 AnchorState::Resolves => {}
802 AnchorState::Orphaned => existence.push((eid.as_ref().to_string(), anchor, state)),
803 AnchorState::Drifted | AnchorState::Recheck => {
804 if anchor.class.is_hash_bearing() {
807 candidates.push((eid.as_ref().to_string(), anchor, state));
808 }
809 }
810 }
811 }
812 for (entity, anchor, state) in &existence {
813 if let Some(f) = adjudicate_anchor(&key, &facet, entity, anchor, *state, &now) {
814 findings.push(f);
815 }
816 }
817 let window: Option<BTreeSet<String>> = if cap == 0 {
820 None
821 } else {
822 let mut keys: Vec<String> = candidates
823 .iter()
824 .map(|(e, a, _)| candidate_key(e, a))
825 .collect();
826 keys.sort();
827 keys.dedup();
828 next_rotation_batch(
829 &cache_root,
830 &binding_id,
831 ROTATION_ANCHOR_ADJUDICATION,
832 keys,
833 cap as usize,
834 )
835 .map(|b| b.files.into_iter().collect())
836 };
837 findings.extend(adjudicate_candidates(
838 &key,
839 &facet,
840 &candidates,
841 window.as_ref(),
842 &now,
843 ));
844
845 let sample_files: Vec<String> = if full_resync.is_full_walk() {
851 let mut all: Vec<String> = Vec::new();
852 for source in &resolved.sources {
853 if let ResolvedSource::Primary(p) = source
854 && medium_capabilities(p.medium_type).enumerable
855 {
856 all.extend(enumerate_facet_files(
857 p,
858 &resolved.deny_paths,
859 workspace_root,
860 ));
861 }
862 }
863 all.sort();
864 all.dedup();
865 all
866 } else {
867 next_batch(resolved, workspace_root, &cache_root, sample_batch)
868 .map(|b| b.files)
869 .unwrap_or_default()
870 };
871 for file in sample_files {
872 let covered = engine
873 .anchors_referencing_artifact(&file)
874 .iter()
875 .any(|(eid, _)| eid.mem() == resolved.destination_mem.as_str());
876 if !covered {
877 findings.push(Finding {
878 key: key.clone(),
879 facet: facet.clone(),
880 target: FindingTarget::Artifact { artifact: file },
881 class: FindingClass::Uncovered,
882 detail: "source artifact in scope has no anchor in the destination mem".to_string(),
883 created_at: now.clone(),
884 });
885 }
886 }
887
888 let backlog = findings
889 .iter()
890 .filter(|f| f.class == FindingClass::QueuedForAdjudication)
891 .count();
892
893 let mut store = read_findings_store(workspace_root, &mem, &name)
896 .map_err(FindingsError::Store)?
897 .unwrap_or_else(|| FindingsStore {
898 binding: binding_id.clone(),
899 ..Default::default()
900 });
901 let recorded = findings.len();
902 store.record(key.clone(), now, findings);
903 let superseded = store.superseded(&key).len();
904 write_findings_store(workspace_root, &mem, &name, &store).map_err(FindingsError::Store)?;
905
906 Ok(VerifyOutcome {
907 binding: binding_id,
908 key,
909 recorded,
910 superseded,
911 backlog,
912 full_resync,
913 })
914}
915
916fn medium_type_wire(t: crate::pipeline::MediumType) -> String {
919 serde_json::to_value(t)
920 .ok()
921 .and_then(|v| v.as_str().map(str::to_string))
922 .unwrap_or_default()
923}
924
925#[cfg(test)]
926mod tests {
927 use super::*;
928 use crate::anchor::{Anchor, AnchorGrain, AnchorHashStability, AnchorProvenanceClass};
929
930 fn key(hash: &str, head: &str) -> FindingKey {
931 FindingKey {
932 binding_hash: hash.to_string(),
933 source_head: head.to_string(),
934 }
935 }
936
937 fn anchor(class: AnchorProvenanceClass) -> Anchor {
938 Anchor {
939 artifact: "src/lib.rs".to_string(),
940 grain: AnchorGrain::File,
941 class,
942 at_version: None,
943 hash: if class.is_hash_bearing() {
944 Some("h1".to_string())
945 } else {
946 None
947 },
948 hash_stability: AnchorHashStability::Stable,
949 derived_from: Vec::new(),
950 binding: None,
951 }
952 }
953
954 #[test]
957 fn store_round_trips_on_disk_and_delete_is_idempotent() {
958 let tmp = tempfile::tempdir().unwrap();
959 let root = tmp.path();
960 assert!(
961 read_findings_store(root, "engine", "graph")
962 .unwrap()
963 .is_none()
964 );
965
966 let mut store = FindingsStore {
967 binding: "engine/graph".to_string(),
968 ..Default::default()
969 };
970 let k = key("hashA", "head1");
971 store.record(
972 k.clone(),
973 "1".to_string(),
974 vec![Finding {
975 key: k.clone(),
976 facet: "src".to_string(),
977 target: FindingTarget::Artifact {
978 artifact: "src/a.rs".to_string(),
979 },
980 class: FindingClass::Uncovered,
981 detail: "d".to_string(),
982 created_at: "1".to_string(),
983 }],
984 );
985 write_findings_store(root, "engine", "graph", &store).unwrap();
986 assert!(findings_store_path(root, "engine", "graph").exists());
987
988 let back = read_findings_store(root, "engine", "graph")
990 .unwrap()
991 .unwrap();
992 assert_eq!(back, store);
993 assert_eq!(back.current(&k).len(), 1);
994
995 delete_findings_store(root, "engine", "graph").unwrap();
996 assert!(
997 read_findings_store(root, "engine", "graph")
998 .unwrap()
999 .is_none()
1000 );
1001 delete_findings_store(root, "engine", "graph").unwrap();
1003 }
1004
1005 #[test]
1008 fn changed_binding_hash_supersedes_prior_findings() {
1009 let mut store = FindingsStore::default();
1010 let old = key("hashOLD", "head1");
1011 let new = key("hashNEW", "head1");
1012 let f_old = Finding {
1013 key: old.clone(),
1014 facet: "src".to_string(),
1015 target: FindingTarget::Artifact {
1016 artifact: "src/old.rs".to_string(),
1017 },
1018 class: FindingClass::Uncovered,
1019 detail: "old".to_string(),
1020 created_at: "1".to_string(),
1021 };
1022 store.record(old.clone(), "1".to_string(), vec![f_old.clone()]);
1023
1024 store.record(new.clone(), "2".to_string(), Vec::new());
1026 assert!(store.current(&new).is_empty(), "new key has its own view");
1027 let superseded = store.superseded(&new);
1028 assert_eq!(superseded.len(), 1, "old batch is segregated as superseded");
1029 assert_eq!(superseded[0], &f_old);
1030 assert!(!store.current(&new).contains(&f_old));
1032 }
1033
1034 #[test]
1037 fn moved_source_head_supersedes_prior_findings() {
1038 let mut store = FindingsStore::default();
1039 let before = key("hashA", "head1");
1040 let after = key("hashA", "head2");
1041 let f = Finding {
1042 key: before.clone(),
1043 facet: "src".to_string(),
1044 target: FindingTarget::Anchor {
1045 entity: "engine--e".to_string(),
1046 artifact: "src/x.rs".to_string(),
1047 },
1048 class: FindingClass::UnresolvableAnchor,
1049 detail: "gone".to_string(),
1050 created_at: "1".to_string(),
1051 };
1052 store.record(before.clone(), "1".to_string(), vec![f.clone()]);
1053 store.record(after.clone(), "2".to_string(), Vec::new());
1054
1055 assert!(store.current(&after).is_empty());
1056 assert_eq!(store.superseded(&after), vec![&f]);
1057 store.record(after.clone(), "3".to_string(), Vec::new());
1059 assert_eq!(store.batches.len(), 2, "one batch per distinct key");
1060 }
1061
1062 #[test]
1065 fn informed_by_anchor_never_drifts() {
1066 let k = key("h", "s");
1067 for class in [
1068 AnchorProvenanceClass::InformedBy,
1069 AnchorProvenanceClass::Authored,
1070 ] {
1071 let a = anchor(class);
1072 assert!(
1073 adjudicate_anchor(&k, "f", "engine--e", &a, AnchorState::Drifted, "1").is_none(),
1074 "{class:?} must not produce a drift finding"
1075 );
1076 assert!(
1077 adjudicate_anchor(&k, "f", "engine--e", &a, AnchorState::Recheck, "1").is_none(),
1078 "{class:?} must not produce a queued finding"
1079 );
1080 }
1081 }
1082
1083 #[test]
1086 fn hash_bearing_drifts_and_orphan_is_class_independent() {
1087 let k = key("h", "s");
1088 let anchored = anchor(AnchorProvenanceClass::Anchored);
1089 let drifted =
1090 adjudicate_anchor(&k, "f", "engine--e", &anchored, AnchorState::Drifted, "1").unwrap();
1091 assert_eq!(drifted.class, FindingClass::Drifted);
1092 assert_eq!(drifted.key, k, "the finding carries its recording key (A2)");
1093
1094 let queued =
1095 adjudicate_anchor(&k, "f", "engine--e", &anchored, AnchorState::Recheck, "1").unwrap();
1096 assert_eq!(queued.class, FindingClass::QueuedForAdjudication);
1097
1098 let informed = anchor(AnchorProvenanceClass::InformedBy);
1100 let orphan =
1101 adjudicate_anchor(&k, "f", "engine--e", &informed, AnchorState::Orphaned, "1").unwrap();
1102 assert_eq!(orphan.class, FindingClass::UnresolvableAnchor);
1103
1104 assert!(
1106 adjudicate_anchor(&k, "f", "engine--e", &anchored, AnchorState::Resolves, "1")
1107 .is_none()
1108 );
1109 }
1110
1111 #[test]
1113 fn finding_class_wire_round_trips() {
1114 for w in FindingClass::WIRE_VALUES {
1115 let c = FindingClass::from_wire(w).expect("known wire value");
1116 assert_eq!(c.as_wire(), *w);
1117 }
1118 assert!(FindingClass::from_wire("nonsense").is_none());
1119 }
1120
1121 #[test]
1123 fn malformed_binding_id_refuses() {
1124 assert!(matches!(
1125 split_binding_id("../escape"),
1126 Err(FindingsError::MalformedId(_))
1127 ));
1128 assert!(matches!(
1129 split_binding_id("no-slash"),
1130 Err(FindingsError::MalformedId(_))
1131 ));
1132 assert_eq!(
1133 split_binding_id("engine/graph").unwrap(),
1134 ("engine".to_string(), "graph".to_string())
1135 );
1136 }
1137
1138 use crate::anchor::AnchorSidecar;
1141 use crate::binding::{
1142 BINDING_VERSION, BuildMode, BuildOperation, CoverageSemantics, DEFAULT_ADJUDICATION_CAP,
1143 DEFAULT_FULL_RESYNC_EVERY, Operations, VerifyOperation,
1144 };
1145 use crate::ingest::resolve::resolve_binding_run;
1146 use crate::pipeline::{Facet, IngestTrigger, Medium, MediumType, PatternEntry, PatternMode};
1147 use crate::pipeline_store::{load_pipeline_configs, write_binding, write_facet, write_medium};
1148 use crate::workspace::{
1149 Mount, MountCapability, MountLifecycle, MountStorage, Workspace, WorkspaceSettings,
1150 };
1151 use crate::workspace_store::WorkspaceStoreAdapter;
1152
1153 #[test]
1160 fn verify_persists_findings_readable_fresh() {
1161 let tmp = tempfile::tempdir().unwrap();
1162 let root = tmp.path();
1163 let mem_dir = root.join("mem");
1164 std::fs::create_dir_all(mem_dir.join(".memstead")).unwrap();
1165 std::fs::write(
1166 mem_dir.join(".memstead").join("config.json"),
1167 r#"{"format":1,"schema":"default@1.0.0","version":"1.0.0"}"#,
1168 )
1169 .unwrap();
1170
1171 std::fs::create_dir_all(root.join(".memstead")).unwrap();
1174 std::fs::write(
1175 root.join(".memstead").join("workspace.toml"),
1176 "format = \"memstead-git-branch-2\"\n\n[persistence_adapter]\nname = \"file-two-layer\"\n",
1177 )
1178 .unwrap();
1179 let mount = Mount {
1180 mem: "engine".to_string(),
1181 schema: Some("default@1.0.0".parse().unwrap()),
1182 storage: MountStorage::Folder {
1183 path: mem_dir.clone(),
1184 },
1185 capability: MountCapability::Write,
1186 lifecycle: MountLifecycle::Eager,
1187 cross_linkable: false,
1188 migration_target: None,
1189 };
1190 crate::FileWorkspaceStore::new()
1191 .save_state(
1192 root,
1193 &Workspace {
1194 mounts: vec![mount],
1195 settings: WorkspaceSettings::default(),
1196 },
1197 )
1198 .unwrap();
1199
1200 let out = std::process::Command::new("git")
1204 .args(["init", "-q"])
1205 .current_dir(root)
1206 .output()
1207 .unwrap();
1208 assert!(out.status.success());
1209 std::fs::create_dir_all(root.join("src")).unwrap();
1210 std::fs::write(root.join("src").join("present.rs"), "fn a() {}\n").unwrap();
1211 std::fs::write(root.join("src").join("uncovered.rs"), "fn b() {}\n").unwrap();
1212
1213 let mk = |artifact: &str, class: AnchorProvenanceClass| Anchor {
1216 artifact: artifact.to_string(),
1217 grain: AnchorGrain::File,
1218 class,
1219 at_version: None,
1220 hash: class.is_hash_bearing().then(|| "recorded".to_string()),
1221 hash_stability: AnchorHashStability::Stable,
1222 derived_from: Vec::new(),
1223 binding: None,
1224 };
1225 let mut sidecar = AnchorSidecar::default();
1226 sidecar.set(
1227 "engine--e",
1228 vec![
1229 mk("src/present.rs", AnchorProvenanceClass::Anchored), mk("src/gone.rs", AnchorProvenanceClass::Anchored), mk("src/present.rs", AnchorProvenanceClass::InformedBy), ],
1233 );
1234 std::fs::write(
1235 mem_dir.join(crate::anchor::ANCHOR_SIDECAR_PATH),
1236 sidecar.to_bytes(),
1237 )
1238 .unwrap();
1239
1240 write_medium(
1242 root,
1243 "engine",
1244 "graph",
1245 &Medium {
1246 name: "graph".to_string(),
1247 medium_type: MediumType::Codebase,
1248 pointer: String::new(),
1249 change_detection: Some("git".to_string()),
1250 },
1251 )
1252 .unwrap();
1253 write_facet(
1254 root,
1255 "engine",
1256 "graph",
1257 &Facet {
1258 name: "graph".to_string(),
1259 medium: "graph".to_string(),
1260 scope: vec![PatternEntry {
1261 path: "src/**/*.rs".to_string(),
1262 mode: PatternMode::Allow,
1263 }],
1264 engagement: None,
1265 preparation: None,
1266 },
1267 )
1268 .unwrap();
1269 write_binding(
1270 root,
1271 "engine",
1272 "graph",
1273 &BindingV1 {
1274 version: BINDING_VERSION,
1275 intent: None,
1276 source_facets: vec!["graph".to_string()],
1277 reference_mems: Vec::new(),
1278 destination_mem: "engine".to_string(),
1279 deny_paths: Vec::new(),
1280 coverage_semantics: CoverageSemantics::Exhaustive,
1281 rules: None,
1282 prune: None,
1283 operations: Operations {
1284 build: Some(BuildOperation {
1285 mode: BuildMode::Discovery,
1286 trigger: IngestTrigger::Loop,
1287 batch_size: 20,
1288 post_actions: None,
1289 }),
1290 sync: None,
1291 verify: Some(VerifyOperation {
1292 trigger: IngestTrigger::Manual,
1293 batch_size: 20,
1294 adjudication_cap: DEFAULT_ADJUDICATION_CAP,
1295 full_resync_every: DEFAULT_FULL_RESYNC_EVERY,
1296 }),
1297 },
1298 },
1299 )
1300 .unwrap();
1301
1302 let engine = Engine::from_workspace_root(root).unwrap();
1303
1304 let configs = load_pipeline_configs(root).unwrap();
1305 let binding = &configs.bindings[0].config;
1306 let resolved = resolve_binding_run(&configs, "engine/graph", binding).unwrap();
1307
1308 let outcome = verify_binding(&engine, root, binding, &resolved).unwrap();
1310 assert!(
1311 outcome.recorded >= 3,
1312 "orphan + queued + uncovered at least"
1313 );
1314 assert_eq!(outcome.superseded, 0, "no prior key yet");
1315 assert_eq!(outcome.backlog, 1, "the present hash-bearing anchor queued");
1316
1317 let store = read_findings_store(root, "engine", "graph")
1319 .unwrap()
1320 .unwrap();
1321 let current = store.current(&outcome.key);
1322 assert_eq!(current.len(), outcome.recorded);
1323
1324 let has = |c: FindingClass, art: &str| {
1325 current.iter().any(|f| {
1326 f.class == c
1327 && match &f.target {
1328 FindingTarget::Anchor { artifact, .. } => artifact == art,
1329 FindingTarget::Artifact { artifact } => artifact == art,
1330 }
1331 })
1332 };
1333 assert!(has(FindingClass::UnresolvableAnchor, "src/gone.rs"));
1334 assert!(has(FindingClass::QueuedForAdjudication, "src/present.rs"));
1335 assert!(has(FindingClass::Uncovered, "src/uncovered.rs"));
1336 assert!(
1338 !current
1339 .iter()
1340 .any(|f| f.class == FindingClass::Drifted || f.class == FindingClass::Wrong),
1341 "no drift finding from a non-hash / present-clean anchor"
1342 );
1343 assert!(!has(FindingClass::Uncovered, "src/present.rs"));
1345 }
1346
1347 #[test]
1354 fn adjudication_cap_queues_the_remainder() {
1355 let k = key("h", "s");
1356 let mk = |art: &str| {
1357 let mut a = anchor(AnchorProvenanceClass::Anchored);
1358 a.artifact = art.to_string();
1359 a
1360 };
1361 let candidates = vec![
1362 (
1363 "engine--a".to_string(),
1364 mk("src/a.rs"),
1365 AnchorState::Drifted,
1366 ),
1367 (
1368 "engine--b".to_string(),
1369 mk("src/b.rs"),
1370 AnchorState::Drifted,
1371 ),
1372 (
1373 "engine--c".to_string(),
1374 mk("src/c.rs"),
1375 AnchorState::Drifted,
1376 ),
1377 ];
1378 let window: BTreeSet<String> = [candidate_key("engine--a", &mk("src/a.rs"))]
1380 .into_iter()
1381 .collect();
1382 let out = adjudicate_candidates(&k, "f", &candidates, Some(&window), "1");
1383 let drifted = out
1384 .iter()
1385 .filter(|f| f.class == FindingClass::Drifted)
1386 .count();
1387 let queued = out
1388 .iter()
1389 .filter(|f| f.class == FindingClass::QueuedForAdjudication)
1390 .count();
1391 assert_eq!(drifted, 1, "only the in-window candidate is adjudicated");
1392 assert_eq!(queued, 2, "the remainder is queued as the tier-3 backlog");
1393 assert!(
1395 out.iter()
1396 .any(|f| f.class == FindingClass::QueuedForAdjudication
1397 && f.detail.contains("cap reached")),
1398 "capped remainder states it was deferred by the cap"
1399 );
1400
1401 let uncapped = adjudicate_candidates(&k, "f", &candidates, None, "1");
1403 assert_eq!(
1404 uncapped
1405 .iter()
1406 .filter(|f| f.class == FindingClass::Drifted)
1407 .count(),
1408 3,
1409 "uncapped adjudicates every candidate"
1410 );
1411 assert_eq!(
1412 uncapped
1413 .iter()
1414 .filter(|f| f.class == FindingClass::QueuedForAdjudication)
1415 .count(),
1416 0
1417 );
1418 }
1419
1420 #[test]
1426 fn full_resync_schedule_disabled_notdue_due() {
1427 let codebase = FacetEnumerability {
1428 facet: "src".to_string(),
1429 medium_type: "codebase".to_string(),
1430 enumerable: true,
1431 };
1432 assert_eq!(
1433 schedule_full_resync(0, 5, std::slice::from_ref(&codebase)),
1434 FullResyncDecision::Disabled
1435 );
1436 match schedule_full_resync(3, 2, std::slice::from_ref(&codebase)) {
1437 FullResyncDecision::NotDue { runs_until_due, .. } => assert_eq!(runs_until_due, 1),
1438 other => panic!("expected NotDue, got {other:?}"),
1439 }
1440 match schedule_full_resync(3, 3, std::slice::from_ref(&codebase)) {
1441 FullResyncDecision::Due {
1442 walked_facets,
1443 refused,
1444 ..
1445 } => {
1446 assert_eq!(walked_facets, vec!["src".to_string()]);
1447 assert!(refused.is_empty(), "enumerable facet is not refused");
1448 }
1449 other => panic!("expected Due, got {other:?}"),
1450 }
1451 }
1452
1453 #[test]
1456 fn full_resync_refuses_non_enumerable_medium() {
1457 let web = FacetEnumerability {
1458 facet: "manual".to_string(),
1459 medium_type: "web".to_string(),
1460 enumerable: false,
1461 };
1462 let d = schedule_full_resync(1, 1, &[web]);
1463 assert!(
1464 d.is_full_walk(),
1465 "a due sweep is a full walk even when refused"
1466 );
1467 match d {
1468 FullResyncDecision::Due {
1469 walked_facets,
1470 refused,
1471 ..
1472 } => {
1473 assert!(walked_facets.is_empty(), "nothing enumerable to walk");
1474 assert_eq!(refused.len(), 1, "the non-enumerable facet is refused");
1475 assert_eq!(refused[0].facet, "manual");
1476 assert_eq!(refused[0].medium_type, "web");
1477 assert!(
1478 refused[0].reason.contains("non-enumerable"),
1479 "the refusal is typed and states why"
1480 );
1481 }
1482 other => panic!("expected Due with a refusal, got {other:?}"),
1483 }
1484 }
1485
1486 #[test]
1491 fn full_resync_full_walk_covers_whole_source() {
1492 let tmp = tempfile::tempdir().unwrap();
1493 let root = tmp.path();
1494 let mem_dir = root.join("mem");
1495 std::fs::create_dir_all(mem_dir.join(".memstead")).unwrap();
1496 std::fs::write(
1497 mem_dir.join(".memstead").join("config.json"),
1498 r#"{"format":1,"schema":"default@1.0.0","version":"1.0.0"}"#,
1499 )
1500 .unwrap();
1501 std::fs::create_dir_all(root.join(".memstead")).unwrap();
1502 std::fs::write(
1503 root.join(".memstead").join("workspace.toml"),
1504 "format = \"memstead-git-branch-2\"\n\n[persistence_adapter]\nname = \"file-two-layer\"\n",
1505 )
1506 .unwrap();
1507 let mount = Mount {
1508 mem: "engine".to_string(),
1509 schema: Some("default@1.0.0".parse().unwrap()),
1510 storage: MountStorage::Folder {
1511 path: mem_dir.clone(),
1512 },
1513 capability: MountCapability::Write,
1514 lifecycle: MountLifecycle::Eager,
1515 cross_linkable: false,
1516 migration_target: None,
1517 };
1518 crate::FileWorkspaceStore::new()
1519 .save_state(
1520 root,
1521 &Workspace {
1522 mounts: vec![mount],
1523 settings: WorkspaceSettings::default(),
1524 },
1525 )
1526 .unwrap();
1527 let out = std::process::Command::new("git")
1528 .args(["init", "-q"])
1529 .current_dir(root)
1530 .output()
1531 .unwrap();
1532 assert!(out.status.success());
1533 std::fs::create_dir_all(root.join("src")).unwrap();
1534 for f in ["a.rs", "b.rs", "c.rs"] {
1535 std::fs::write(root.join("src").join(f), "fn x() {}\n").unwrap();
1536 }
1537
1538 write_medium(
1539 root,
1540 "engine",
1541 "graph",
1542 &Medium {
1543 name: "graph".to_string(),
1544 medium_type: MediumType::Codebase,
1545 pointer: String::new(),
1546 change_detection: Some("git".to_string()),
1547 },
1548 )
1549 .unwrap();
1550 write_facet(
1551 root,
1552 "engine",
1553 "graph",
1554 &Facet {
1555 name: "graph".to_string(),
1556 medium: "graph".to_string(),
1557 scope: vec![PatternEntry {
1558 path: "src/**/*.rs".to_string(),
1559 mode: PatternMode::Allow,
1560 }],
1561 engagement: None,
1562 preparation: None,
1563 },
1564 )
1565 .unwrap();
1566 write_binding(
1567 root,
1568 "engine",
1569 "graph",
1570 &BindingV1 {
1571 version: BINDING_VERSION,
1572 intent: None,
1573 source_facets: vec!["graph".to_string()],
1574 reference_mems: Vec::new(),
1575 destination_mem: "engine".to_string(),
1576 deny_paths: Vec::new(),
1577 coverage_semantics: CoverageSemantics::Exhaustive,
1578 rules: None,
1579 prune: None,
1580 operations: Operations {
1581 build: Some(BuildOperation {
1582 mode: BuildMode::Discovery,
1583 trigger: IngestTrigger::Loop,
1584 batch_size: 20,
1585 post_actions: None,
1586 }),
1587 sync: None,
1588 verify: Some(VerifyOperation {
1589 trigger: IngestTrigger::Manual,
1590 batch_size: 1, adjudication_cap: DEFAULT_ADJUDICATION_CAP,
1592 full_resync_every: 1, }),
1594 },
1595 },
1596 )
1597 .unwrap();
1598
1599 let engine = Engine::from_workspace_root(root).unwrap();
1600 let configs = load_pipeline_configs(root).unwrap();
1601 let binding = &configs.bindings[0].config;
1602 let resolved = resolve_binding_run(&configs, "engine/graph", binding).unwrap();
1603
1604 let outcome = verify_binding(&engine, root, binding, &resolved).unwrap();
1605 match &outcome.full_resync {
1607 FullResyncDecision::Due {
1608 walked_facets,
1609 refused,
1610 run_count,
1611 ..
1612 } => {
1613 assert_eq!(*run_count, 1);
1614 assert_eq!(walked_facets, &vec!["graph".to_string()]);
1615 assert!(refused.is_empty());
1616 }
1617 other => panic!("expected a due full walk, got {other:?}"),
1618 }
1619 let store = read_findings_store(root, "engine", "graph")
1621 .unwrap()
1622 .unwrap();
1623 let uncovered = store
1624 .current(&outcome.key)
1625 .iter()
1626 .filter(|f| f.class == FindingClass::Uncovered)
1627 .count();
1628 assert_eq!(
1629 uncovered, 3,
1630 "the scheduled full walk covers the whole source, not a batch of one"
1631 );
1632 }
1633}