use super::*;
use crate::atomic_runner::{
run_prepared_atomic_unit, AtomicOpPlan, PreparedAtomicError, PreparedAtomicOp,
PreparedAtomicOutcome,
};
use crate::{KhiveRuntime, Namespace, NamespaceToken, RuntimeError, RuntimeResult, VerbRegistry};
use khive_storage::{SqlStatement, StorageError};
use serde_json::json;
use uuid::Uuid;
struct PreparedFix {
owner: KhiveRuntime,
token: NamespaceToken,
guards: Vec<SqlStatement>,
plan: AtomicOpPlan,
committed: CommittedFixAction,
}
enum ActionRefusal {
Conflict(String),
Failed(RuntimeError),
}
enum Candidate {
Ready(Box<PreparedFix>),
Unsupported(String),
Conflict(String),
}
impl LintCapture {
fn unique_owner(&self, ids: &[Uuid]) -> Option<usize> {
let owner = *self.record_owners.get(ids.first()?)?.first()?;
ids.iter()
.all(|id| {
self.record_owners
.get(id)
.is_some_and(|owners| owners.as_slice() == [owner])
})
.then_some(owner)
}
fn tombstone_guard(&self, id: Uuid) -> Option<SqlStatement> {
if let Some(entity) = self
.all_entities
.get(&id)
.filter(|entity| entity.deleted_at.is_some())
{
return Some(khive_db::graph_lint::entity_witness(entity));
}
if let Some(note) = self
.all_notes
.get(&id)
.filter(|note| note.deleted_at.is_some())
{
return Some(khive_db::graph_lint::note_witness(note));
}
self.all_edges
.get(&id)
.filter(|edge| edge.deleted_at.is_some())
.and_then(|_| self.edge_guards.get(&id).cloned())
}
async fn prepare_fix(
&self,
registry: &VerbRegistry,
token: &NamespaceToken,
identity: &ActionIdentity,
violation: &LintViolation,
) -> RuntimeResult<Candidate> {
let id = violation.subject.id();
let captured_subject = match &violation.subject {
LintSubject::Entity {
namespace,
entity_id,
entity_name,
entity_kind,
} => self.entities.iter().any(|entity| {
entity.id == *entity_id
&& entity.namespace == *namespace
&& entity.name == *entity_name
&& entity.kind == *entity_kind
}),
LintSubject::Edge { namespace, edge_id } => self.edges.iter().any(|inspected| {
Uuid::from(inspected.edge.id) == *edge_id && inspected.edge.namespace == *namespace
}),
LintSubject::Note { .. } => false,
};
if !captured_subject {
return Err(RuntimeError::InvalidInput(
"lint action subject was not in the visible capture".into(),
));
}
let unsupported = || {
Candidate::Unsupported(
"the observed action has no unique same-database guarded owner".into(),
)
};
let ns = Namespace::parse(violation.subject.namespace())
.map_err(|error| RuntimeError::InvalidInput(error.to_string()))?;
if !token.visible_namespaces().contains(&ns) {
return Err(RuntimeError::InvalidInput(
"lint action is outside its captured visible scope".into(),
));
}
let write_token = token.with_namespace(ns).with_request_id(token.request_id());
let (owner, guards, verb, args) = match identity.action {
FixAction::SoftDeleteEdge => {
let Some(edge) = self
.all_edges
.get(&id)
.filter(|edge| edge.deleted_at.is_none())
else {
return Ok(Candidate::Conflict("captured edge is not live".into()));
};
let Some((owner, endpoint_guard)) = [edge.source_id, edge.target_id]
.into_iter()
.find_map(|endpoint| {
let owner = self.unique_owner(&[id, endpoint])?;
self.tombstone_guard(endpoint).map(|guard| (owner, guard))
})
else {
return Ok(unsupported());
};
(
owner,
vec![self.edge_guards[&id].clone(), endpoint_guard],
"delete",
json!({"id":id, "hard":false}),
)
}
FixAction::ProposeDuplicateMerge => {
let members = violation.member_ids.as_deref().ok_or_else(|| {
RuntimeError::Internal("duplicate lint violation lacks members".into())
})?;
if members.len() < 2 {
return Err(RuntimeError::Internal(
"duplicate lint group needs two distinct members".into(),
));
}
let mut distinct = members.to_vec();
distinct.sort();
distinct.dedup();
let first = self.all_entities.get(&id).ok_or_else(|| {
RuntimeError::Internal("duplicate representative was not captured".into())
})?;
if distinct.len() != members.len()
|| distinct.first() != Some(&id)
|| members.iter().any(|member| {
self.all_entities.get(member).is_none_or(|entity| {
entity.namespace != first.namespace
|| entity.kind != first.kind
|| entity.name != first.name
})
})
{
return Err(RuntimeError::InvalidInput(
"lint duplicate group does not match the captured identity".into(),
));
}
let Some(owner) = self.unique_owner(members) else {
return Ok(unsupported());
};
let mut guards = Vec::new();
for member in members {
let entity = self.all_entities.get(member).ok_or_else(|| {
RuntimeError::Internal("lint duplicate member was not captured".into())
})?;
if entity.deleted_at.is_some() {
return Ok(Candidate::Conflict(
"captured duplicate member is not live".into(),
));
}
guards.push(khive_db::graph_lint::entity_witness(entity));
}
let content = format!("Possible duplicate entities in namespace {}: {}. Review these records before deciding whether to merge them; no winner has been selected.", violation.subject.namespace(), members.iter().map(Uuid::to_string).collect::<Vec<_>>().join(", "));
(
owner,
guards,
"create",
json!({"kind":"question", "content":content, "properties":{"lint_rule":"duplicate-name", "member_ids":members}}),
)
}
FixAction::MarkDeprecated => {
let Some(entity) = self.all_entities.get(&id).filter(|entity| {
entity.deleted_at.is_none()
&& entity
.properties
.as_ref()
.and_then(|properties| properties.get("status"))
.and_then(serde_json::Value::as_str)
!= Some("deprecated")
}) else {
return Ok(Candidate::Conflict(
"captured entity is not an unmarked live target".into(),
));
};
let witness = self.edges.iter().find_map(|inspected| {
let edge = &inspected.edge;
if edge.target_id != id
|| edge.relation != khive_storage::EdgeRelation::Supersedes
|| !inspected.source_live
|| inspected.contract_valid != Some(true)
{
return None;
}
let source = self.all_entities.get(&edge.source_id)?;
let owner = self.unique_owner(&[id, Uuid::from(edge.id), source.id])?;
Some((owner, edge, source))
});
let Some((owner, edge, source)) = witness else {
return Ok(unsupported());
};
(
owner,
vec![
khive_db::graph_lint::entity_witness(entity),
self.edge_guards[&Uuid::from(edge.id)].clone(),
khive_db::graph_lint::entity_witness(source),
],
"update",
json!({"id":id,"properties":{"status":"deprecated"}}),
)
}
};
let runtime = &self.owners[owner];
{
let mut reader = runtime.sql().reader().await?;
for guard in &guards {
if reader.query_all(guard.clone()).await?.len() != 1 {
return Ok(Candidate::Conflict(
"an observed lint predicate changed before preparation".into(),
));
}
}
}
registry
.authorize_effective_kg_read(&write_token, verb, args.clone(), id)
.await?;
let mut plan = match identity.action {
FixAction::SoftDeleteEdge => {
crate::atomic_prepare::prepare_delete(
runtime,
&write_token,
&args,
Some(crate::atomic_prepare::AtomicDeleteKind::Edge),
)
.await?
}
FixAction::ProposeDuplicateMerge => {
crate::atomic_prepare::prepare_add_note(runtime, &write_token, &args).await?
}
FixAction::MarkDeprecated => {
crate::atomic_prepare::prepare_update_entity_plan_with_version(
runtime,
&write_token,
id,
crate::curation::EntityPatch {
properties: Some(json!({"status":"deprecated"})),
..Default::default()
},
Some(self.all_entities[&id].version),
)
.await?
}
};
if let AtomicOpPlan::AddNote(note) = &mut plan {
note.statements.extend(crate::atomic_prepare::event_append_statements(
&write_token, write_token.namespace().as_str(), "create",
khive_types::EventKind::NoteCreated, khive_types::SubstrateKind::Note,
note.note_id, json!({"id":note.note_id, "namespace":write_token.namespace().as_str(), "kind":"question", "salience":null}),
)?);
}
let committed = match &plan {
AtomicOpPlan::AddNote(plan) => CommittedFixAction::ProposeDuplicateMerge {
violation_index: identity.violation_index,
question_note_id: plan.note_id,
},
AtomicOpPlan::Delete(_) => CommittedFixAction::SoftDeleteEdge {
violation_index: identity.violation_index,
},
AtomicOpPlan::Update(_) => CommittedFixAction::MarkDeprecated {
violation_index: identity.violation_index,
},
_ => return Err(RuntimeError::Internal("unexpected graph lint plan".into())),
};
Ok(Candidate::Ready(Box::new(PreparedFix {
owner: runtime.clone(),
token: write_token,
guards,
plan,
committed,
})))
}
pub async fn apply_fixes(
&self,
registry: &VerbRegistry,
token: &NamespaceToken,
mut report: LintReport,
) -> RuntimeResult<(LintReport, Vec<FixReceipt>)> {
if token.actor() != self.token.actor()
|| token.namespace() != self.token.namespace()
|| token.visible_namespaces() != self.token.visible_namespaces()
|| token.request_id() != self.token.request_id()
|| token.process_ref() != self.token.process_ref()
{
return Err(RuntimeError::InvalidInput(
"lint capture belongs to a different request scope".into(),
));
}
let mut prepared = Vec::new();
for (index, violation) in report.violations.iter().enumerate() {
let action = match violation.rule {
LintRuleId::DanglingEdge => FixAction::SoftDeleteEdge,
LintRuleId::DuplicateName => FixAction::ProposeDuplicateMerge,
LintRuleId::SupersededNotMarked => FixAction::MarkDeprecated,
_ => continue,
};
let identity = ActionIdentity {
violation_index: index as u64,
action,
};
let candidate = self
.prepare_fix(registry, token, &identity, violation)
.await;
prepared.push((identity, candidate));
}
report.summary.fixed = Some(0);
let mut receipts = Vec::new();
for (identity, candidate) in prepared {
let candidate = candidate.map_err(|source| {
failure(
source,
LintFixOutcome::Failed,
identity.clone(),
&receipts,
&report,
)
})?;
match candidate {
Candidate::Unsupported(reason) => {
receipts.push(FixReceipt::Unsupported { identity, reason })
}
Candidate::Conflict(reason) => {
receipts.push(FixReceipt::Conflict { identity, reason })
}
Candidate::Ready(prepared) => {
let PreparedFix {
owner,
token,
guards,
plan,
committed,
} = *prepared;
let operation: PreparedAtomicOp<(), ActionRefusal> = Box::new(move |writer| {
Box::pin(async move {
for guard in guards {
if writer.query_all(guard).await?.len() != 1 {
return Err(PreparedAtomicError::Refused {
failure: ActionRefusal::Conflict(
"an observed lint predicate changed".into(),
),
message: "graph lint witness conflict".into(),
});
}
}
let applied = crate::atomic_runner::apply_plan_preserving_storage(
writer, &plan, true,
)
.await
.map_err(|failure| match failure {
crate::atomic_runner::PlanExecutionFailure::Storage {
source,
..
} => PreparedAtomicError::Storage(source),
crate::atomic_runner::PlanExecutionFailure::Operation(failure) => {
let failure = match failure {
crate::atomic_runner::AtomicOpFailure::SqlError {
message,
..
} => ActionRefusal::Failed(RuntimeError::Internal(message)),
_ => ActionRefusal::Conflict(
"an ordinary prepared mutation guard changed".into(),
),
};
PreparedAtomicError::Refused {
failure,
message: "graph lint plan refused".into(),
}
}
})?;
Ok(((), applied.effect.into_iter().collect()))
})
});
match run_prepared_atomic_unit(action_access(&owner).as_ref(), operation).await
{
Ok(PreparedAtomicOutcome::RolledBack(ActionRefusal::Conflict(reason))) => {
receipts.push(FixReceipt::Conflict { identity, reason })
}
Ok(PreparedAtomicOutcome::RolledBack(ActionRefusal::Failed(source))) => {
return Err(failure(
source,
LintFixOutcome::Failed,
identity,
&receipts,
&report,
))
}
Ok(PreparedAtomicOutcome::Committed { post_commit, .. }) => {
receipts.push(FixReceipt::Committed { action: committed });
report.summary.fixed = Some(
receipts
.iter()
.filter(|receipt| receipt.is_committed())
.count() as u64,
);
for effect in post_commit.into_effects() {
if let Err(source) =
crate::atomic_prepare::apply_one_post_commit_effect(
&owner, &token, effect,
)
.await
{
return Err(failure(
source,
LintFixOutcome::Failed,
identity,
&receipts,
&report,
));
}
}
}
Err(source) => {
let outcome = if unknown(&source) {
LintFixOutcome::Unknown
} else {
LintFixOutcome::Failed
};
return Err(failure(
source.into(),
outcome,
identity,
&receipts,
&report,
));
}
}
}
}
}
Ok((report, receipts))
}
}
fn unknown(error: &StorageError) -> bool {
if error
.sqlite_write_failure()
.is_some_and(|failure| failure.settlement_unknown)
{
return true;
}
match error.without_sqlite_write_stage() {
StorageError::WriterTaskRequestFailed { request_state, .. }
| StorageError::WriterTaskTerminated { request_state, .. } => {
*request_state == khive_storage::WriterTaskRequestState::SideEffectsUnknown
}
_ => false,
}
}
fn failure(
source: RuntimeError,
outcome: LintFixOutcome,
failed_action: ActionIdentity,
receipts: &[FixReceipt],
report: &LintReport,
) -> RuntimeError {
RuntimeError::LintFix {
source: Box::new(source),
detail: Box::new(LintFixFailureDetail {
outcome,
failed_action,
confirmed_receipts: receipts.to_vec(),
report: report.clone(),
}),
}
}
fn action_access(owner: &KhiveRuntime) -> std::sync::Arc<dyn khive_storage::SqlAccess> {
#[cfg(test)]
if let Ok(access) = TEST_ACTION_ACCESS.try_with(std::sync::Arc::clone) {
return access;
}
owner.sql()
}
#[cfg(test)]
tokio::task_local! {
pub(super) static TEST_ACTION_ACCESS: std::sync::Arc<dyn khive_storage::SqlAccess>;
}