use std::collections::{HashMap, HashSet};
use super::event::ExecutionEvent;
use super::{ExecutionLedgerError, Result};
pub const GENESIS: &str = "genesis";
pub fn hash_event(event: &ExecutionEvent) -> serde_json::Result<String> {
let canonical = event.hashable_json()?;
Ok(crate::core::savings_ledger::event::compute_hash(
event.prev_hash(),
&canonical,
))
}
pub fn verify_events(events: &[ExecutionEvent]) -> Result<bool> {
let mut expected_previous = GENESIS.to_string();
let mut expected_sequence = 1_u64;
let mut tasks = HashMap::<String, TaskState>::new();
let mut receipt_chains = HashMap::<String, (u64, String)>::new();
let mut canonical_receipt_ids = HashSet::<String>::new();
for (index, event) in events.iter().enumerate() {
if event.sequence_number() != expected_sequence {
return Ok(false);
}
if event.prev_hash() != expected_previous {
return Ok(false);
}
let actual_hash = hash_event(event)?;
if event
.entry_hash()
.is_some_and(|entry_hash| entry_hash != actual_hash)
{
return Ok(false);
}
if index > 0 && actual_hash.is_empty() {
return Err(ExecutionLedgerError::InvalidChain(
"event hash unexpectedly empty".to_owned(),
));
}
if !validate_lifecycle(
event,
&mut tasks,
&mut receipt_chains,
&mut canonical_receipt_ids,
) {
return Ok(false);
}
expected_previous = actual_hash;
expected_sequence = expected_sequence.checked_add(1).ok_or_else(|| {
ExecutionLedgerError::InvalidChain(
"execution ledger sequence number overflow".to_owned(),
)
})?;
}
Ok(true)
}
#[derive(Default)]
struct TaskState {
trace_id: String,
plans: HashSet<String>,
invocations: HashSet<String>,
legacy_receipts: HashSet<String>,
canonical_receipts: HashSet<String>,
canonical_outcomes: HashSet<String>,
outcomes: HashSet<String>,
decisions: HashSet<String>,
context_delivered: bool,
}
fn validate_lifecycle(
event: &ExecutionEvent,
tasks: &mut HashMap<String, TaskState>,
receipt_chains: &mut HashMap<String, (u64, String)>,
canonical_receipt_ids: &mut HashSet<String>,
) -> bool {
if event.task_id().is_empty() || event.trace_id().is_empty() || event.timestamp().is_empty() {
return false;
}
if let ExecutionEvent::TaskStarted {
task_id, trace_id, ..
} = event
{
return tasks
.insert(
task_id.clone(),
TaskState {
trace_id: trace_id.clone(),
..TaskState::default()
},
)
.is_none();
}
let Some(state) = tasks.get_mut(event.task_id()) else {
return false;
};
if state.trace_id != event.trace_id() {
return false;
}
match event {
ExecutionEvent::PlanCreated { plan_id, .. } => {
state.plans.is_empty() && !plan_id.is_empty() && state.plans.insert(plan_id.clone())
}
ExecutionEvent::ContextDelivered { .. } => {
if state.context_delivered || state.plans.is_empty() {
false
} else {
state.context_delivered = true;
true
}
}
ExecutionEvent::ModelInvoked {
plan_id,
invocation_id,
invocation_ref,
provider,
..
} => {
(state.context_delivered || provider == "savings-ledger")
&& state.plans.contains(plan_id)
&& !invocation_id.is_empty()
&& !invocation_ref.is_empty()
&& state.invocations.insert(invocation_id.clone())
}
ExecutionEvent::EngineInvoked {
plan_id,
invocation_id,
invocation_ref,
capability_id,
capability_version,
..
} => {
state.context_delivered
&& state.plans.contains(plan_id)
&& !invocation_id.is_empty()
&& !invocation_ref.is_empty()
&& !capability_id.is_empty()
&& !capability_version.is_empty()
&& state.invocations.insert(invocation_id.clone())
}
ExecutionEvent::ReceiptSigned { receipt_id, .. } => {
!state.invocations.is_empty()
&& !receipt_id.is_empty()
&& state.legacy_receipts.insert(receipt_id.clone())
}
ExecutionEvent::CanonicalReceiptRecorded {
invocation_id,
receipt_id,
receipt_ref,
receipt_digest,
receipt_chain_id,
receipt_sequence_number,
previous_receipt_id,
previous_signature_digest,
..
} => {
let predecessor_pair = (
previous_receipt_id.as_deref(),
previous_signature_digest.as_deref(),
);
let valid_predecessor = match receipt_chains.get(receipt_chain_id) {
None => *receipt_sequence_number == 1 && predecessor_pair == (None, None),
Some((sequence, head)) => {
sequence.checked_add(1) == Some(*receipt_sequence_number)
&& previous_receipt_id.as_deref() == Some(head.as_str())
&& previous_signature_digest.as_deref().is_some_and(is_sha256)
}
};
if !state.context_delivered
|| !state.invocations.contains(invocation_id)
|| !is_sha256(receipt_id)
|| !is_sha256(receipt_digest)
|| receipt_ref != &format!("id:{receipt_digest}")
|| receipt_chain_id.is_empty()
|| !valid_predecessor
|| state.canonical_receipts.contains(receipt_id)
|| canonical_receipt_ids.contains(receipt_id)
{
return false;
}
state.canonical_receipts.insert(receipt_id.clone());
canonical_receipt_ids.insert(receipt_id.clone());
receipt_chains.insert(
receipt_chain_id.clone(),
(*receipt_sequence_number, receipt_id.clone()),
);
true
}
ExecutionEvent::OutcomeRecorded {
outcome_id,
receipt_id,
..
} => {
state.canonical_receipts.contains(receipt_id)
&& !outcome_id.is_empty()
&& state.canonical_outcomes.insert(receipt_id.clone())
&& state.outcomes.insert(outcome_id.clone())
}
ExecutionEvent::DecisionRecorded { decision_id, .. } => {
!decision_id.is_empty() && state.decisions.insert(decision_id.clone())
}
ExecutionEvent::TaskStarted { .. } => false,
}
}
fn is_sha256(value: &str) -> bool {
value.strip_prefix("sha256:").is_some_and(|hex| {
hex.len() == 64
&& hex
.bytes()
.all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase())
})
}