vela_protocol/

state_integrity.rs

//! Structural integrity checks for accepted frontier state.
//!
//! These checks are intentionally about substrate correctness, not scientific
//! completeness. Missing evidence spans can keep a frontier out of strict proof
//! use; duplicate events, broken replay, and applied proposals without events
//! are harder failures because they mean the state history itself is suspect.

use std::collections::{BTreeMap, BTreeSet};
use std::path::Path;

use serde::{Deserialize, Serialize};

use crate::events::{self, ReplayReport};
use crate::project::Project;
use crate::repo;

pub const STATE_INTEGRITY_SCHEMA: &str = "vela.state_integrity_report.v0.1";

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct IntegrityIssue {
    pub rule_id: String,
    pub message: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub object_id: Option<String>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct StateIntegrityReport {
    pub schema: String,
    pub status: String,
    #[serde(default)]
    pub structural_errors: Vec<IntegrityIssue>,
    #[serde(default)]
    pub warnings: Vec<IntegrityIssue>,
    pub proof_freshness: String,
    pub replay: ReplayReport,
    #[serde(default)]
    pub summary: BTreeMap<String, usize>,
}

pub fn analyze_path(path: &Path) -> Result<StateIntegrityReport, String> {
    let frontier = repo::load_from_path(path)?;
    let mut report = analyze(&frontier);
    for layout_issue in crate::frontier_repo::layout_issues(path, &frontier) {
        report.structural_errors.push(IntegrityIssue {
            rule_id: layout_issue.rule_id,
            message: layout_issue.message,
            object_id: None,
        });
    }
    if !report.structural_errors.is_empty() {
        report.status = "fail".to_string();
    } else if !report.warnings.is_empty() {
        report.status = "warn".to_string();
    } else {
        report.status = "ok".to_string();
    }
    report.summary.insert(
        "structural_errors".to_string(),
        report.structural_errors.len(),
    );
    report
        .summary
        .insert("warnings".to_string(), report.warnings.len());
    Ok(report)
}

pub fn analyze(frontier: &Project) -> StateIntegrityReport {
    let replay = events::replay_report(frontier);
    let mut structural_errors = Vec::new();
    let mut warnings = Vec::new();
    let event_ids = frontier
        .events
        .iter()
        .map(|event| event.id.as_str())
        .collect::<BTreeSet<_>>();

    for id in &replay.event_log.duplicate_ids {
        structural_errors.push(issue(
            "duplicate_event_id",
            format!("Duplicate canonical event id {id}."),
            Some(id.clone()),
        ));
    }
    for id in &replay.event_log.orphan_targets {
        structural_errors.push(issue(
            "orphan_event_target",
            format!("Canonical event targets missing finding {id}."),
            Some(id.clone()),
        ));
    }
    if !replay.ok {
        for conflict in &replay.conflicts {
            if conflict.starts_with("duplicate event id:")
                || conflict.starts_with("orphan event target:")
            {
                continue;
            }
            structural_errors.push(issue("replay_conflict", conflict.clone(), None));
        }
    }

    for event in &frontier.events {
        if is_accepted_state_event(&event.kind)
            && event
                .payload
                .get("proposal_id")
                .and_then(|value| value.as_str())
                .is_none_or(|value| value.trim().is_empty())
        {
            structural_errors.push(issue(
                "accepted_event_missing_proposal_id",
                format!("Accepted event {} has no payload.proposal_id.", event.id),
                Some(event.id.clone()),
            ));
        }
    }

    for proposal in &frontier.proposals {
        if matches!(proposal.status.as_str(), "accepted" | "applied") {
            let Some(event_id) = proposal.applied_event_id.as_deref() else {
                structural_errors.push(issue(
                    "applied_proposal_missing_event",
                    format!(
                        "Proposal {} is {} without an applied event id.",
                        proposal.id, proposal.status
                    ),
                    Some(proposal.id.clone()),
                ));
                continue;
            };
            if !event_ids.contains(event_id) {
                structural_errors.push(issue(
                    "applied_proposal_event_missing",
                    format!(
                        "Proposal {} points to missing event {event_id}.",
                        proposal.id
                    ),
                    Some(proposal.id.clone()),
                ));
            }
        }

        if proposal.kind == "artifact.assert"
            && matches!(proposal.status.as_str(), "accepted" | "applied")
        {
            let artifact = proposal.payload.get("artifact");
            let locator_missing = artifact
                .and_then(|value| value.get("locator"))
                .and_then(|value| value.as_str())
                .is_none_or(|value| value.trim().is_empty());
            let hash_missing = artifact
                .and_then(|value| value.get("content_hash"))
                .and_then(|value| value.as_str())
                .is_none_or(|value| !value.starts_with("sha256:"));
            if locator_missing || hash_missing {
                structural_errors.push(issue(
                    "accepted_artifact_missing_locator_or_hash",
                    format!(
                        "Artifact proposal {} is accepted without locator or content hash.",
                        proposal.id
                    ),
                    Some(proposal.id.clone()),
                ));
            }
        }
    }

    let proof_freshness = proof_freshness(frontier);
    if proof_freshness == "stale" {
        structural_errors.push(issue(
            "stale_proof_packet",
            "Recorded proof packet is stale relative to accepted events.".to_string(),
            None,
        ));
    } else if proof_freshness == "unknown" {
        warnings.push(issue(
            "proof_freshness_unknown",
            "No current proof packet is recorded for this frontier.".to_string(),
            None,
        ));
    }

    let status = if structural_errors.is_empty() {
        if warnings.is_empty() { "ok" } else { "warn" }
    } else {
        "fail"
    }
    .to_string();

    let mut summary = BTreeMap::new();
    summary.insert("events".to_string(), frontier.events.len());
    summary.insert("proposals".to_string(), frontier.proposals.len());
    summary.insert("structural_errors".to_string(), structural_errors.len());
    summary.insert("warnings".to_string(), warnings.len());

    StateIntegrityReport {
        schema: STATE_INTEGRITY_SCHEMA.to_string(),
        status,
        structural_errors,
        warnings,
        proof_freshness,
        replay,
        summary,
    }
}

fn issue(rule_id: &str, message: String, object_id: Option<String>) -> IntegrityIssue {
    IntegrityIssue {
        rule_id: rule_id.to_string(),
        message,
        object_id,
    }
}

fn is_accepted_state_event(kind: &str) -> bool {
    matches!(
        kind,
        "finding.asserted"
            | "finding.reviewed"
            | "finding.noted"
            | "finding.caveated"
            | "finding.confidence_revised"
            | "finding.rejected"
            | "finding.retracted"
            | "finding.dependency_invalidated"
            | "artifact.asserted"
            | "artifact.reviewed"
            | "artifact.retracted"
            | "negative_result.asserted"
            | "negative_result.reviewed"
            | "negative_result.retracted"
            | "trajectory.created"
            | "trajectory.step_appended"
            | "trajectory.reviewed"
            | "trajectory.retracted"
    )
}

fn proof_freshness(frontier: &Project) -> String {
    let state = &frontier.proof_state.latest_packet;
    if state.status == "never_exported" {
        return "unknown".to_string();
    }
    if state.status == "stale" {
        return "stale".to_string();
    }
    let current_event_hash = events::event_log_hash(&frontier.events);
    match state.event_log_hash.as_deref() {
        Some(hash) if hash == current_event_hash => "fresh".to_string(),
        Some(_) => "stale".to_string(),
        None => "unknown".to_string(),
    }
}