From 6f8089e32013fe608722b93634d620f026900ac3 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 11 Jun 2026 21:33:15 +0000 Subject: [PATCH 1/3] docs(review): deep review 2026-06-11 + targeted July-1 plan Full fresh review of the Rust core, Elixir orchestration, VQL/VCL surface, docs and CI, with build/test ground truth (295/295 Rust tests green), a claims-vs-reality audit, and a three-week wiring-not-invention plan: make the drift->normalise loop real, truth-sweep the outward claims, connect VCL-total via a minimal admissibility gate, tag v0.2.0. https://claude.ai/code/session_01E8BpV19yxhf67UrCrvkfTr --- ...026-06-11-deep-review-and-july-1-plan.adoc | 263 ++++++++++++++++++ 1 file changed, 263 insertions(+) create mode 100644 docs/reviews/2026-06-11-deep-review-and-july-1-plan.adoc diff --git a/docs/reviews/2026-06-11-deep-review-and-july-1-plan.adoc b/docs/reviews/2026-06-11-deep-review-and-july-1-plan.adoc new file mode 100644 index 0000000..750e76d --- /dev/null +++ b/docs/reviews/2026-06-11-deep-review-and-july-1-plan.adoc @@ -0,0 +1,263 @@ +// SPDX-License-Identifier: MPL-2.0 +// Copyright (c) 2026 Jonathan D.A. Jewell (hyperpolymath) + += VeriSimDB Deep Review & July-1 Plan (2026-06-11) +:toc: left +:toclevels: 2 +:sectnums: + +== Purpose and verdict + +A full fresh review of VeriSimDB (Rust core, Elixir orchestration, VQL/VCL +surface, docs, CI), conducted 2026-06-11, with a targeted three-week plan to +make the project genuinely plausible and respectable by 2026-07-01. The plan +optimises for *truth and demonstrability*, not feature count: close the gap +between what the README claims and what the code does, make the first hour +flawless for a visitor, and connect VCL-total to the engine for real. + +*Verdict in one paragraph.* VeriSimDB is a real, working, well-tested +multi-modal CRUD engine — ~41.6k LOC Rust (295 tests, all green, verified +this session), ~15.5k LOC Elixir (615 test cases), near-zero TODO debt, +REUSE-clean, 19 CI workflows. The octad coordinator, WAL, persistence layer, +planner, three protocol surfaces (REST/GraphQL/gRPC), REPL, provenance +hash-chain, and federation adapters are genuine engineering. The two +*differentiating* claims, however — drift detection and self-normalisation — +are currently disconnected at the integration layer: real drift mathematics +exists but has zero callers, and normalisation derives repairs without ever +writing them back. Both ends of every broken seam already exist; the July-1 +work is wiring, not invention. + +== What is verified real (build + test ground truth) + +* `cargo check --workspace --all-targets`: clean, 78 s (this session). +* `cargo test --workspace`: *295 passed, 0 failed* across 15 suites + (this session). +* Test culture is substantive: dedicated atomicity suite for the octad + store, proptests on the WAL, insta snapshot tests on the planner, + criterion benches, a libfuzzer target for the VQL tokenizer. +* Zero `todo!`/`unimplemented!`/`FIXME` in the Rust tree; several crates + carry `#![forbid(unsafe_code)]`. + +Per-crate reality (abridged): + +[cols="1,1,3"] +|=== +|Crate |Verdict |Notes + +|verisim-octad (4.8k LOC, 57 tests) +|REAL — strongest crate +|WAL intent -> per-modality locks -> writes -> COMMITTED marker -> replay +recovery. Rollback covers vector/document/tensor; graph/semantic rollback +gap acknowledged at `store.rs:624`. + +|verisim-wal (2.0k, 39 tests) +|REAL +|CRC32-framed segmented log, fsync modes, corrupted-tail handling, +proptests. Production-grade shape. + +|verisim-storage (1.9k, 41 tests) +|REAL +|redb abstraction consumed by six persistent backends (feature-gated; +load-all-into-RAM on open, honestly documented). + +|verisim-graph / -document / -temporal / -provenance +|REAL +|Indexed triples (oxigraph behind off-by-default flag); Tantivy; versioning; +SHA-256 hash chain with `verify_chain`. + +|verisim-vector (1.2k, 12 tests) +|REAL, mis-deployed +|Default + persistent paths are brute-force. A hand-rolled, *correct* +HNSW (660 LOC, same `VectorStore` trait) sits orphaned. + +|verisim-spatial (1.3k, 39 tests) +|REAL but brute-force +|Haversine scan; no R-tree despite docs claim (`lib.rs:17,269` admits it). + +|verisim-tensor (0.5k) +|PARTIAL +|Storage only, zero tensor ops; `burn` declared and entirely unused +(pure compile-time weight). + +|verisim-semantic (3.3k, 54 tests) +|REAL code, oversold branding +|"ZKP" = SHA-256 commitment-reveal + Merkle inclusion + R1CS constraint +checking. Sound, tested (52/52), *not zero-knowledge*. One verify arm +returns `true` unconditionally for bare commitments +(`zkp.rs:190-206`); nonce is deterministic (`zkp_bridge.rs:339`); +`verification_key` is never read (`circuit_registry.rs:268`). +`zkp_bridge.rs:10`: "Full ZK-SNARK via sanctify is designed but not yet +compiled in." + +|verisim-drift (1.2k, 11 tests) +|Real maths, ORPHANED +|`DriftCalculator` has zero callers in the workspace. +`DriftDetector::record` is never invoked by API, octad, or normalizer. + +|verisim-normalizer (3.9k, 59 tests) +|SKELETON at the seam +|Decision + derivation logic real; strategies return descriptions with +`success: true` without mutating stores; sole `ModalityRegenerator` impl +returns drift 0.0 "to signal a successful conceptual repair" +(`regeneration.rs:535-543`). KNOWN-ISSUES marks the 2026-02-12 fix +RESOLVED — true for derivation, not yet for write-back or API wiring. + +|verisim-planner (7.0k, 121 tests) +|REAL, advisory-only +|Cost model, optimizer, prepared statements, slow-query log, +EXPLAIN/EXPLAIN ANALYZE; wired into `/query/plan` + `/query/explain`. +No executor consumes `PhysicalPlan` yet — honest framing needed. + +|verisim-api (9.0k, 53 tests) +|REAL CRUD/search/VQL; theatre at drift/normalize endpoints +|REST + GraphQL + gRPC genuinely served. `/drift/status` returns +initialized-to-zero metrics forever; `/drift/entity/{id}` returns global +zeros labelled per-entity (`lib.rs:1189-1233`); `POST +/normalizer/normalize/{id}` checks existence, returns 202, does nothing +(`lib.rs:1262`); `/transactions/*` drive a manager no write path +participates in; search scores are fabricated from rank +(`lib.rs:1088,1124`: `1.0 - 0.1*i`). + +|verisim-repl (2.6k, 67 tests) +|REAL +|rustyline REPL -> HTTP `/vql/execute`; linter, formatter, highlighter, +completion, table output. + +|verisim-nif (204 LOC, 0 tests) +|100% STUB — dangerous +|Every NIF returns canned success JSON ("store integration pending"). +If `VERISIM_TRANSPORT=nif|auto` selects it, writes silently vanish. +|=== + +Elixir layer (static review; Elixir not installed in this container): +real OTP application — supervision tree boots Telemetry, EntityRegistry + +DynamicSupervisor, DriftMonitor, QueryRouter, SchemaRegistry, KRaft +consensus node (single-node bootstrap), Federation Resolver. Transport +defaults to HTTP. 14 federation adapters (237–650 LOC each) with real +modality->capability mapping and optional native wire drivers (postgrex, +redix, exqlite, bolt_sips). VQL surface exists three times over: ReScript +(`src/vql`, incl. bidirectional type checker + proof obligations), Elixir +(`query/vql_*`), and the Rust planner bridge consuming the ReScript AST +JSON. Coq models for Octad/Drift/Normalizer/Planner under `formal/` with a +CI build. + +== Where the claims outrun the code + +. *Drift is never measured on real data.* The flagship sentence of the + README describes a loop that does not run. +. *Normalisation never repairs.* Derivation without write-back; the API + endpoint is a 202 no-op. +. *Per-entity drift endpoint misreports* global zeros as entity-specific. +. *Time-travel (`at_time`) returns current state* with a version number + attached (`verisim-octad/src/store.rs:1481-1486`) — the stored + `OctadSnapshot` contains everything needed to do it honestly. +. *NIF fakes success.* +. *Search scores are rank-derived constants*, though Tantivy BM25 and + cosine scores are available. +. *"ZKP" naming* oversells commitment/Merkle/R1CS machinery. +. *"HNSW similarity search" / "R-tree spatial"* in docs vs brute-force in + both default paths (HNSW exists, unused; R-tree absent). +. *VCL-total integration is prose-only.* No code path in this repo invokes + vcl-ut; `vql-bridge/` is a single JS parser port. The README's central + triad (engine / VCL / VCL-total) has no executable seam. +. *Doc duplication/staleness:* CHANGELOG.adoc and CHANGELOG.md both exist; + KNOWN-ISSUES item 1 is over-resolved; CLAUDE.md still says `.scm` + machine-readable artefacts while the tree has migrated to A2ML. + +== The July-1 plan (three weeks, targeted) + +Theme: *make the flagship claims true, make the first hour flawless, +connect the two repos.* Every item below has both ends already built; +the work is wiring and honesty, not research. Effort: S = half-day, +M = 1–2 days. + +=== Week 1 (Jun 12–18) — make the consonance loop real + +. *Wire drift-on-write (M, highest leverage).* After octad commit, compute + embedding-free drift components per entity — temporal consistency + (timestamps already stored), schema drift (modality presence), + provenance drift (`verify_chain` exists), tensor staleness — and call + `DriftDetector::record`. `/drift/status` becomes live immediately. +. *Honest per-entity drift (S).* `/drift/entity/{id}` computes on demand + from that entity's modalities; delete the global-zeros shim. +. *One real regenerator (M).* In verisim-api (which owns concrete + stores): regenerate vector from document via deterministic + hash-embedding, tensor from vector reshape; wire through + `RegenerationEngine::with_regenerator`; write back via the octad store. +. *Connect `POST /normalizer/normalize/{id}` (M).* Measured drift -> + regenerate -> write -> re-measure -> respond with before/after scores. + This closes detect->repair end-to-end. +. *Scripted demo I: `just demo-consonance` (M).* Create entity -> mutate + document only -> drift detected (real numbers) -> normalize -> drift + falls -> provenance chain verifies. Transcript captured into README. + This is the artefact to show on July 1. + +=== Week 2 (Jun 19–25) — truth sweep + cheap wins + +[start=6] +. *Swap default vector store to the existing HNSW (S).* Same trait; + turns an orphaned 660-LOC asset into a true headline. +. *Plumb real search scores (S).* BM25 + cosine through + `SearchResultResponse`. +. *NIF: fail loudly (S).* Every stub returns `{:error, + :not_implemented}`; `auto` must not prefer a fake transport. (Full NIF + wiring is a non-goal for July.) +. *Honest `at_time` (S/M).* Reconstruct from the stored `OctadSnapshot`, + or return 501 until done. +. *Transactions endpoints (S).* Bridge to the octad store's real + `TransactionManager` or return 501; no fake 200s. +. *Rename the ZKP surface (S).* "Integrity commitments + Merkle + membership + R1CS constraint checking; ZK-SNARK planned via sanctify." + Fix the unconditional-true verify arm; KNOWN-ISSUES entry for the + deterministic nonce. +. *Docs truth pass (M).* README/CLAUDE.md claims aligned with code + (spatial = brute-force for now; HNSW true after item 6; drop unused + `burn`; planner framed as advisory); KNOWN-ISSUES normalizer entry + re-scoped to "derivation real, write-back landed "; delete + duplicate CHANGELOG.md; refresh `.machine_readable/` and CLAUDE.md + (`.scm` -> A2ML). +. *Auth reachability decision (S).* Either add a key-registration path or + state plainly that authn ships disabled in 0.x. + +=== Week 3 (Jun 26 – Jul 1) — connect VCL-total + release + +[start=15] +. *Minimal real admissibility gate (M, cross-repo).* vcl-ut's + `vcltotal-parse` already builds standalone (verified: 19/19 tests) and + has a wire format + decider. Ship a tiny `vclt-gate` binary there + (stdin: statement, stdout: verdict JSON with level-tagged reasons, + static levels L1–L4). VeriSimDB's VQL execute path invokes it when + present (env/feature-gated). The README triad becomes executable. +. *Scripted demo II: admissibility (S).* Well-formed DECLARE/ASSERT + admitted; malformed/unsafe statement rejected with reasons. Transcript + into README. +. *Release hygiene (M).* CI green on both repos; CHANGELOG entries; tag + `v0.2.0`; demo transcripts linked from READMEs; KNOWN-ISSUES refreshed + as of 2026-07-01. + +=== Explicit non-goals for July 1 + +Real ZK-SNARKs; live federation certification against external DBs; +R-tree spatial; full NIF data path; multi-node KRaft; physical-plan +executor; LSP/DAP completion; fleet dispatch. Each stays on the roadmap, +honestly labelled. + +== Definition of "respectable" (acceptance criteria) + +. Fresh clone builds and tests green with zero sibling checkouts — + both repos. +. One-command demo shows detect -> repair with real, changing numbers. +. The admissibility gate connects VCL-total to the engine in code, not + prose, for at least the static levels. +. Every README claim is either true today or explicitly labelled + *planned*. +. Tagged release with an honest CHANGELOG and a refreshed KNOWN-ISSUES. + +== Strengths worth showcasing (unchanged by this review) + +The octad atomicity machinery, the WAL, provenance verification, the +planner's EXPLAIN surface, three live protocol surfaces from one process, +the polished REPL, 14-adapter federation design, the Coq models, and the +repo's hygiene (REUSE, fuzzing, benches, zero TODO debt). The honest +KNOWN-ISSUES culture is itself a differentiator — this plan extends it +rather than replacing it. From e6ba85e9094e9cda91c2adfc95056d38b1257faa Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 11 Jun 2026 21:59:08 +0000 Subject: [PATCH 2/3] feat(drift): wire drift measurement into the write path + honest per-entity endpoint - verisim-octad: expose graph_store()/temporal_store() accessors (mirrors the existing provenance/spatial accessors) - verisim-api: new drift_compute module measuring per-entity cross-modal drift from live modality data: graph-document reflection, temporal consistency over version history, tensor shape/data coherence, schema baseline comparison, provenance chain integrity + staleness, spatial coordinate validity. Semantic-vector drift is honestly marked unmeasurable until a type-embedding registry exists. - create/update handlers record measured drift into the shared DriftDetector (best-effort; never fails a write). Updates use the pre-update modality set as the schema baseline so silent modality loss registers as schema drift. - /drift/entity/{id} now measures THIS entity on demand with a per-component breakdown, replacing the global-zeros shim. - Non-computable components are excluded from detector recording, so /drift/status aggregates reflect real measurements only. - Tests: 4 unit + 3 integration (drift recorded on create; per-entity component breakdown; graph-document drift falls to zero after a consonant update). Workspace: 617 passed / 0 failed; clippy clean under -D warnings. Closes Week-1 items 1-2 of docs/reviews/2026-06-11-deep-review-and-july-1-plan.adoc https://claude.ai/code/session_01E8BpV19yxhf67UrCrvkfTr --- rust-core/verisim-api/src/drift_compute.rs | 560 +++++++++++++++++++++ rust-core/verisim-api/src/lib.rs | 266 +++++++++- rust-core/verisim-octad/src/store.rs | 10 + 3 files changed, 815 insertions(+), 21 deletions(-) create mode 100644 rust-core/verisim-api/src/drift_compute.rs diff --git a/rust-core/verisim-api/src/drift_compute.rs b/rust-core/verisim-api/src/drift_compute.rs new file mode 100644 index 0000000..2c51b6c --- /dev/null +++ b/rust-core/verisim-api/src/drift_compute.rs @@ -0,0 +1,560 @@ +// SPDX-License-Identifier: MPL-2.0 +//! Per-entity drift computation. +//! +//! Bridges the modality stores to [`verisim_drift::DriftCalculator`]: reads an +//! entity's actual cross-modal state and produces measured drift scores, which +//! callers record into the shared [`verisim_drift::DriftDetector`]. +//! +//! Components that cannot be measured for an entity (missing modality, or — +//! for semantic-vector drift — no type-embedding registry yet) are reported +//! with `computable: false` and are NOT recorded into the detector, so the +//! moving averages only ever reflect real measurements. + +use chrono::{DateTime, Utc}; +use serde::{Deserialize, Serialize}; +use std::collections::hash_map::DefaultHasher; +use std::hash::{Hash, Hasher}; +use tracing::warn; +use verisim_drift::{DriftCalculator, DriftDetector, DriftType}; +use verisim_graph::{GraphObject, GraphStore}; +use verisim_octad::Octad; +use verisim_provenance::ProvenanceStore; +use verisim_temporal::TemporalStore; + +use crate::ConcreteOctadStore; + +/// Maximum number of versions inspected for temporal-consistency drift. +const TEMPORAL_HISTORY_LIMIT: usize = 256; + +/// Provenance gap beyond which staleness contributes to drift (30 days). +const PROVENANCE_MAX_GAP_SECONDS: u64 = 30 * 24 * 60 * 60; + +/// One measured (or explicitly unmeasurable) drift component for an entity. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct DriftComponent { + pub drift_type: String, + pub score: f64, + /// False when the component could not be measured for this entity + /// (missing modality or missing reference data). Non-computable + /// components are excluded from detector recording. + pub computable: bool, + pub detail: String, +} + +/// Full per-entity drift report computed from the entity's live modality data. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct EntityDriftReport { + pub entity_id: String, + pub overall_score: f64, + pub primary_drift_type: String, + pub components: Vec, + pub computed_at: DateTime, +} + +struct ComponentScores { + semantic_vector: (f64, bool, String), + graph_document: (f64, bool, String), + temporal: (f64, bool, String), + tensor: (f64, bool, String), + schema: (f64, bool, String), + provenance: (f64, bool, String), + spatial: (f64, bool, String), +} + +/// Names of the modalities currently populated on an octad, in the +/// vocabulary used by schema-drift comparison. +pub fn present_modalities(octad: &Octad) -> Vec<&'static str> { + let mut present = Vec::with_capacity(8); + if octad.graph_node.is_some() { + present.push("graph"); + } + if octad.embedding.is_some() { + present.push("vector"); + } + if octad.tensor.is_some() { + present.push("tensor"); + } + if octad.semantic.is_some() { + present.push("semantic"); + } + if octad.document.is_some() { + present.push("document"); + } + if octad.version_count > 0 { + present.push("temporal"); + } + if octad.provenance_chain_length > 0 { + present.push("provenance"); + } + if octad.spatial_data.is_some() { + present.push("spatial"); + } + present +} + +/// Compute measured drift for one entity from its live modality data. +/// +/// `required_modalities` is the schema baseline to compare against: pass the +/// modalities present *before* an update to detect modality loss, or `None` +/// to baseline against the entity's current shape (no schema drift). +pub async fn compute_entity_drift( + store: &ConcreteOctadStore, + octad: &Octad, + required_modalities: Option<&[&str]>, +) -> EntityDriftReport { + let calculator = DriftCalculator::default(); + let entity_id = octad.id.0.clone(); + + let scores = ComponentScores { + semantic_vector: semantic_vector_component(), + graph_document: graph_document_component(store, octad, &calculator).await, + temporal: temporal_component(store, &entity_id, &calculator).await, + tensor: tensor_component(octad, &calculator), + schema: schema_component(octad, required_modalities, &calculator), + provenance: provenance_component(store, &entity_id, &calculator).await, + spatial: spatial_component(octad, &calculator), + }; + + let overall_score = calculator.quality_drift_octad( + scores.semantic_vector.0, + scores.graph_document.0, + scores.temporal.0, + scores.tensor.0, + scores.schema.0, + scores.provenance.0, + scores.spatial.0, + ); + let primary = calculator.primary_drift_type_octad( + scores.semantic_vector.0, + scores.graph_document.0, + scores.temporal.0, + scores.tensor.0, + scores.schema.0, + scores.provenance.0, + scores.spatial.0, + ); + + let component = + |drift_type: DriftType, (score, computable, detail): (f64, bool, String)| DriftComponent { + drift_type: drift_type.to_string(), + score, + computable, + detail, + }; + + EntityDriftReport { + entity_id, + overall_score, + primary_drift_type: primary.to_string(), + components: vec![ + component(DriftType::SemanticVectorDrift, scores.semantic_vector), + component(DriftType::GraphDocumentDrift, scores.graph_document), + component(DriftType::TemporalConsistencyDrift, scores.temporal), + component(DriftType::TensorDrift, scores.tensor), + component(DriftType::SchemaDrift, scores.schema), + component(DriftType::ProvenanceDrift, scores.provenance), + component(DriftType::SpatialDrift, scores.spatial), + ], + computed_at: Utc::now(), + } +} + +/// Record a computed report into the shared detector (best-effort). +/// +/// Only computable components are recorded, plus the overall quality score, +/// so `/drift/status` aggregates reflect real measurements only. Errors are +/// logged, never propagated: drift bookkeeping must not fail a write. +pub async fn record_entity_drift(detector: &DriftDetector, report: &EntityDriftReport) { + let drift_type_of = |name: &str| -> Option { + match name { + "semantic_vector_drift" => Some(DriftType::SemanticVectorDrift), + "graph_document_drift" => Some(DriftType::GraphDocumentDrift), + "temporal_consistency_drift" => Some(DriftType::TemporalConsistencyDrift), + "tensor_drift" => Some(DriftType::TensorDrift), + "schema_drift" => Some(DriftType::SchemaDrift), + "provenance_drift" => Some(DriftType::ProvenanceDrift), + "spatial_drift" => Some(DriftType::SpatialDrift), + _ => None, + } + }; + + for c in report.components.iter().filter(|c| c.computable) { + if let Some(dt) = drift_type_of(&c.drift_type) { + if let Err(e) = detector + .record(dt, c.score, vec![report.entity_id.clone()]) + .await + { + warn!(entity = %report.entity_id, drift_type = %c.drift_type, error = %e, + "failed to record drift measurement"); + } + } + } + + if let Err(e) = detector + .record( + DriftType::QualityDrift, + report.overall_score, + vec![report.entity_id.clone()], + ) + .await + { + warn!(entity = %report.entity_id, error = %e, "failed to record quality drift"); + } +} + +fn semantic_vector_component() -> (f64, bool, String) { + ( + 0.0, + false, + "not measured: no type-embedding registry configured".to_string(), + ) +} + +async fn graph_document_component( + store: &ConcreteOctadStore, + octad: &Octad, + calculator: &DriftCalculator, +) -> (f64, bool, String) { + let (Some(node), Some(document)) = (&octad.graph_node, &octad.document) else { + return ( + 0.0, + false, + "not measured: requires both graph and document modalities".to_string(), + ); + }; + + let edges = match store.graph_store().outgoing(node).await { + Ok(edges) => edges, + Err(e) => return (0.0, false, format!("not measured: graph read failed: {e}")), + }; + if edges.is_empty() { + return ( + 0.0, + false, + "not measured: entity has no outgoing graph relationships".to_string(), + ); + } + + let relationships: Vec<(String, String)> = edges + .iter() + .map(|e| { + let target = match &e.object { + GraphObject::Node(n) => n.local_name.clone(), + GraphObject::Literal { value, .. } => value.clone(), + }; + (e.predicate.local_name.clone(), target) + }) + .collect(); + + // Without entity extraction, the measurable direction is graph -> document: + // a relationship target the document never mentions is unreflected structure. + let text = format!("{} {}", document.title, document.body).to_lowercase(); + let document_entities: Vec = relationships + .iter() + .map(|(_, target)| target.clone()) + .filter(|target| text.contains(&target.to_lowercase())) + .collect(); + + let score = if document_entities.is_empty() { + // Limit case the calculator cannot express (it returns 0.0 for an + // empty entity list): every relationship target is unreflected in + // the document, which by its own combination formula + // ((1 - coverage) + extra_graph_ratio) / 2 evaluates to 0.5. + 0.5 + } else { + calculator.graph_document_drift(&text, &document_entities, &relationships) + }; + let detail = format!( + "{}/{} graph relationship targets mentioned in document", + document_entities.len(), + relationships.len() + ); + (score, true, detail) +} + +async fn temporal_component( + store: &ConcreteOctadStore, + entity_id: &str, + calculator: &DriftCalculator, +) -> (f64, bool, String) { + let versions = match store + .temporal_store() + .history(entity_id, TEMPORAL_HISTORY_LIMIT) + .await + { + Ok(v) => v, + Err(e) => { + return ( + 0.0, + false, + format!("not measured: temporal read failed: {e}"), + ) + } + }; + if versions.is_empty() { + return (0.0, false, "not measured: no version history".to_string()); + } + + let timestamps: Vec = versions.iter().map(|v| v.timestamp.timestamp()).collect(); + let hashes: Vec = versions + .iter() + .map(|v| { + let serialized = serde_json::to_string(&v.data).unwrap_or_default(); + let mut hasher = DefaultHasher::new(); + serialized.hash(&mut hasher); + hasher.finish() + }) + .collect(); + + let score = calculator.temporal_consistency_drift(×tamps, &hashes); + ( + score, + true, + format!("{} versions inspected", versions.len()), + ) +} + +fn tensor_component(octad: &Octad, calculator: &DriftCalculator) -> (f64, bool, String) { + let Some(tensor) = &octad.tensor else { + return (0.0, false, "not measured: no tensor modality".to_string()); + }; + + let declared_len: usize = tensor.shape.iter().product(); + // The declared shape is the expectation; a data buffer that does not fill + // it is the measurable inconsistency. + let actual_shape: Vec = if declared_len == tensor.data.len() { + tensor.shape.clone() + } else { + vec![tensor.data.len()] + }; + + let score = calculator.tensor_drift(&tensor.data, &tensor.shape, &actual_shape, None); + let detail = format!( + "shape {:?} declares {} elements, data has {}", + tensor.shape, + declared_len, + tensor.data.len() + ); + (score, true, detail) +} + +fn schema_component( + octad: &Octad, + required_modalities: Option<&[&str]>, + calculator: &DriftCalculator, +) -> (f64, bool, String) { + let present = present_modalities(octad); + let required: Vec<&str> = match required_modalities { + Some(req) => req.to_vec(), + None => present.clone(), + }; + + let score = calculator.schema_drift(&required, &present, 0, 0); + let missing: Vec<&&str> = required.iter().filter(|m| !present.contains(*m)).collect(); + let detail = if missing.is_empty() { + format!("all {} required modalities present", required.len()) + } else { + format!("missing required modalities: {missing:?}") + }; + (score, true, detail) +} + +async fn provenance_component( + store: &ConcreteOctadStore, + entity_id: &str, + calculator: &DriftCalculator, +) -> (f64, bool, String) { + let provenance = store.provenance_store(); + + let chain = match provenance.get_chain(entity_id).await { + Ok(chain) => chain, + Err(verisim_provenance::ProvenanceError::NotFound(_)) => { + return ( + 0.0, + false, + "not measured: no provenance events recorded".to_string(), + ) + } + Err(e) => { + return ( + 0.0, + false, + format!("not measured: provenance read failed: {e}"), + ) + } + }; + let chain_valid = provenance.verify_chain(entity_id).await.unwrap_or(false); + + let seconds_since_last = chain + .latest() + .map(|record| (Utc::now() - record.timestamp).num_seconds().max(0) as u64); + + let score = calculator.provenance_drift( + chain_valid, + chain.len(), + seconds_since_last, + PROVENANCE_MAX_GAP_SECONDS, + ); + let detail = format!( + "chain length {}, integrity {}", + chain.len(), + if chain_valid { "verified" } else { "BROKEN" } + ); + (score, true, detail) +} + +fn spatial_component(octad: &Octad, calculator: &DriftCalculator) -> (f64, bool, String) { + let Some(spatial) = &octad.spatial_data else { + return (0.0, false, "not measured: no spatial modality".to_string()); + }; + + let lat = spatial.coordinates.latitude; + let lon = spatial.coordinates.longitude; + let coordinates_valid = lat.is_finite() + && lon.is_finite() + && (-90.0..=90.0).contains(&lat) + && (-180.0..=180.0).contains(&lon); + + // Location mentions are only detectable through the spatial properties the + // entity itself declares (address, region, ...): check whether any of those + // strings appear in the document text. + let doc_text = octad + .document + .as_ref() + .map(|d| format!("{} {}", d.title, d.body).to_lowercase()); + let mentioned = doc_text.as_ref().map(|text| { + spatial + .properties + .values() + .any(|v| !v.is_empty() && text.contains(&v.to_lowercase())) + }); + let has_location_mentions = mentioned.unwrap_or(false); + let coordinate_matches_mentions = has_location_mentions; + + let score = calculator.spatial_drift( + true, + has_location_mentions, + coordinate_matches_mentions, + coordinates_valid, + ); + let detail = format!( + "coordinates ({lat:.4}, {lon:.4}) {}", + if coordinates_valid { + "valid" + } else { + "INVALID" + } + ); + (score, true, detail) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn octad_with(document: bool, tensor_mismatch: Option) -> Octad { + use std::collections::HashMap; + use verisim_document::Document; + use verisim_octad::{ModalityStatus, OctadId, OctadStatus}; + use verisim_tensor::{DType, Tensor}; + + Octad { + id: OctadId::new("test-entity"), + status: OctadStatus { + id: OctadId::new("test-entity"), + created_at: Utc::now(), + modified_at: Utc::now(), + version: 1, + modality_status: ModalityStatus::default(), + }, + graph_node: None, + embedding: None, + tensor: tensor_mismatch.map(|mismatch| Tensor { + id: "test-entity".to_string(), + shape: vec![2, 2], + dtype: DType::Float64, + data: if mismatch { + vec![1.0, 2.0, 3.0] + } else { + vec![1.0, 2.0, 3.0, 4.0] + }, + metadata: HashMap::new(), + }), + semantic: None, + document: if document { + Some(Document::new("test-entity", "Title", "Body")) + } else { + None + }, + version_count: 1, + provenance_chain_length: 1, + spatial_data: None, + } + } + + #[test] + fn present_modalities_reflects_populated_fields() { + let octad = octad_with(true, Some(false)); + let present = present_modalities(&octad); + assert!(present.contains(&"document")); + assert!(present.contains(&"tensor")); + assert!(present.contains(&"temporal")); + assert!(present.contains(&"provenance")); + assert!(!present.contains(&"vector")); + assert!(!present.contains(&"graph")); + } + + #[test] + fn schema_drift_detects_dropped_modality() { + let calculator = DriftCalculator::default(); + let octad = octad_with(false, None); + + // Baseline that previously had a document: dropping it is drift. + let (score, computable, detail) = schema_component( + &octad, + Some(&["document", "temporal", "provenance"]), + &calculator, + ); + assert!(computable); + assert!( + score > 0.0, + "dropping a modality must register schema drift" + ); + assert!(detail.contains("document")); + + // Self-baseline: no drift. + let (score, _, _) = schema_component(&octad, None, &calculator); + assert_eq!(score, 0.0); + } + + #[test] + fn tensor_drift_detects_shape_data_mismatch() { + let calculator = DriftCalculator::default(); + + let consistent = octad_with(false, Some(false)); + let (score, computable, _) = tensor_component(&consistent, &calculator); + assert!(computable); + assert_eq!(score, 0.0); + + let mismatched = octad_with(false, Some(true)); + let (score, computable, _) = tensor_component(&mismatched, &calculator); + assert!(computable); + assert!( + score > 0.0, + "shape/data mismatch must register tensor drift" + ); + } + + #[test] + fn absent_modalities_are_not_computable() { + let calculator = DriftCalculator::default(); + let octad = octad_with(false, None); + + let (_, computable, _) = tensor_component(&octad, &calculator); + assert!(!computable); + let (_, computable, _) = spatial_component(&octad, &calculator); + assert!(!computable); + let (_, computable, _) = semantic_vector_component(); + assert!(!computable); + } +} diff --git a/rust-core/verisim-api/src/lib.rs b/rust-core/verisim-api/src/lib.rs index 3af49c6..10618f3 100644 --- a/rust-core/verisim-api/src/lib.rs +++ b/rust-core/verisim-api/src/lib.rs @@ -6,6 +6,7 @@ #![forbid(unsafe_code)] pub mod auth; +pub mod drift_compute; pub mod federation; pub mod graphql; pub mod grpc; @@ -988,6 +989,11 @@ async fn create_octad_handler( .await .map_err(|e| ApiError::Internal(e.to_string()))?; + // Measure cross-modal drift for the new entity (best-effort bookkeeping; + // never fails the write). + let report = drift_compute::compute_entity_drift(&state.octad_store, &octad, None).await; + drift_compute::record_entity_drift(&state.drift_detector, &report).await; + Ok((StatusCode::CREATED, Json(OctadResponse::from(&octad)))) } @@ -1021,6 +1027,10 @@ async fn update_octad_handler( let octad_id = OctadId::new(&id); let input = request.to_octad_input(); + // The pre-update modality set is the schema baseline: an update that + // silently drops a previously-populated modality is schema drift. + let previous = state.octad_store.get(&octad_id).await.ok().flatten(); + let octad = state .octad_store .update(&octad_id, input) @@ -1032,6 +1042,11 @@ async fn update_octad_handler( _ => ApiError::Internal(e.to_string()), })?; + let baseline = previous.as_ref().map(drift_compute::present_modalities); + let report = + drift_compute::compute_entity_drift(&state.octad_store, &octad, baseline.as_deref()).await; + drift_compute::record_entity_drift(&state.drift_detector, &report).await; + Ok(Json(OctadResponse::from(&octad))) } @@ -1182,9 +1197,13 @@ pub struct EntityDriftResponse { pub score: f64, pub drift_type: String, pub status: String, + /// Per-modality drift components measured for THIS entity. + pub components: Vec, + pub computed_at: chrono::DateTime, } -/// Entity drift handler — get drift info for a single entity +/// Entity drift handler — measure drift for a single entity on demand +/// from its own modality data. #[instrument(skip(state))] async fn entity_drift_handler( State(state): State, @@ -1193,32 +1212,20 @@ async fn entity_drift_handler( validate_octad_id(&id)?; let octad_id = OctadId::new(&id); - // Verify octad exists - let _octad = state + let octad = state .octad_store .get(&octad_id) .await .map_err(|e| ApiError::Internal(e.to_string()))? .ok_or_else(|| ApiError::NotFound(format!("Octad {} not found", id)))?; - // Get aggregate health from drift detector - let all_metrics = state - .drift_detector - .all_metrics() - .map_err(|e| ApiError::Internal(e.to_string()))?; - let (worst_type, worst_score) = all_metrics - .iter() - .max_by(|a, b| { - a.1.current_score - .partial_cmp(&b.1.current_score) - .unwrap_or(std::cmp::Ordering::Equal) - }) - .map(|(dt, m)| (dt.to_string(), m.current_score)) - .unwrap_or_else(|| ("none".to_string(), 0.0)); + let report = drift_compute::compute_entity_drift(&state.octad_store, &octad, None).await; + // An on-demand read is still a real measurement: feed the aggregates. + drift_compute::record_entity_drift(&state.drift_detector, &report).await; - let status = if worst_score >= 0.7 { + let status = if report.overall_score >= 0.7 { "critical" - } else if worst_score >= 0.3 { + } else if report.overall_score >= 0.3 { "warning" } else { "healthy" @@ -1226,9 +1233,11 @@ async fn entity_drift_handler( Ok(Json(EntityDriftResponse { entity_id: id, - score: worst_score, - drift_type: worst_type, + score: report.overall_score, + drift_type: report.primary_drift_type.clone(), status: status.to_string(), + components: report.components, + computed_at: report.computed_at, })) } @@ -2522,4 +2531,219 @@ mod tests { assert_eq!(response.status(), StatusCode::OK); } + + /// Request whose graph relationship target is not mentioned in the + /// document body — a real cross-modal inconsistency. + fn dissonant_octad_request() -> OctadRequest { + OctadRequest { + title: Some("Consonance Note".to_string()), + body: Some("A short note that does not mention its relation".to_string()), + embedding: Some(vec![0.1, 0.2, 0.3]), + types: None, + relationships: Some(vec![( + "relates_to".to_string(), + "orphan-target".to_string(), + )]), + tensor: None, + metadata: None, + provenance: Some(ProvenanceRequest { + event_type: "created".to_string(), + actor: "test-suite".to_string(), + source: None, + description: "created by drift wiring test".to_string(), + }), + spatial: None, + } + } + + async fn post_octad(app: &axum::Router, request: &OctadRequest) -> OctadResponse { + let response = app + .clone() + .oneshot( + Request::builder() + .method("POST") + .uri("/octads") + .header("content-type", "application/json") + .body(Body::from(serde_json::to_string(request).unwrap())) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::CREATED); + let body = axum::body::to_bytes(response.into_body(), 1024 * 1024) + .await + .unwrap(); + serde_json::from_slice(&body).unwrap() + } + + /// Creating an entity must record real drift measurements in the + /// aggregates — `/drift/status` may never report initialized zeros + /// after a write. + #[tokio::test] + async fn test_drift_recorded_on_create() { + let state = create_test_state().await; + let app = build_router(state); + + post_octad(&app, &dissonant_octad_request()).await; + + let response = app + .oneshot( + Request::builder() + .uri("/drift/status") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + let body = axum::body::to_bytes(response.into_body(), 1024 * 1024) + .await + .unwrap(); + let statuses: Vec = serde_json::from_slice(&body).unwrap(); + + let quality = statuses + .iter() + .find(|s| s.drift_type == "quality_drift") + .expect("quality_drift must be tracked"); + assert!( + quality.measurement_count >= 1, + "create must record a quality-drift measurement" + ); + + let graph_doc = statuses + .iter() + .find(|s| s.drift_type == "graph_document_drift") + .expect("graph_document_drift must be tracked"); + assert!(graph_doc.measurement_count >= 1); + assert!( + graph_doc.current_score > 0.0, + "unmentioned graph target must register graph-document drift" + ); + } + + /// `/drift/entity/{id}` must measure THIS entity's drift from its own + /// modality data, with a per-component breakdown. + #[tokio::test] + async fn test_entity_drift_measured_per_entity() { + let state = create_test_state().await; + let app = build_router(state); + + let created = post_octad(&app, &dissonant_octad_request()).await; + + let response = app + .oneshot( + Request::builder() + .uri(format!("/drift/entity/{}", created.id)) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + let body = axum::body::to_bytes(response.into_body(), 1024 * 1024) + .await + .unwrap(); + let report: EntityDriftResponse = serde_json::from_slice(&body).unwrap(); + + assert_eq!(report.entity_id, created.id); + assert!( + report.score > 0.0, + "dissonant entity must show non-zero overall drift" + ); + + let component = |name: &str| { + report + .components + .iter() + .find(|c| c.drift_type == name) + .unwrap_or_else(|| panic!("{name} component missing")) + }; + + let graph_doc = component("graph_document_drift"); + assert!(graph_doc.computable); + assert!(graph_doc.score > 0.0); + + let provenance = component("provenance_drift"); + assert!(provenance.computable); + assert_eq!( + provenance.score, 0.0, + "fresh verified chain must show zero provenance drift" + ); + + let semantic_vector = component("semantic_vector_drift"); + assert!( + !semantic_vector.computable, + "semantic-vector drift must be honestly marked unmeasurable" + ); + } + + /// Updating the document to mention the graph target must measurably + /// reduce graph-document drift: detection responds to repair. + #[tokio::test] + async fn test_drift_falls_after_consonant_update() { + let state = create_test_state().await; + let app = build_router(state); + + let created = post_octad(&app, &dissonant_octad_request()).await; + + let entity_drift = |app: axum::Router, id: String| async move { + let response = app + .oneshot( + Request::builder() + .uri(format!("/drift/entity/{id}")) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + let body = axum::body::to_bytes(response.into_body(), 1024 * 1024) + .await + .unwrap(); + let report: EntityDriftResponse = serde_json::from_slice(&body).unwrap(); + report + .components + .iter() + .find(|c| c.drift_type == "graph_document_drift") + .expect("graph_document_drift component missing") + .score + }; + + let before = entity_drift(app.clone(), created.id.clone()).await; + assert!(before > 0.0); + + // Repair the dissonance: the document now mentions the target. + let update = OctadRequest { + title: Some("Consonance Note".to_string()), + body: Some("This note now mentions orphan-target explicitly".to_string()), + embedding: None, + types: None, + relationships: None, + tensor: None, + metadata: None, + provenance: None, + spatial: None, + }; + let response = app + .clone() + .oneshot( + Request::builder() + .method("PUT") + .uri(format!("/octads/{}", created.id)) + .header("content-type", "application/json") + .body(Body::from(serde_json::to_string(&update).unwrap())) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + + let after = entity_drift(app, created.id.clone()).await; + assert!( + after < before, + "graph-document drift must fall after the document mentions the target \ + (before={before}, after={after})" + ); + assert_eq!(after, 0.0); + } } diff --git a/rust-core/verisim-octad/src/store.rs b/rust-core/verisim-octad/src/store.rs index 0b7d547..e2c9c88 100644 --- a/rust-core/verisim-octad/src/store.rs +++ b/rust-core/verisim-octad/src/store.rs @@ -381,6 +381,16 @@ where &self.spatial } + /// Access the graph store for direct queries. + pub fn graph_store(&self) -> &Arc { + &self.graph + } + + /// Access the temporal store for direct queries. + pub fn temporal_store(&self) -> &Arc { + &self.temporal + } + /// Process graph input for a octad async fn process_graph( &self, From af4734576fff93d26cef994851940d3ea7baa139 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 11 Jun 2026 22:12:24 +0000 Subject: [PATCH 3/3] feat(normalizer): real store-backed regeneration + working normalize endpoint - verisim-api regenerator module: StoreRegenerator implements the normalizer's ModalityRegenerator against the live octad store. Supported repairs (all real write-backs through OctadStore::update, so WAL-logged, versioned, and recorded on the provenance chain as 'normalized' events): Graph -> Document (rebuild relations section from actual edges) Document -> Vector (deterministic feature-hashing embedding) Document/Vector -> Tensor (1-D feature tensor) Unsupported pairs return honest errors instead of fake success. measure_drift re-reads persisted state and scores via drift_compute. - AppState gains a RegenerationEngine wired to StoreRegenerator, with a content-modality authority order (provenance/temporal excluded as regeneration sources: audit records, not content). - POST /normalizer/trigger/{id} replaces the 202 no-op: measures drift, picks the worst repairable component, runs the engine, re-measures from persisted state, feeds the aggregates, and returns action/before/after + the full post-repair component report. - Tests: hash-embedding determinism, relations-section idempotency, and an end-to-end integration test (drift 0.5 -> repaired -> 0.0, provenance chain extended, second call no_action_needed). Workspace: 620 passed / 0 failed; clippy clean under -D warnings. Closes Week-1 items 3-4 of docs/reviews/2026-06-11-deep-review-and-july-1-plan.adoc https://claude.ai/code/session_01E8BpV19yxhf67UrCrvkfTr --- Cargo.lock | 1 + rust-core/verisim-api/Cargo.toml | 1 + rust-core/verisim-api/src/lib.rs | 245 ++++++++++++++- rust-core/verisim-api/src/regenerator.rs | 363 +++++++++++++++++++++++ 4 files changed, 602 insertions(+), 8 deletions(-) create mode 100644 rust-core/verisim-api/src/regenerator.rs diff --git a/Cargo.lock b/Cargo.lock index dc33a34..a6d3985 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -8491,6 +8491,7 @@ version = "0.1.0" dependencies = [ "async-graphql", "async-graphql-axum", + "async-trait", "axum", "axum-server", "chrono", diff --git a/rust-core/verisim-api/Cargo.toml b/rust-core/verisim-api/Cargo.toml index 0395b9f..094dd13 100644 --- a/rust-core/verisim-api/Cargo.toml +++ b/rust-core/verisim-api/Cargo.toml @@ -25,6 +25,7 @@ verisim-planner = { path = "../verisim-planner" } axum.workspace = true tokio.workspace = true +async-trait.workspace = true tower.workspace = true hyper.workspace = true serde.workspace = true diff --git a/rust-core/verisim-api/src/lib.rs b/rust-core/verisim-api/src/lib.rs index 10618f3..b3f339b 100644 --- a/rust-core/verisim-api/src/lib.rs +++ b/rust-core/verisim-api/src/lib.rs @@ -11,6 +11,7 @@ pub mod federation; pub mod graphql; pub mod grpc; pub mod rbac; +pub mod regenerator; pub mod transaction; pub mod vql; @@ -36,6 +37,9 @@ use verisim_drift::{DriftDetector, DriftMetrics, DriftThresholds, DriftType}; use verisim_graph::RedbGraphStore; #[cfg(not(feature = "persistent"))] use verisim_graph::SimpleGraphStore; +use verisim_normalizer::regeneration::{ + Modality as RegenModality, RegenerationConfig, RegenerationEngine, RegenerationResult, +}; use verisim_normalizer::{create_default_normalizer, Normalizer, NormalizerStatus}; use verisim_octad::{ BoundingBox, Coordinates, InMemoryOctadStore, OctadConfig, OctadDocumentInput, OctadGraphInput, @@ -501,6 +505,7 @@ pub struct AppState { pub octad_store: Arc, pub drift_detector: Arc, pub normalizer: Arc, + pub regeneration_engine: Arc, pub planner: Arc>, pub plan_cache: Arc, pub slow_query_log: Arc, @@ -666,6 +671,27 @@ impl AppState { let drift_detector = Arc::new(DriftDetector::new(DriftThresholds::default())); let normalizer = Arc::new(create_default_normalizer(drift_detector.clone()).await); + // Authority order restricted to content modalities: provenance and + // temporal are audit/history records, not content a modality can be + // regenerated FROM (under the default order, the ever-present + // provenance chain would win source selection for every repair). + let regeneration_engine = Arc::new(RegenerationEngine::with_regenerator( + RegenerationConfig { + authority_order: vec![ + RegenModality::Document, + RegenModality::Semantic, + RegenModality::Graph, + RegenModality::Vector, + RegenModality::Tensor, + RegenModality::Spatial, + ], + ..Default::default() + }, + Arc::new(regenerator::StoreRegenerator::new( + octad_store.clone(), + config.vector_dimension, + )), + )); let planner = Arc::new(Mutex::new(Planner::new(PlannerConfig::default()))); let plan_cache = Arc::new(PlanCache::new(CacheConfig::default())); @@ -688,6 +714,7 @@ impl AppState { octad_store, drift_detector, normalizer, + regeneration_engine, planner, plan_cache, slow_query_log, @@ -1250,28 +1277,141 @@ async fn normalizer_status_handler( Ok(Json(status)) } -/// Trigger normalization handler +/// Outcome of a normalization request: measured drift before/after plus +/// what the regeneration engine actually did. +#[derive(Debug, Serialize, Deserialize)] +pub struct NormalizeResponse { + pub entity_id: String, + /// "repaired" | "no_action_needed" | "pending_resolution" | "failed" + pub action: String, + /// The modality that was regenerated, when a repair was attempted. + pub modality: Option, + /// Source modality used for the repair, when known. + pub source_modality: Option, + pub reason: Option, + pub before_score: f64, + pub after_score: Option, + /// Full per-component drift report measured after the action. + pub report_after: drift_compute::EntityDriftReport, +} + +/// Modality targeted for repair, derived from the worst measured component. +fn repair_target(report: &drift_compute::EntityDriftReport) -> Option<(RegenModality, f64)> { + report + .components + .iter() + .filter(|c| c.computable && c.score > 0.0) + .filter_map(|c| { + // Only components with an implemented regeneration path map to + // a repair target; the rest are detect-only for now. + let modality = match c.drift_type.as_str() { + "graph_document_drift" => Some(RegenModality::Document), + "tensor_drift" => Some(RegenModality::Tensor), + "semantic_vector_drift" => Some(RegenModality::Vector), + _ => None, + }; + modality.map(|m| (m, c.score)) + }) + .max_by(|a, b| a.1.partial_cmp(&b.1).unwrap_or(std::cmp::Ordering::Equal)) +} + +/// Trigger normalization: measure the entity's drift, pick the worst +/// repairable component, run the regeneration engine (which writes the +/// repair back through the store), and report measured before/after scores. #[instrument(skip(state))] async fn trigger_normalization_handler( State(state): State, Path(id): Path, -) -> Result { +) -> Result<(StatusCode, Json), ApiError> { validate_octad_id(&id)?; let octad_id = OctadId::new(&id); - // Check if octad exists - let _octad = state + let octad = state .octad_store .get(&octad_id) .await .map_err(|e| ApiError::Internal(e.to_string()))? .ok_or_else(|| ApiError::NotFound(format!("Octad {} not found", id)))?; - // In a full implementation, this would trigger actual normalization - // For now, we just verify the octad exists and return accepted - info!(id = %id, "Normalization triggered for octad"); + let before = drift_compute::compute_entity_drift(&state.octad_store, &octad, None).await; - Ok(StatusCode::ACCEPTED) + let Some((target, score)) = repair_target(&before) else { + return Ok(( + StatusCode::OK, + Json(NormalizeResponse { + entity_id: id, + action: "no_action_needed".to_string(), + modality: None, + source_modality: None, + reason: Some("no repairable drift component above zero".to_string()), + before_score: before.overall_score, + after_score: None, + report_after: before, + }), + )); + }; + + info!(id = %id, modality = %target, score, "Normalization triggered for octad"); + let result = state + .regeneration_engine + .regenerate(&octad, target, score) + .await; + + // Re-measure from the persisted state and feed the aggregates, so + // /drift/status reflects the post-repair reality. + let fresh = state + .octad_store + .get(&octad_id) + .await + .map_err(|e| ApiError::Internal(e.to_string()))? + .ok_or_else(|| ApiError::NotFound(format!("Octad {} not found", id)))?; + let after = drift_compute::compute_entity_drift(&state.octad_store, &fresh, None).await; + drift_compute::record_entity_drift(&state.drift_detector, &after).await; + + let response = match result { + RegenerationResult::Repaired { event } => NormalizeResponse { + entity_id: id, + action: "repaired".to_string(), + modality: Some(event.drifted_modality.to_string()), + source_modality: event.source_modality.map(|m| m.to_string()), + reason: None, + before_score: score, + after_score: event.post_drift_score, + report_after: after, + }, + RegenerationResult::NoActionNeeded => NormalizeResponse { + entity_id: id, + action: "no_action_needed".to_string(), + modality: Some(target.to_string()), + source_modality: None, + reason: Some("drift score at or below the engine threshold".to_string()), + before_score: score, + after_score: None, + report_after: after, + }, + RegenerationResult::PendingResolution { reason, .. } => NormalizeResponse { + entity_id: id, + action: "pending_resolution".to_string(), + modality: Some(target.to_string()), + source_modality: None, + reason: Some(reason), + before_score: score, + after_score: None, + report_after: after, + }, + RegenerationResult::Failed { error } => NormalizeResponse { + entity_id: id, + action: "failed".to_string(), + modality: Some(target.to_string()), + source_modality: None, + reason: Some(error), + before_score: score, + after_score: None, + report_after: after, + }, + }; + + Ok((StatusCode::OK, Json(response))) } // --- Query Planner Handlers --- @@ -2746,4 +2886,93 @@ mod tests { ); assert_eq!(after, 0.0); } + + /// The full loop: measured drift -> normalize -> regeneration engine + /// repairs the document from the graph -> re-measured drift falls to + /// zero -> the repair is recorded in the provenance chain. + #[tokio::test] + async fn test_normalize_repairs_dissonant_entity() { + let state = create_test_state().await; + let app = build_router(state); + + let created = post_octad(&app, &dissonant_octad_request()).await; + assert_eq!(created.provenance_chain_length, 1); + + let response = app + .clone() + .oneshot( + Request::builder() + .method("POST") + .uri(format!("/normalizer/trigger/{}", created.id)) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + let body = axum::body::to_bytes(response.into_body(), 1024 * 1024) + .await + .unwrap(); + let normalized: NormalizeResponse = serde_json::from_slice(&body).unwrap(); + + assert_eq!( + normalized.action, "repaired", + "reason: {:?}", + normalized.reason + ); + assert_eq!(normalized.modality.as_deref(), Some("document")); + assert_eq!(normalized.source_modality.as_deref(), Some("graph")); + assert!(normalized.before_score > 0.0); + assert_eq!( + normalized.after_score, + Some(0.0), + "regenerated document must fully restore graph-document consonance" + ); + let graph_doc_after = normalized + .report_after + .components + .iter() + .find(|c| c.drift_type == "graph_document_drift") + .expect("graph_document_drift component missing"); + assert_eq!(graph_doc_after.score, 0.0); + + // The repair itself must be on the provenance chain. + let response = app + .clone() + .oneshot( + Request::builder() + .uri(format!("/octads/{}", created.id)) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + let body = axum::body::to_bytes(response.into_body(), 1024 * 1024) + .await + .unwrap(); + let fetched: OctadResponse = serde_json::from_slice(&body).unwrap(); + assert_eq!( + fetched.provenance_chain_length, 2, + "the normalized event must extend the provenance chain" + ); + + // A consonant entity has nothing to repair. + let response = app + .oneshot( + Request::builder() + .method("POST") + .uri(format!("/normalizer/trigger/{}", created.id)) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + let body = axum::body::to_bytes(response.into_body(), 1024 * 1024) + .await + .unwrap(); + let second: NormalizeResponse = serde_json::from_slice(&body).unwrap(); + assert_eq!(second.action, "no_action_needed"); + } } diff --git a/rust-core/verisim-api/src/regenerator.rs b/rust-core/verisim-api/src/regenerator.rs new file mode 100644 index 0000000..774f2f7 --- /dev/null +++ b/rust-core/verisim-api/src/regenerator.rs @@ -0,0 +1,363 @@ +// SPDX-License-Identifier: MPL-2.0 +//! Store-backed modality regeneration. +//! +//! [`StoreRegenerator`] implements the normalizer's [`ModalityRegenerator`] +//! trait against the live octad store: repairs are written back through +//! `OctadStore::update` (so they are WAL-logged, versioned, and recorded in +//! the provenance chain as `normalized` events), and post-repair drift is +//! re-measured from the freshly persisted state via [`crate::drift_compute`]. +//! +//! Supported regeneration pairs (v1, all real write-backs): +//! +//! * Graph -> Document: rebuild the document's relations section from the +//! entity's actual outgoing edges. +//! * Document -> Vector: deterministic feature-hashing embedding of the +//! document text. This is an honest fallback embedder (no ML model): it +//! makes vector content reproducibly derived from the document, which is +//! the consonance property the octad needs. +//! * Document -> Tensor / Vector -> Tensor: tensor derived from the hash +//! embedding (or existing embedding) as a 1-D feature tensor. +//! +//! Unsupported pairs return an error so the engine reports an honest +//! `Failed` instead of a fake success. + +use std::sync::Arc; + +use async_trait::async_trait; +use verisim_graph::{GraphObject, GraphStore}; +use verisim_normalizer::regeneration::{Modality, ModalityRegenerator}; +use verisim_normalizer::NormalizerError; +use verisim_octad::{ + Octad, OctadDocumentInput, OctadId, OctadInput, OctadProvenanceInput, OctadStore, + OctadTensorInput, OctadVectorInput, +}; + +use crate::drift_compute; +use crate::ConcreteOctadStore; + +/// Marker line that opens the regenerated relations section in a document +/// body. Used to strip a previous section before appending a fresh one, so +/// repeated normalisation stays idempotent. +const RELATIONS_HEADER: &str = "Relations:"; + +fn fail(entity_id: &OctadId, message: impl Into) -> NormalizerError { + NormalizerError::NormalizationFailed { + entity_id: entity_id.to_string(), + message: message.into(), + } +} + +pub struct StoreRegenerator { + store: Arc, + /// Dimension for regenerated embeddings; must match the store's + /// configured vector dimension or the write-back is rejected. + vector_dimension: usize, +} + +impl StoreRegenerator { + pub fn new(store: Arc, vector_dimension: usize) -> Self { + Self { + store, + vector_dimension, + } + } + + /// Deterministic feature-hashing embedding of `text` (FNV-1a per token, + /// signed hashing trick, L2-normalised). + pub fn hash_embedding(text: &str, dimension: usize) -> Vec { + let mut vector = vec![0.0f32; dimension.max(1)]; + for token in text + .to_lowercase() + .split(|c: char| !c.is_alphanumeric()) + .filter(|t| !t.is_empty()) + { + let mut hash: u64 = 0xcbf2_9ce4_8422_2325; + for byte in token.as_bytes() { + hash ^= u64::from(*byte); + hash = hash.wrapping_mul(0x0000_0100_0000_01b3); + } + let index = (hash % dimension.max(1) as u64) as usize; + let sign = if (hash >> 63) == 0 { 1.0 } else { -1.0 }; + vector[index] += sign; + } + let norm: f32 = vector.iter().map(|v| v * v).sum::().sqrt(); + if norm > 0.0 { + for v in &mut vector { + *v /= norm; + } + } + vector + } + + /// Rebuild a document body whose relations section reflects the actual + /// outgoing graph edges. + fn document_with_relations(original_body: &str, relations: &[(String, String)]) -> String { + let base = original_body + .split(RELATIONS_HEADER) + .next() + .unwrap_or(original_body) + .trim_end(); + let listed: Vec = relations + .iter() + .map(|(predicate, target)| format!("{predicate} {target}")) + .collect(); + format!("{base}\n\n{RELATIONS_HEADER} {}.", listed.join("; ")) + } + + async fn outgoing_relations(&self, octad: &Octad) -> Result, String> { + let Some(node) = &octad.graph_node else { + return Err("entity has no graph node".to_string()); + }; + let edges = self + .store + .graph_store() + .outgoing(node) + .await + .map_err(|e| format!("graph read failed: {e}"))?; + Ok(edges + .iter() + .map(|e| { + let target = match &e.object { + GraphObject::Node(n) => n.local_name.clone(), + GraphObject::Literal { value, .. } => value.clone(), + }; + (e.predicate.local_name.clone(), target) + }) + .collect()) + } + + /// Write one regenerated modality back through the store with a + /// `normalized` provenance event, so the repair is versioned and audited. + async fn write_back( + &self, + id: &OctadId, + input: OctadInput, + description: String, + ) -> Result<(), NormalizerError> { + let input = OctadInput { + provenance: Some(OctadProvenanceInput { + event_type: "normalized".to_string(), + actor: "verisim-normalizer".to_string(), + source: None, + description, + }), + ..input + }; + self.store + .update(id, input) + .await + .map(|_| ()) + .map_err(|e| fail(id, format!("write-back failed: {e}"))) + } +} + +#[async_trait] +impl ModalityRegenerator for StoreRegenerator { + async fn regenerate_from( + &self, + octad: &Octad, + source: Modality, + target: Modality, + ) -> Result { + let id = octad.id.clone(); + match (source, target) { + (Modality::Graph, Modality::Document) => { + let relations = self + .outgoing_relations(octad) + .await + .map_err(|m| fail(&id, m))?; + if relations.is_empty() { + return Err(fail( + &id, + "entity has no outgoing graph relationships to regenerate from", + )); + } + let document = octad + .document + .as_ref() + .ok_or_else(|| fail(&id, "entity has no document to repair"))?; + + let body = Self::document_with_relations(&document.body, &relations); + let summary = format!( + "document body rebuilt with relations section ({} relationships)", + relations.len() + ); + self.write_back( + &id, + OctadInput { + document: Some(OctadDocumentInput { + title: document.title.clone(), + body, + fields: document.fields.clone(), + }), + ..Default::default() + }, + summary.clone(), + ) + .await?; + Ok(summary) + } + + (Modality::Document, Modality::Vector) => { + let document = octad + .document + .as_ref() + .ok_or_else(|| fail(&id, "entity has no document"))?; + let text = format!("{} {}", document.title, document.body); + let embedding = Self::hash_embedding(&text, self.vector_dimension); + let summary = format!( + "embedding regenerated from document via deterministic feature hashing (dim {})", + embedding.len() + ); + self.write_back( + &id, + OctadInput { + vector: Some(OctadVectorInput { + embedding, + model: Some("hash-embedding-v1".to_string()), + }), + ..Default::default() + }, + summary.clone(), + ) + .await?; + Ok(summary) + } + + (Modality::Document, Modality::Tensor) | (Modality::Vector, Modality::Tensor) => { + let data: Vec = if source == Modality::Vector { + octad + .embedding + .as_ref() + .ok_or_else(|| fail(&id, "entity has no embedding"))? + .vector + .iter() + .map(|v| f64::from(*v)) + .collect() + } else { + let document = octad + .document + .as_ref() + .ok_or_else(|| fail(&id, "entity has no document"))?; + let text = format!("{} {}", document.title, document.body); + Self::hash_embedding(&text, self.vector_dimension) + .iter() + .map(|v| f64::from(*v)) + .collect() + }; + let shape = vec![data.len()]; + let summary = format!( + "tensor regenerated from {source} as 1-D feature tensor (len {})", + data.len() + ); + self.write_back( + &id, + OctadInput { + tensor: Some(OctadTensorInput { shape, data }), + ..Default::default() + }, + summary.clone(), + ) + .await?; + Ok(summary) + } + + (source, target) => Err(fail( + &id, + format!("no regeneration path implemented for {source} -> {target}"), + )), + } + } + + async fn merge_into( + &self, + octad: &Octad, + _sources: &[(Modality, f64)], + target: Modality, + ) -> Result { + Err(fail( + &octad.id, + format!("merge regeneration into {target} is not implemented"), + )) + } + + /// Re-measure drift for `modality` from the freshly persisted entity + /// state (the octad handed in by the engine predates the write-back). + async fn measure_drift( + &self, + octad: &Octad, + modality: Modality, + ) -> Result { + let fresh = self + .store + .get(&octad.id) + .await + .map_err(|e| fail(&octad.id, format!("re-read failed: {e}")))? + .ok_or_else(|| fail(&octad.id, "entity vanished during normalization"))?; + + let report = drift_compute::compute_entity_drift(&self.store, &fresh, None).await; + Ok(drift_score_for_modality(&report, modality)) + } +} + +/// Map a normalizer modality onto the drift component that measures it. +/// +/// Graph/document consonance is one shared component; vector maps to the +/// semantic-vector component (0.0 while that component is unmeasurable); +/// modalities without a dedicated component fall back to the overall score. +pub fn drift_score_for_modality( + report: &drift_compute::EntityDriftReport, + modality: Modality, +) -> f64 { + let component_score = |name: &str| { + report + .components + .iter() + .find(|c| c.drift_type == name) + .map(|c| c.score) + }; + + match modality { + Modality::Document | Modality::Graph => component_score("graph_document_drift"), + Modality::Vector => component_score("semantic_vector_drift"), + Modality::Tensor => component_score("tensor_drift"), + Modality::Temporal => component_score("temporal_consistency_drift"), + Modality::Provenance => component_score("provenance_drift"), + Modality::Spatial => component_score("spatial_drift"), + Modality::Semantic => component_score("schema_drift"), + } + .unwrap_or(report.overall_score) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn hash_embedding_is_deterministic_and_normalised() { + let a = StoreRegenerator::hash_embedding("the quick brown fox", 8); + let b = StoreRegenerator::hash_embedding("the quick brown fox", 8); + assert_eq!(a, b); + assert_eq!(a.len(), 8); + let norm: f32 = a.iter().map(|v| v * v).sum::().sqrt(); + assert!((norm - 1.0).abs() < 1e-5); + + let c = StoreRegenerator::hash_embedding("a completely different text", 8); + assert_ne!(a, c, "different text must hash to a different embedding"); + } + + #[test] + fn document_with_relations_is_idempotent() { + let relations = vec![("relates_to".to_string(), "orphan-target".to_string())]; + let once = StoreRegenerator::document_with_relations("Original body", &relations); + assert!(once.contains("Original body")); + assert!(once.contains("orphan-target")); + + let twice = StoreRegenerator::document_with_relations(&once, &relations); + assert_eq!( + twice.matches("orphan-target").count(), + 1, + "re-normalising must not stack relations sections" + ); + } +}