semantic_memory/

authority.rs

//! Atomic, capability-gated authority mutations over the SQLite fact graph.
//!
//! This is deliberately a narrow write surface. Facts remain append-only; the
//! authority tables record lineage heads while graph edges preserve the
//! supersession or redaction event that caused a head transition.

use crate::authority_contracts::{
    AuthorityAdmission, AuthorityFaultStage, AuthorityOperationKind, AuthorityPermit,
    AuthorityReceiptV1, AuthoritySnapshotId, AuthorityStateV1, RetrievalEpoch,
};
use crate::db::with_transaction;
use crate::error::MemoryError;
use crate::origin_authority::{
    decide, label_digest, GovernedAccessRequestV1, GovernedFactAccessV1,
    GovernedFactListResponseV1, GovernedGraphResponseV1, GovernedReplayResponseV1,
    GovernedSearchResponseV1, GovernedStateResolutionResponseV1, OriginAuthorityLabelV1,
    OriginAuthorityRecordV1, OriginDerivationKindV1,
};
use crate::quantize::{self, Quantizer};
use crate::transition_contracts::{
    MemoryTransitionCandidateV1, MemoryTransitionOutcomeV1, MemoryTransitionRecordV1,
    MemoryTransitionVerificationV1, TransitionDisposition, TransitionOperation,
    MEMORY_TRANSITION_RECORD_V1,
};
use crate::transition_verifier::{digest as transition_digest, verify_candidate};
use crate::MemoryStore;
use chrono::Utc;
use rusqlite::{params, Connection, OptionalExtension, Transaction};
use serde::Serialize;
use std::sync::{Arc, Mutex};

const RECEIPT_SCHEMA: &str = "authority_receipt_v1";
const REDACTED_CONTENT: &str = "[REDACTED]";

#[derive(Clone)]
pub struct MemoryAuthority {
    store: MemoryStore,
}

impl MemoryAuthority {
    pub(crate) fn new(store: MemoryStore) -> Self {
        Self { store }
    }

    /// Append a new fact and start a new authority lineage.
    pub async fn append(
        &self,
        permit: AuthorityPermit,
        caller_idempotency_key: String,
        namespace: String,
        content: String,
        source: Option<String>,
    ) -> Result<AuthorityReceiptV1, MemoryError> {
        self.append_with_metadata(
            permit,
            caller_idempotency_key,
            namespace,
            content,
            source,
            None,
        )
        .await
    }

    /// Append a new governed fact with caller metadata atomically bound to the
    /// fact row, authority lineage, journal entry, and idempotency payload.
    ///
    /// Caller metadata must be a JSON object. Reserved `authority` metadata is
    /// always generated by this authority boundary and cannot be caller-set.
    pub async fn append_with_metadata(
        &self,
        permit: AuthorityPermit,
        caller_idempotency_key: String,
        namespace: String,
        content: String,
        source: Option<String>,
        metadata: Option<serde_json::Value>,
    ) -> Result<AuthorityReceiptV1, MemoryError> {
        self.mutate(
            permit,
            caller_idempotency_key,
            Mutation::Append {
                namespace,
                content,
                source,
                metadata,
            },
        )
        .await
    }

    /// Append a replacement fact and supersede the current lineage head.
    pub async fn supersede(
        &self,
        permit: AuthorityPermit,
        caller_idempotency_key: String,
        target_fact_id: String,
        content: String,
        source: Option<String>,
    ) -> Result<AuthorityReceiptV1, MemoryError> {
        self.mutate(
            permit,
            caller_idempotency_key,
            Mutation::Supersede {
                target_fact_id,
                content,
                source,
            },
        )
        .await
    }

    /// Append a redaction tombstone and make it the current lineage head.
    pub async fn redact(
        &self,
        permit: AuthorityPermit,
        caller_idempotency_key: String,
        target_fact_id: String,
        reason: String,
    ) -> Result<AuthorityReceiptV1, MemoryError> {
        self.mutate(
            permit,
            caller_idempotency_key,
            Mutation::Redact {
                target_fact_id,
                reason,
            },
        )
        .await
    }

    /// Forget a canonical fact and every accessible derived artifact in one transaction.
    pub async fn forget(
        &self,
        permit: AuthorityPermit,
        caller_idempotency_key: String,
        request: crate::ForgettingClosureRequestV1,
    ) -> Result<crate::ForgettingClosureReceiptV1, MemoryError> {
        crate::forgetting::forget(&self.store, permit, caller_idempotency_key, request).await
    }

    /// Governed forgetting validates every root against the same policy evaluator before the
    /// closure planner can inspect or invalidate it. Administrative authority is an explicit,
    /// time-bounded elevation rather than an implicit consequence of a write permit.
    pub async fn forget_governed(
        &self,
        permit: AuthorityPermit,
        caller_idempotency_key: String,
        request: crate::ForgettingClosureRequestV1,
        access: GovernedAccessRequestV1,
    ) -> Result<crate::ForgettingClosureReceiptV1, MemoryError> {
        let access = access.with_purpose(crate::GovernedAccessPurposeV1::Admin);
        for fact_id in &request.root_fact_ids {
            let governed = self.get_fact_governed(fact_id, access.clone()).await?;
            if !governed.decision.allowed {
                return Err(MemoryError::OriginAuthorityRejected {
                    principal: access.caller.0.clone(),
                    reason: format!(
                        "governed forgetting denied for '{}': {}",
                        fact_id,
                        governed.decision.reasons.join(",")
                    ),
                });
            }
        }
        self.forget(permit, caller_idempotency_key, request).await
    }

    /// Deterministically verify a source-grounded candidate immediately before authority commit.
    ///
    /// Failed verification is durably quarantined and never invokes canonical mutation. A passing
    /// verification, its immutable evidence, and the existing authority mutation commit in one
    /// SQLite transaction.
    pub async fn verify_and_commit(
        &self,
        permit: AuthorityPermit,
        caller_idempotency_key: String,
        candidate: MemoryTransitionCandidateV1,
    ) -> Result<MemoryTransitionOutcomeV1, MemoryError> {
        let kind = candidate_operation_kind(&candidate);
        validate_authority_request(&permit, &caller_idempotency_key, kind)?;
        let prepared = match candidate_content_for_embedding(&candidate) {
            Some(content) => {
                let (embedding, sparse, sparse_representation) = self
                    .store
                    .embed_text_with_sparse_internal(content, crate::EmbeddingPurpose::Document)
                    .await?;
                self.store.validate_embedding_dimensions(&embedding)?;
                let embedding_bytes = crate::db::embedding_to_bytes(&embedding);
                let q8_bytes = Quantizer::new(self.store.inner.config.embedding.dimensions)
                    .quantize(&embedding)
                    .map(|qv| quantize::pack_quantized(&qv))
                    .ok();
                Some(FactEmbedding {
                    embedding: embedding_bytes,
                    q8: q8_bytes,
                    sparse,
                    sparse_representation,
                })
            }
            None => None,
        };
        let fault = self.store.inner.authority_fault.clone();
        let journal = self.store.replication_journal_identity();
        let outcome = self
            .store
            .with_write_conn(move |conn| {
                execute_compiled_transition(
                    conn,
                    &permit,
                    &caller_idempotency_key,
                    candidate,
                    &fault,
                    prepared,
                    journal,
                )
            })
            .await?;
        if matches!(outcome, MemoryTransitionOutcomeV1::Committed { .. }) {
            self.store.clear_search_cache();
        }
        Ok(outcome)
    }

    /// Inspect a committed or quarantined transition by caller idempotency key.
    pub async fn get_transition_by_idempotency_key(
        &self,
        caller_idempotency_key: &str,
    ) -> Result<Option<MemoryTransitionRecordV1>, MemoryError> {
        let key = caller_idempotency_key.to_string();
        self.store
            .with_read_conn(move |conn| get_transition_record(conn, &key))
            .await
    }

    /// Load an immutable selective-forgetting receipt by caller idempotency key.
    pub async fn get_forgetting_receipt_by_idempotency_key(
        &self,
        caller_idempotency_key: &str,
    ) -> Result<Option<crate::ForgettingClosureReceiptV1>, MemoryError> {
        crate::forgetting::get_receipt(&self.store, caller_idempotency_key).await
    }

    /// Find a committed authority receipt by operation identity.
    pub async fn get_receipt_by_operation_id(
        &self,
        operation_id: &str,
    ) -> Result<Option<AuthorityReceiptV1>, MemoryError> {
        let operation_id = operation_id.to_string();
        self.store
            .with_read_conn(move |conn| get_receipt(conn, "operation_id", &operation_id))
            .await
    }

    /// Find a committed authority receipt by the caller's idempotency key.
    pub async fn get_receipt_by_idempotency_key(
        &self,
        caller_idempotency_key: &str,
    ) -> Result<Option<AuthorityReceiptV1>, MemoryError> {
        let key = caller_idempotency_key.to_string();
        self.store
            .with_read_conn(move |conn| get_receipt(conn, "caller_idempotency_key", &key))
            .await
    }

    /// Load the immutable write-time origin label for a canonical fact.
    pub async fn get_origin_authority(
        &self,
        fact_id: &str,
    ) -> Result<Option<OriginAuthorityRecordV1>, MemoryError> {
        let fact_id = fact_id.to_string();
        self.store
            .with_read_conn(move |conn| load_origin_record(conn, &fact_id))
            .await
    }

    /// Direct-ID governed recall. Content is never returned when the decision denies access.
    pub async fn get_fact_governed(
        &self,
        fact_id: &str,
        request: GovernedAccessRequestV1,
    ) -> Result<GovernedFactAccessV1, MemoryError> {
        let fact_id = fact_id.to_string();
        self.store
            .with_read_conn(move |conn| governed_fact_access(conn, &fact_id, &request))
            .await
    }

    /// Governed export uses the same decision path as direct recall.
    pub async fn export_fact_governed(
        &self,
        fact_id: &str,
        request: GovernedAccessRequestV1,
    ) -> Result<GovernedFactAccessV1, MemoryError> {
        self.get_fact_governed(
            fact_id,
            request.with_purpose(crate::GovernedAccessPurposeV1::Export),
        )
        .await
    }

    /// Governed hybrid search filters every result after cache retrieval, preventing cache bypass.
    pub async fn search_governed(
        &self,
        query: &str,
        top_k: Option<usize>,
        request: GovernedAccessRequestV1,
    ) -> Result<GovernedSearchResponseV1, MemoryError> {
        self.search_governed_with_view(query, top_k, request, crate::StateView::Current)
            .await
    }

    /// Read the current authority snapshot and retrieval epoch for cache validation.
    pub async fn current_state(&self) -> Result<AuthorityStateV1, MemoryError> {
        self.store
            .with_read_conn(|conn| {
                with_transaction(conn, |tx| {
                    let epoch = current_epoch(tx)?;
                    let snapshot_id = snapshot_id(tx, epoch)?;
                    Ok(AuthorityStateV1 {
                        snapshot_id,
                        retrieval_epoch: RetrievalEpoch(epoch),
                    })
                })
            })
            .await
    }

    /// Read-only compatibility shim for callers that only need the retrieval epoch.
    pub(crate) async fn current_retrieval_epoch(&self) -> Result<RetrievalEpoch, MemoryError> {
        Ok(self.current_state().await?.retrieval_epoch)
    }

    /// Governed current or historical search. The candidate source is intentionally irrelevant:
    /// every returned row is passed through the same canonical evaluator used by direct access.
    pub async fn search_governed_with_view(
        &self,
        query: &str,
        top_k: Option<usize>,
        request: GovernedAccessRequestV1,
        view: crate::StateView,
    ) -> Result<GovernedSearchResponseV1, MemoryError> {
        let namespace = request.scope.namespace.clone();
        let raw = self
            .store
            .search_with_view(query, top_k, Some(&[namespace.as_str()]), None, view)
            .await?;
        let mut results = Vec::new();
        let mut decisions = Vec::new();
        for result in raw {
            let fact_id = match &result.source {
                crate::SearchSource::Fact { fact_id, .. } => fact_id.clone(),
                crate::SearchSource::Chunk { chunk_id, .. } => format!("chunk:{chunk_id}"),
                crate::SearchSource::Message { message_id, .. } => format!("message:{message_id}"),
                crate::SearchSource::Episode { episode_id, .. } => format!("episode:{episode_id}"),
                crate::SearchSource::Projection { projection_id, .. } => {
                    format!("projection:{projection_id}")
                }
            };
            let decision = if matches!(result.source, crate::SearchSource::Fact { .. }) {
                self.get_fact_governed(&fact_id, request.clone())
                    .await?
                    .decision
            } else {
                decide(&fact_id, None, None, None, &request)
            };
            if decision.allowed {
                results.push(result);
            }
            decisions.push(decision);
        }
        Ok(GovernedSearchResponseV1 { results, decisions })
    }

    /// Governed enumeration applies the same direct-ID policy check to every listed fact.
    pub async fn list_facts_governed(
        &self,
        request: GovernedAccessRequestV1,
        limit: usize,
        offset: usize,
        view: crate::StateView,
    ) -> Result<GovernedFactListResponseV1, MemoryError> {
        let facts = self
            .store
            .list_facts_with_view(&request.scope.namespace, limit, offset, view)
            .await?;
        let mut admitted = Vec::new();
        let mut decisions = Vec::new();
        for fact in facts {
            let response = self.get_fact_governed(&fact.id, request.clone()).await?;
            if let Some(fact) = response.fact {
                admitted.push(fact);
            }
            decisions.push(response.decision);
        }
        Ok(GovernedFactListResponseV1 {
            facts: admitted,
            decisions,
        })
    }

    /// State resolution is candidate generation only; this wrapper removes denied assertions
    /// before an answer can leave the crate, including historical state views and cache-backed
    /// searches. The resolution receipt remains an audit record, not a content channel.
    pub async fn resolve_memory_governed(
        &self,
        query: &str,
        top_k: Option<usize>,
        request: GovernedAccessRequestV1,
        mode: crate::StateResolutionMode,
        budget: usize,
    ) -> Result<GovernedStateResolutionResponseV1, MemoryError> {
        let namespaces = [request.scope.namespace.as_str()];
        let mut response = self
            .store
            .resolve_memory(query, top_k, Some(&namespaces), mode, budget)
            .await?;
        let mut decisions = Vec::new();
        response.assertions.retain(|assertion| {
            let Some(fact_id) = assertion.memory_id.strip_prefix("fact:") else {
                return false;
            };
            // This async call cannot occur in `retain`; decisions are collected below.
            !fact_id.is_empty()
        });
        let mut admitted = Vec::new();
        for assertion in response.assertions.drain(..) {
            let fact_id = assertion
                .memory_id
                .strip_prefix("fact:")
                .unwrap_or_default();
            let decision = self
                .get_fact_governed(fact_id, request.clone())
                .await?
                .decision;
            if decision.allowed {
                admitted.push(assertion);
            }
            decisions.push(decision);
        }
        response.assertions = admitted;
        response.alternatives.retain(|alternative| {
            response
                .assertions
                .iter()
                .any(|assertion| assertion.memory_id == alternative.assertion.memory_id)
        });
        response.answer = response
            .assertions
            .first()
            .map(|assertion| assertion.content.clone());
        Ok(GovernedStateResolutionResponseV1 {
            response,
            decisions,
        })
    }

    /// Governed graph traversal authorizes every fact endpoint before returning an edge.
    pub async fn list_graph_edges_for_node_governed(
        &self,
        node_id: &str,
        request: GovernedAccessRequestV1,
    ) -> Result<GovernedGraphResponseV1, MemoryError> {
        let edges = self.store.list_graph_edges_for_node(node_id).await?;
        let mut admitted = Vec::new();
        let mut decisions = Vec::new();
        for edge in edges {
            let mut edge_allowed = true;
            for endpoint in [&edge.source, &edge.target] {
                if let Some(fact_id) = endpoint.strip_prefix("fact:") {
                    let decision = self
                        .get_fact_governed(fact_id, request.clone())
                        .await?
                        .decision;
                    edge_allowed &= decision.allowed;
                    decisions.push(decision);
                }
            }
            if edge_allowed {
                admitted.push(edge);
            }
        }
        Ok(GovernedGraphResponseV1 {
            edges: admitted,
            decisions,
        })
    }

    /// Replay a durable receipt, then authorize every replay result for the current caller.
    pub async fn replay_search_receipt_governed(
        &self,
        receipt_id: &str,
        query: &str,
        top_k: Option<usize>,
        request: GovernedAccessRequestV1,
    ) -> Result<GovernedReplayResponseV1, MemoryError> {
        let request = request.with_purpose(crate::GovernedAccessPurposeV1::Replay);
        let namespace = request.scope.namespace.clone();
        let replay = self
            .store
            .replay_search_receipt(receipt_id, query, top_k, Some(&[namespace.as_str()]), None)
            .await?;
        let mut allowed_result_ids = Vec::new();
        let mut decisions = Vec::new();
        for result_id in &replay.replay_receipt.result_ids {
            let decision = if let Some(fact_id) = result_id.strip_prefix("fact:") {
                self.get_fact_governed(fact_id, request.clone())
                    .await?
                    .decision
            } else {
                decide(result_id, None, None, None, &request)
            };
            if decision.allowed {
                allowed_result_ids.push(result_id.clone());
            }
            decisions.push(decision);
        }
        Ok(GovernedReplayResponseV1 {
            replay,
            allowed_result_ids,
            decisions,
        })
    }

    /// Append an origin revocation without mutating the immutable write-time label.
    pub async fn revoke_origin(
        &self,
        permit: AuthorityPermit,
        caller_idempotency_key: String,
        fact_id: &str,
        revocation_reference: String,
    ) -> Result<crate::OriginAuthorityDecisionV1, MemoryError> {
        if permit.capability != AuthorityPermit::REVOKE_ORIGIN_CAPABILITY
            || permit.principal.trim().is_empty()
            || caller_idempotency_key.trim().is_empty()
            || revocation_reference.trim().is_empty()
            || permit.origin_authority.is_none()
        {
            return Err(MemoryError::OriginAuthorityRejected {
                principal: permit.principal,
                reason: "revocation requires an origin-bound revoke capability and reference"
                    .into(),
            });
        }
        let fact_id = fact_id.to_string();
        let principal = permit.principal.clone();
        let request = GovernedAccessRequestV1::new(
            &principal,
            &principal,
            crate::GovernedAccessPurposeV1::Recall,
            "general",
        );
        self.store
            .with_write_conn(move |conn| {
                with_transaction(conn, |tx| {
                    let existing: Option<(String, String)> = tx
                        .query_row(
                            "SELECT fact_id, revocation_reference FROM origin_authority_revocations
                             WHERE caller_idempotency_key = ?1",
                            params![caller_idempotency_key],
                            |row| Ok((row.get(0)?, row.get(1)?)),
                        )
                        .optional()?;
                    if let Some((existing_fact, existing_reference)) = existing {
                        if existing_fact != fact_id || existing_reference != revocation_reference {
                            return Err(MemoryError::AuthorityIdempotencyConflict {
                                key: caller_idempotency_key,
                            });
                        }
                    } else {
                        let exists: bool = tx.query_row(
                            "SELECT EXISTS(SELECT 1 FROM origin_authority_labels WHERE fact_id = ?1)",
                            params![fact_id],
                            |row| row.get(0),
                        )?;
                        if !exists {
                            return Err(MemoryError::OriginAuthorityRejected {
                                principal: principal.clone(),
                                reason: "cannot revoke a fact without a canonical origin label".into(),
                            });
                        }
                        tx.execute(
                            "INSERT INTO origin_authority_revocations
                             (revocation_id, fact_id, caller_idempotency_key, principal,
                              revocation_reference, revoked_at)
                             VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
                            params![
                                uuid::Uuid::new_v4().to_string(),
                                fact_id,
                                caller_idempotency_key,
                                principal,
                                revocation_reference,
                                Utc::now().format("%Y-%m-%d %H:%M:%S%.6f").to_string(),
                            ],
                        )?;
                    }
                    let fact = crate::knowledge::get_fact(tx, &fact_id)?;
                    let origin = load_origin_record(tx, &fact_id)?;
                    Ok(decide(
                        &fact_id,
                        fact.as_ref().map(|fact| fact.namespace.as_str()),
                        origin.as_ref(),
                        Some(&revocation_reference),
                        &request,
                    ))
                })
            })
            .await
    }

    /// Install a one-shot fault for the next matching authority stage.
    #[cfg(any(test, feature = "testing"))]
    pub fn set_fault(&self, stage: Option<AuthorityFaultStage>) {
        let mut guard = self
            .store
            .inner
            .authority_fault
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        *guard = stage;
    }

    async fn mutate(
        &self,
        permit: AuthorityPermit,
        caller_idempotency_key: String,
        mutation: Mutation,
    ) -> Result<AuthorityReceiptV1, MemoryError> {
        validate_authority_request(&permit, &caller_idempotency_key, mutation.kind())?;

        let prepared = match mutation.content_for_embedding() {
            Some(content) => {
                let (embedding, sparse, sparse_representation) = self
                    .store
                    .embed_text_with_sparse_internal(content, crate::EmbeddingPurpose::Document)
                    .await?;
                self.store.validate_embedding_dimensions(&embedding)?;
                let embedding_bytes = crate::db::embedding_to_bytes(&embedding);
                let q8_bytes = Quantizer::new(self.store.inner.config.embedding.dimensions)
                    .quantize(&embedding)
                    .map(|qv| quantize::pack_quantized(&qv))
                    .ok();

                Some(FactEmbedding {
                    embedding: embedding_bytes,
                    q8: q8_bytes,
                    sparse,
                    sparse_representation,
                })
            }
            None => None,
        };

        let fault = self.store.inner.authority_fault.clone();
        let journal = self.store.replication_journal_identity();
        let result = self
            .store
            .with_write_conn(move |conn| {
                execute_mutation(
                    conn,
                    &permit,
                    &caller_idempotency_key,
                    mutation,
                    &fault,
                    prepared,
                    journal,
                )
            })
            .await?;
        self.store.clear_search_cache();
        Ok(result)
    }
}

fn validate_authority_request(
    permit: &AuthorityPermit,
    caller_idempotency_key: &str,
    kind: AuthorityOperationKind,
) -> Result<(), MemoryError> {
    if permit.principal.trim().is_empty()
        || permit.caller_id.trim().is_empty()
        || permit.capability != kind.capability()
        || caller_idempotency_key.trim().is_empty()
    {
        return Err(MemoryError::AuthorityUnauthorized {
            operation: kind.as_str().to_string(),
            principal: permit.principal.clone(),
        });
    }
    let append_admitted = match &permit.admission {
        AuthorityAdmission::OperatorSystem => true,
        AuthorityAdmission::Evidence { evidence_refs } => {
            !evidence_refs.is_empty()
                && evidence_refs.iter().all(|reference| {
                    let Some(hex) = reference.strip_prefix("blake3:") else {
                        return false;
                    };
                    hex.len() == 64 && hex.bytes().all(|byte| byte.is_ascii_hexdigit())
                })
        }
        AuthorityAdmission::Unspecified => false,
    };
    if kind == AuthorityOperationKind::Append && !append_admitted {
        return Err(MemoryError::AuthorityAdmissionRejected {
            principal: permit.principal.clone(),
            reason: "authoritative append requires resolver-produced evidence digests or an operator/system permit".into(),
        });
    }
    let origin =
        permit
            .origin_authority
            .as_ref()
            .ok_or_else(|| MemoryError::OriginAuthorityRejected {
                principal: permit.principal.clone(),
                reason: "governed canonical writes require an immutable origin label".into(),
            })?;
    if origin.origin_principal != permit.principal
        || origin.schema_version != crate::origin_authority::ORIGIN_AUTHORITY_LABEL_V1
        || origin.revocation_status != crate::RevocationStatusV1::Active
        || label_digest(origin).is_err()
    {
        return Err(MemoryError::OriginAuthorityRejected {
            principal: permit.principal.clone(),
            reason: "origin label is inconsistent, inactive, or bound to another principal".into(),
        });
    }
    Ok(())
}

#[derive(Debug)]
enum Mutation {
    Append {
        namespace: String,
        content: String,
        source: Option<String>,
        metadata: Option<serde_json::Value>,
    },
    Supersede {
        target_fact_id: String,
        content: String,
        source: Option<String>,
    },
    Redact {
        target_fact_id: String,
        reason: String,
    },
}

impl Mutation {
    fn content_for_embedding(&self) -> Option<&str> {
        match self {
            Self::Append { content, .. } => Some(content),
            Self::Supersede { content, .. } => Some(content),
            Self::Redact { .. } => None,
        }
    }

    fn kind(&self) -> AuthorityOperationKind {
        match self {
            Self::Append { .. } => AuthorityOperationKind::Append,
            Self::Supersede { .. } => AuthorityOperationKind::Supersede,
            Self::Redact { .. } => AuthorityOperationKind::Redact,
        }
    }
}

impl AuthorityOperationKind {
    fn as_str(self) -> &'static str {
        match self {
            Self::Append => "append",
            Self::Supersede => "supersede",
            Self::Redact => "redact",
        }
    }

    fn capability(self) -> &'static str {
        match self {
            Self::Append => AuthorityPermit::APPEND_CAPABILITY,
            Self::Supersede => AuthorityPermit::SUPERSEDE_CAPABILITY,
            Self::Redact => AuthorityPermit::REDACT_CAPABILITY,
        }
    }
}

#[derive(Debug)]
struct LineageTarget {
    fact_id: String,
    lineage_id: String,
    namespace: String,
    version: i64,
}

#[derive(Serialize)]
struct Payload<'a> {
    operation_kind: AuthorityOperationKind,
    mutation: &'a MutationForDigest,
    origin_label_digest: &'a str,
}

#[derive(Serialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
enum MutationForDigest {
    Append {
        namespace: String,
        content: String,
        source: Option<String>,
        metadata: Option<serde_json::Value>,
    },
    Supersede {
        target_fact_id: String,
        content: String,
        source: Option<String>,
    },
    Redact {
        target_fact_id: String,
        reason: String,
    },
}

fn candidate_operation_kind(candidate: &MemoryTransitionCandidateV1) -> AuthorityOperationKind {
    match candidate.operation {
        TransitionOperation::Append { .. } => AuthorityOperationKind::Append,
        TransitionOperation::Supersede { .. } => AuthorityOperationKind::Supersede,
        TransitionOperation::Retract { .. } => AuthorityOperationKind::Redact,
    }
}

fn candidate_content_for_embedding(candidate: &MemoryTransitionCandidateV1) -> Option<&str> {
    let assertion_id = match &candidate.operation {
        TransitionOperation::Append { assertion_id } => assertion_id,
        TransitionOperation::Supersede { draft } => &draft.replacement_assertion_id,
        TransitionOperation::Retract { .. } => return None,
    };
    candidate
        .assertions
        .iter()
        .find(|assertion| &assertion.assertion_id == assertion_id)
        .map(|assertion| assertion.content.as_str())
}

fn mutation_from_candidate(
    candidate: &MemoryTransitionCandidateV1,
    candidate_digest: &str,
) -> Result<Mutation, MemoryError> {
    let source = |spans: &[crate::SourceSpanRefV1]| {
        serde_json::to_string(&serde_json::json!({
            "memory_transition_candidate_id": candidate.candidate_id,
            "candidate_digest": candidate_digest,
            "source_spans": spans,
        }))
        .map(Some)
        .map_err(|error| MemoryError::Other(format!("serialize transition source: {error}")))
    };
    match &candidate.operation {
        TransitionOperation::Append { assertion_id } => {
            let assertion = candidate
                .assertions
                .iter()
                .find(|assertion| &assertion.assertion_id == assertion_id)
                .ok_or_else(|| MemoryError::InvalidConfig {
                    field: "transition.operation.assertion_id",
                    reason: "referenced assertion draft is missing".into(),
                })?;
            Ok(Mutation::Append {
                namespace: assertion.namespace.clone(),
                content: assertion.content.clone(),
                source: source(&assertion.source_spans)?,
                metadata: None,
            })
        }
        TransitionOperation::Supersede { draft } => {
            let assertion = candidate
                .assertions
                .iter()
                .find(|assertion| assertion.assertion_id == draft.replacement_assertion_id)
                .ok_or_else(|| MemoryError::InvalidConfig {
                    field: "transition.operation.replacement_assertion_id",
                    reason: "referenced replacement assertion draft is missing".into(),
                })?;
            Ok(Mutation::Supersede {
                target_fact_id: draft.target_fact_id.clone(),
                content: assertion.content.clone(),
                source: source(&assertion.source_spans)?,
            })
        }
        TransitionOperation::Retract {
            target_fact_id,
            reason,
            ..
        } => Ok(Mutation::Redact {
            target_fact_id: target_fact_id.clone(),
            reason: reason.clone(),
        }),
    }
}

fn execute_compiled_transition(
    conn: &Connection,
    permit: &AuthorityPermit,
    key: &str,
    candidate: MemoryTransitionCandidateV1,
    fault: &Arc<Mutex<Option<AuthorityFaultStage>>>,
    prepared: Option<FactEmbedding>,
    journal: Option<(String, String, u64)>,
) -> Result<MemoryTransitionOutcomeV1, MemoryError> {
    let candidate_digest = transition_digest(&candidate)?;
    // Safety: verification, quarantine/evidence persistence, and the existing canonical authority
    // mutation share this transaction, so no verified partial transition can become observable.
    with_transaction(conn, |tx| {
        if let Some(record) = get_transition_record(tx, key)? {
            if record.candidate_digest != candidate_digest
                || record.principal != permit.principal
                || record.caller_id != permit.caller_id
            {
                return Err(MemoryError::AuthorityIdempotencyConflict {
                    key: key.to_string(),
                });
            }
            return outcome_from_record(tx, record);
        }

        let verification = verify_candidate(tx, &candidate)?;
        if verification.candidate_digest != candidate_digest {
            return Err(MemoryError::DigestError(
                "transition verifier candidate digest drift".into(),
            ));
        }
        if verification.disposition == TransitionDisposition::Quarantine {
            let record = build_transition_record(key, permit, candidate, verification, None);
            insert_transition_record(tx, &record)?;
            return Ok(MemoryTransitionOutcomeV1::Quarantined { record });
        }

        let mutation = mutation_from_candidate(&candidate, &candidate_digest)?;
        let authority_receipt =
            execute_mutation_tx(tx, permit, key, mutation, fault, prepared, journal)?;
        let record = build_transition_record(
            key,
            permit,
            candidate,
            verification.clone(),
            Some(authority_receipt.receipt_id.clone()),
        );
        insert_transition_record(tx, &record)?;
        Ok(MemoryTransitionOutcomeV1::Committed {
            record,
            verification,
            authority_receipt,
        })
    })
}

fn build_transition_record(
    key: &str,
    permit: &AuthorityPermit,
    candidate: MemoryTransitionCandidateV1,
    verification: MemoryTransitionVerificationV1,
    authority_receipt_id: Option<String>,
) -> MemoryTransitionRecordV1 {
    MemoryTransitionRecordV1 {
        schema_version: MEMORY_TRANSITION_RECORD_V1.into(),
        record_id: uuid::Uuid::new_v4().to_string(),
        caller_idempotency_key: key.to_string(),
        principal: permit.principal.clone(),
        caller_id: permit.caller_id.clone(),
        candidate_digest: verification.candidate_digest.clone(),
        disposition: verification.disposition,
        candidate,
        verification,
        authority_receipt_id,
        created_at: Utc::now().format("%Y-%m-%d %H:%M:%S%.6f").to_string(),
    }
}

fn insert_transition_record(
    tx: &Transaction<'_>,
    record: &MemoryTransitionRecordV1,
) -> Result<(), MemoryError> {
    let candidate_json = serde_json::to_string(&record.candidate)
        .map_err(|error| MemoryError::Other(format!("serialize transition candidate: {error}")))?;
    let verification_json = serde_json::to_string(&record.verification).map_err(|error| {
        MemoryError::Other(format!("serialize transition verification: {error}"))
    })?;
    tx.execute(
        "INSERT INTO memory_transition_records
         (record_id, caller_idempotency_key, principal, caller_id, candidate_digest,
          candidate_json, verification_json, disposition, authority_receipt_id, created_at)
         VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)",
        params![
            record.record_id,
            record.caller_idempotency_key,
            record.principal,
            record.caller_id,
            record.candidate_digest,
            candidate_json,
            verification_json,
            match record.disposition {
                TransitionDisposition::Commit => "commit",
                TransitionDisposition::Quarantine => "quarantine",
            },
            record.authority_receipt_id,
            record.created_at,
        ],
    )?;
    Ok(())
}

fn outcome_from_record(
    conn: &Connection,
    record: MemoryTransitionRecordV1,
) -> Result<MemoryTransitionOutcomeV1, MemoryError> {
    match record.disposition {
        TransitionDisposition::Quarantine => Ok(MemoryTransitionOutcomeV1::Quarantined { record }),
        TransitionDisposition::Commit => {
            let receipt_id =
                record
                    .authority_receipt_id
                    .as_deref()
                    .ok_or_else(|| MemoryError::CorruptData {
                        table: "memory_transition_records",
                        row_id: record.record_id.clone(),
                        detail: "committed transition is missing authority receipt ID".into(),
                    })?;
            let receipt_json: String = conn.query_row(
                "SELECT receipt_json FROM authority_receipts WHERE receipt_id = ?1",
                params![receipt_id],
                |row| row.get(0),
            )?;
            let authority_receipt =
                serde_json::from_str(&receipt_json).map_err(|error| MemoryError::CorruptData {
                    table: "authority_receipts",
                    row_id: receipt_id.to_string(),
                    detail: format!("invalid stored receipt: {error}"),
                })?;
            Ok(MemoryTransitionOutcomeV1::Committed {
                verification: record.verification.clone(),
                record,
                authority_receipt,
            })
        }
    }
}

fn execute_mutation(
    conn: &Connection,
    permit: &AuthorityPermit,
    key: &str,
    mutation: Mutation,
    fault: &Arc<Mutex<Option<AuthorityFaultStage>>>,
    prepared: Option<FactEmbedding>,
    journal: Option<(String, String, u64)>,
) -> Result<AuthorityReceiptV1, MemoryError> {
    // Safety: this closure owns every canonical authority write and only commits after the
    // mutation journal, epoch, lineage, and receipt are complete.
    with_transaction(conn, |tx| {
        execute_mutation_tx(tx, permit, key, mutation, fault, prepared, journal)
    })
}

fn execute_mutation_tx(
    tx: &Transaction<'_>,
    permit: &AuthorityPermit,
    key: &str,
    mutation: Mutation,
    fault: &Arc<Mutex<Option<AuthorityFaultStage>>>,
    mut prepared: Option<FactEmbedding>,
    journal: Option<(String, String, u64)>,
) -> Result<AuthorityReceiptV1, MemoryError> {
    let kind = mutation.kind();
    let origin_label = effective_origin_label(tx, permit, &mutation)?;
    let origin_label_digest = label_digest(&origin_label).map_err(MemoryError::DigestError)?;
    let payload_digest = payload_digest(&mutation, &origin_label_digest)?;
    let operation = kind.as_str().to_string();

    if let Some(existing) = existing_operation(tx, key)? {
        if existing.payload_digest != payload_digest
            || existing.principal != permit.principal
            || existing.caller_id != permit.caller_id
            || existing.operation_kind != operation
        {
            return Err(MemoryError::AuthorityIdempotencyConflict {
                key: key.to_string(),
            });
        }
        let receipt: AuthorityReceiptV1 =
            serde_json::from_str(&existing.receipt_json).map_err(|e| MemoryError::CorruptData {
                table: "authority_receipts",
                row_id: key.to_string(),
                detail: format!("invalid stored receipt: {e}"),
            })?;
        return Ok(receipt);
    }

    verify_all_lineages(tx)?;
    let before_epoch = current_epoch(tx)?;
    let before_snapshot_id = snapshot_id(tx, before_epoch)?;
    let operation_id = uuid::Uuid::new_v4().to_string();
    let content_digest = mutation_content_digest(&mutation)?;

    fault_gate(fault, AuthorityFaultStage::BeforeAppend)?;
    let requires_embedding = matches!(
        mutation.kind(),
        AuthorityOperationKind::Append | AuthorityOperationKind::Supersede
    );
    let prepared = if requires_embedding {
        Some(prepared.take().ok_or_else(|| {
            MemoryError::Other("governed write is missing precomputed embedding".to_string())
        })?)
    } else {
        None
    };
    let (fact_id, lineage_id, target) =
        append_fact(tx, &mutation, &operation_id, &content_digest, prepared)?;
    persist_origin_label(tx, &fact_id, &origin_label, &origin_label_digest)?;
    fault_gate(fault, AuthorityFaultStage::AfterAppend)?;

    fault_gate(fault, AuthorityFaultStage::BeforeLineage)?;
    apply_lineage_transition(
        tx,
        &mutation,
        &operation_id,
        &fact_id,
        &lineage_id,
        target.as_ref(),
        &content_digest,
        before_epoch,
    )?;
    fault_gate(fault, AuthorityFaultStage::AfterLineage)?;
    verify_all_lineages(tx)?;

    let after_epoch = before_epoch
        .checked_add(1)
        .ok_or_else(|| MemoryError::Other("authority retrieval epoch overflow".to_string()))?;
    let affected_ids = affected_ids(&fact_id, &lineage_id, target.as_ref());
    let affected_json = serde_json::to_string(&affected_ids)
        .map_err(|e| MemoryError::Other(format!("serialize affected IDs: {e}")))?;
    let committed_at = Utc::now().format("%Y-%m-%d %H:%M:%S%.6f").to_string();

    fault_gate(fault, AuthorityFaultStage::BeforeJournal)?;
    tx.execute(
        "INSERT INTO operation_journal
             (operation_id, caller_idempotency_key, operation_kind, payload_digest,
              principal, caller_id, before_epoch, after_epoch, affected_ids_json,
              content_digest, committed_at)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11)",
        params![
            operation_id,
            key,
            operation,
            payload_digest,
            permit.principal,
            permit.caller_id,
            before_epoch as i64,
            after_epoch as i64,
            affected_json,
            content_digest,
            committed_at,
        ],
    )?;

    // Governed appends also enter the verified mutation outbox in this same
    // transaction. The outbox row carries the exact canonical payload the
    // replica replays (fact ID, namespace, content, source, metadata) plus the
    // digest chain allocated by the replication stream. Only fact.create has an
    // admitted replication contract today; supersede/redact intentionally emit
    // no outbox row, and a store without construction-time replication identity
    // remains local-only.
    if let Some((device_id, store_id, stream_epoch)) = journal {
        if let Mutation::Append {
            namespace,
            content,
            source,
            metadata,
        } = &mutation
        {
            let payload = crate::journal::encode_fact_create_payload(
                &crate::journal::FactCreatePayloadV1 {
                    fact_id: fact_id.clone(),
                    namespace: namespace.clone(),
                    content: content.clone(),
                    source: source.clone(),
                    metadata: metadata.clone(),
                },
            )?;
            crate::journal::append_verified_in_tx(
                tx,
                &device_id,
                &store_id,
                stream_epoch,
                crate::journal::FACT_CREATE_OPERATION,
                crate::journal::FACT_CREATE_PAYLOAD_SCHEMA,
                &payload,
            )?;
        }
    }
    fault_gate(fault, AuthorityFaultStage::AfterJournal)?;

    fault_gate(fault, AuthorityFaultStage::BeforeEpoch)?;
    let changed = tx.execute(
        "UPDATE authority_state SET retrieval_epoch = ?1 WHERE id = 1 AND retrieval_epoch = ?2",
        params![after_epoch as i64, before_epoch as i64],
    )?;
    if changed != 1 {
        return Err(MemoryError::Other(
            "authority retrieval epoch changed concurrently".to_string(),
        ));
    }
    tx.execute(
        "UPDATE authority_lineages SET updated_epoch = ?1 WHERE lineage_id = ?2",
        params![after_epoch as i64, lineage_id],
    )?;
    fault_gate(fault, AuthorityFaultStage::AfterEpoch)?;

    let after_snapshot_id = snapshot_id(tx, after_epoch)?;
    let receipt_id = uuid::Uuid::new_v4().to_string();
    let mut receipt = AuthorityReceiptV1 {
        schema_version: RECEIPT_SCHEMA.to_string(),
        receipt_id,
        operation_id,
        caller_idempotency_key: key.to_string(),
        principal: permit.principal.clone(),
        caller_id: permit.caller_id.clone(),
        operation_kind: kind,
        before_snapshot_id,
        after_snapshot_id,
        before_epoch: RetrievalEpoch(before_epoch),
        after_epoch: RetrievalEpoch(after_epoch),
        affected_ids,
        content_digest,
        origin_label_digest: Some(origin_label_digest),
        receipt_digest: String::new(),
        committed_at,
    };
    receipt.receipt_digest = digest_serialized(&receipt_without_digest(&receipt)?)?;
    let receipt_json = serde_json::to_string(&receipt)
        .map_err(|e| MemoryError::Other(format!("serialize authority receipt: {e}")))?;

    fault_gate(fault, AuthorityFaultStage::BeforeReceipt)?;
    tx.execute(
            "INSERT INTO authority_receipts
             (receipt_id, operation_id, caller_idempotency_key, receipt_json, receipt_digest, created_at)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
            params![
                receipt.receipt_id,
                receipt.operation_id,
                receipt.caller_idempotency_key,
                receipt_json,
                receipt.receipt_digest,
                receipt.committed_at,
            ],
        )?;
    fault_gate(fault, AuthorityFaultStage::AfterReceipt)?;
    Ok(receipt)
}

fn append_fact(
    tx: &Transaction<'_>,
    mutation: &Mutation,
    operation_id: &str,
    content_digest: &str,
    prepared: Option<FactEmbedding>,
) -> Result<(String, String, Option<LineageTarget>), MemoryError> {
    let (namespace, content, source, caller_metadata, target) = match mutation {
        Mutation::Append {
            namespace,
            content,
            source,
            metadata,
        } => (
            namespace.clone(),
            content.clone(),
            source.clone(),
            metadata.clone(),
            None,
        ),
        Mutation::Supersede {
            target_fact_id,
            content,
            source,
        } => {
            let target = load_active_target(tx, target_fact_id)?;
            (
                target.namespace.clone(),
                content.clone(),
                source.clone(),
                None,
                Some(target),
            )
        }
        Mutation::Redact {
            target_fact_id,
            reason: _,
        } => {
            let target = load_active_target(tx, target_fact_id)?;
            (
                target.namespace.clone(),
                REDACTED_CONTENT.to_string(),
                None,
                None,
                Some(target),
            )
        }
    };
    let source = source.as_deref();
    if namespace.trim().is_empty() || content.is_empty() {
        return Err(MemoryError::InvalidConfig {
            field: "authority.fact",
            reason: "namespace and content must not be empty".to_string(),
        });
    }

    let fact_id = uuid::Uuid::new_v4().to_string();
    let lineage_id = target
        .as_ref()
        .map(|value| value.lineage_id.clone())
        .unwrap_or_else(|| uuid::Uuid::new_v4().to_string());

    match mutation.kind() {
        AuthorityOperationKind::Append | AuthorityOperationKind::Supersede => {
            let prepared = prepared.ok_or_else(|| {
                MemoryError::Other("governed write is missing precomputed embedding".to_string())
            })?;
            if prepared.embedding.is_empty() {
                return Err(MemoryError::Other(
                    "governed write produced empty embedding bytes".to_string(),
                ));
            }
            let mut metadata = match caller_metadata {
                Some(serde_json::Value::Object(map)) => map,
                Some(_) => {
                    return Err(MemoryError::InvalidConfig {
                        field: "authority.fact.metadata",
                        reason: "caller metadata must be a JSON object".to_string(),
                    })
                }
                None => serde_json::Map::new(),
            };
            // Authority-owned fields are generated inside this transaction and
            // overwrite any caller attempt to spoof the reserved key.
            metadata.insert(
                "authority".to_string(),
                serde_json::json!({
                    "operation_id": operation_id,
                    "lineage_id": lineage_id,
                    "content_digest": content_digest,
                }),
            );
            let metadata = serde_json::Value::Object(metadata);
            crate::knowledge::insert_fact_in_tx(
                tx,
                &fact_id,
                &namespace,
                &content,
                &prepared.embedding,
                prepared.q8.as_deref(),
                source,
                Some(&metadata),
            )?;
            if let Some((weights, representation)) = prepared
                .sparse
                .as_ref()
                .zip(prepared.sparse_representation.as_deref())
            {
                crate::db::store_sparse_vector(
                    tx,
                    &format!("fact:{fact_id}"),
                    weights,
                    representation,
                )?;
            }
        }
        AuthorityOperationKind::Redact => {
            let metadata = serde_json::json!({
                "authority": {
                    "operation_id": operation_id,
                    "lineage_id": lineage_id,
                    "content_digest": content_digest,
                }
            })
            .to_string();
            tx.execute(
                "INSERT INTO facts (id, namespace, content, source, embedding, metadata)
                 VALUES (?1, ?2, ?3, ?4, NULL, ?5)",
                params![fact_id, namespace, content, source, metadata],
            )?;
            tx.execute(
                "INSERT INTO facts_rowid_map (fact_id) VALUES (?1)",
                params![fact_id],
            )?;
            let fts_rowid = tx.last_insert_rowid();
            tx.execute(
                "INSERT INTO facts_fts(rowid, content) VALUES (?1, ?2)",
                params![fts_rowid, content],
            )?;
        }
    }
    Ok((fact_id, lineage_id, target))
}

#[derive(Debug)]
struct FactEmbedding {
    embedding: Vec<u8>,
    q8: Option<Vec<u8>>,
    sparse: Option<crate::SparseWeights>,
    sparse_representation: Option<String>,
}

fn apply_lineage_transition(
    tx: &Transaction<'_>,
    mutation: &Mutation,
    operation_id: &str,
    fact_id: &str,
    lineage_id: &str,
    target: Option<&LineageTarget>,
    content_digest: &str,
    before_epoch: u64,
) -> Result<(), MemoryError> {
    let kind = mutation.kind().as_str();
    if let Some(target) = target {
        let relation = if matches!(mutation, Mutation::Redact { .. }) {
            "redacts"
        } else {
            "supersedes"
        };
        let recorded_at = Utc::now().format("%Y-%m-%d %H:%M:%S%.6f").to_string();
        let edge_type = serde_json::json!({"type": "entity", "relation": relation}).to_string();
        let edge_metadata = serde_json::json!({
            "operation_id": operation_id,
            "lineage_id": lineage_id,
        })
        .to_string();
        let edge_digest = digest_serialized(&(
            format!("fact:{fact_id}"),
            format!("fact:{}", target_fact_id(target)),
            edge_type.clone(),
            edge_metadata.clone(),
        ))?;
        tx.execute(
            "INSERT INTO graph_edges
             (id, source, target, edge_type, weight, metadata, content_digest,
              recorded_at, valid_time, recorded_time)
             VALUES (?1, ?2, ?3, ?4, 1.0, ?5, ?6, ?7, ?7, ?7)",
            params![
                uuid::Uuid::new_v4().to_string(),
                format!("fact:{fact_id}"),
                format!("fact:{}", target_fact_id(target)),
                edge_type,
                edge_metadata,
                edge_digest,
                recorded_at,
            ],
        )?;
        tx.execute(
            "UPDATE authority_versions SET is_active = 0 WHERE fact_id = ?1 AND is_active = 1",
            params![target_fact_id(target)],
        )?;
        crate::db::delete_sparse_vector(tx, &format!("fact:{}", target_fact_id(target)))?;
        tx.execute(
            "INSERT INTO authority_versions
             (fact_id, lineage_id, version, operation_kind, is_active, is_redacted, content_digest)
             VALUES (?1, ?2, ?3, ?4, 1, ?5, ?6)",
            params![
                fact_id,
                lineage_id,
                target.version + 1,
                kind,
                if matches!(mutation, Mutation::Redact { .. }) {
                    1
                } else {
                    0
                },
                content_digest,
            ],
        )?;
        tx.execute(
            "UPDATE authority_lineages SET active_head_id = ?1 WHERE lineage_id = ?2",
            params![fact_id, lineage_id],
        )?;
    } else {
        tx.execute(
            "INSERT INTO authority_lineages (lineage_id, active_head_id, updated_epoch)
             VALUES (?1, ?2, ?3)",
            params![lineage_id, fact_id, before_epoch as i64],
        )?;
        tx.execute(
            "INSERT INTO authority_versions
             (fact_id, lineage_id, version, operation_kind, is_active, is_redacted, content_digest)
             VALUES (?1, ?2, 1, ?3, 1, 0, ?4)",
            params![fact_id, lineage_id, kind, content_digest],
        )?;
    }
    verify_lineage(tx, lineage_id)
}

fn target_fact_id(target: &LineageTarget) -> &str {
    &target.fact_id
}

fn load_active_target(tx: &Transaction<'_>, fact_id: &str) -> Result<LineageTarget, MemoryError> {
    let target: Option<(String, String, i64, i64, String)> = tx
        .query_row(
            "SELECT av.lineage_id, f.namespace, av.version, av.is_active, al.active_head_id
             FROM authority_versions av
             JOIN facts f ON f.id = av.fact_id
             JOIN authority_lineages al ON al.lineage_id = av.lineage_id
             WHERE av.fact_id = ?1",
            params![fact_id],
            |row| {
                Ok((
                    row.get(0)?,
                    row.get(1)?,
                    row.get(2)?,
                    row.get(3)?,
                    row.get(4)?,
                ))
            },
        )
        .optional()?;
    let Some((lineage_id, namespace, version, is_active, active_head_id)) = target else {
        return Err(MemoryError::FactNotFound(fact_id.to_string()));
    };
    if is_active != 1 || active_head_id != fact_id {
        return Err(MemoryError::AuthorityLineageInconsistent {
            lineage_id,
            detail: "target is not the active head".to_string(),
        });
    }
    verify_lineage(tx, &lineage_id)?;
    Ok(LineageTarget {
        fact_id: fact_id.to_string(),
        lineage_id,
        namespace,
        version,
    })
}

fn verify_lineage(tx: &Transaction<'_>, lineage_id: &str) -> Result<(), MemoryError> {
    let (active_count, active_head, stored_head): (i64, Option<String>, Option<String>) = tx
        .query_row(
            "SELECT
                 (SELECT COUNT(*) FROM authority_versions WHERE lineage_id = ?1 AND is_active = 1),
                 (SELECT fact_id FROM authority_versions WHERE lineage_id = ?1 AND is_active = 1),
                 (SELECT active_head_id FROM authority_lineages WHERE lineage_id = ?1)",
            params![lineage_id],
            |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
        )?;
    if active_count != 1 || active_head.is_none() || active_head != stored_head {
        return Err(MemoryError::AuthorityLineageInconsistent {
            lineage_id: lineage_id.to_string(),
            detail: format!(
                "expected one active version matching stored head; count={active_count}, active={active_head:?}, stored={stored_head:?}"
            ),
        });
    }
    Ok(())
}

fn verify_all_lineages(tx: &Transaction<'_>) -> Result<(), MemoryError> {
    let mut stmt = tx.prepare("SELECT lineage_id FROM authority_lineages ORDER BY lineage_id")?;
    let lineage_ids: Vec<String> = stmt
        .query_map([], |row| row.get(0))?
        .collect::<Result<Vec<_>, _>>()?;
    for lineage_id in lineage_ids {
        verify_lineage(tx, &lineage_id)?;
    }
    Ok(())
}

fn current_epoch(tx: &Transaction<'_>) -> Result<u64, MemoryError> {
    let value: i64 = tx.query_row(
        "SELECT retrieval_epoch FROM authority_state WHERE id = 1",
        [],
        |row| row.get(0),
    )?;
    u64::try_from(value).map_err(|_| MemoryError::Other("negative authority epoch".to_string()))
}

fn snapshot_id(tx: &Transaction<'_>, epoch: u64) -> Result<AuthoritySnapshotId, MemoryError> {
    let mut stmt = tx
        .prepare("SELECT lineage_id, active_head_id FROM authority_lineages ORDER BY lineage_id")?;
    let heads: Vec<(String, String)> = stmt
        .query_map([], |row| Ok((row.get(0)?, row.get(1)?)))?
        .collect::<Result<Vec<_>, _>>()?;
    let digest = digest_serialized(&(epoch, heads))?;
    Ok(AuthoritySnapshotId(format!("epoch:{epoch}:{digest}")))
}

fn affected_ids(fact_id: &str, lineage_id: &str, target: Option<&LineageTarget>) -> Vec<String> {
    let mut ids = vec![fact_id.to_string(), lineage_id.to_string()];
    if let Some(target) = target {
        ids.push(target_fact_id(target).to_string());
    }
    ids
}

fn payload_digest(mutation: &Mutation, origin_label_digest: &str) -> Result<String, MemoryError> {
    let digest_mutation = match mutation {
        Mutation::Append {
            namespace,
            content,
            source,
            metadata,
        } => MutationForDigest::Append {
            namespace: namespace.clone(),
            content: content.clone(),
            source: source.clone(),
            metadata: metadata.clone(),
        },
        Mutation::Supersede {
            target_fact_id,
            content,
            source,
        } => MutationForDigest::Supersede {
            target_fact_id: target_fact_id.clone(),
            content: content.clone(),
            source: source.clone(),
        },
        Mutation::Redact {
            target_fact_id,
            reason,
        } => MutationForDigest::Redact {
            target_fact_id: target_fact_id.clone(),
            reason: reason.clone(),
        },
    };
    digest_serialized(&Payload {
        operation_kind: mutation.kind(),
        mutation: &digest_mutation,
        origin_label_digest,
    })
}

fn effective_origin_label(
    tx: &Transaction<'_>,
    permit: &AuthorityPermit,
    mutation: &Mutation,
) -> Result<OriginAuthorityLabelV1, MemoryError> {
    let proposed =
        permit
            .origin_authority
            .clone()
            .ok_or_else(|| MemoryError::OriginAuthorityRejected {
                principal: permit.principal.clone(),
                reason: "canonical write has no origin label".into(),
            })?;
    let target_id = match mutation {
        Mutation::Append { namespace, .. } => {
            return proposed
                .bind_resource_scope(crate::NamespaceScopeV1::exact(namespace))
                .map_err(|reason| MemoryError::OriginAuthorityRejected {
                    principal: permit.principal.clone(),
                    reason,
                })
        }
        Mutation::Supersede { target_fact_id, .. } | Mutation::Redact { target_fact_id, .. } => {
            target_fact_id
        }
    };
    let target =
        load_origin_record(tx, target_id)?.ok_or_else(|| MemoryError::OriginAuthorityRejected {
            principal: permit.principal.clone(),
            reason: format!("target '{target_id}' has no canonical origin label"),
        })?;
    // A replacement inherits the target's immutable resource scope. A caller cannot use a
    // freshly supplied, broader label to widen a lineage during supersession or redaction.
    let proposed = proposed
        .bind_resource_scope(target.label.resource_scope.clone())
        .map_err(|reason| MemoryError::OriginAuthorityRejected {
            principal: permit.principal.clone(),
            reason,
        })?;
    let content_digest = mutation_content_digest(mutation)?;
    OriginAuthorityLabelV1::derive(
        &[target.label, proposed],
        OriginDerivationKindV1::Other,
        content_digest,
    )
    .map_err(|reason| MemoryError::OriginAuthorityRejected {
        principal: permit.principal.clone(),
        reason,
    })
}

fn persist_origin_label(
    tx: &Transaction<'_>,
    fact_id: &str,
    label: &OriginAuthorityLabelV1,
    label_digest: &str,
) -> Result<(), MemoryError> {
    let label_json = serde_json::to_string(label)
        .map_err(|error| MemoryError::Other(format!("serialize origin label: {error}")))?;
    tx.execute(
        "INSERT INTO origin_authority_labels (fact_id, label_json, label_digest, recorded_at)
         VALUES (?1, ?2, ?3, ?4)",
        params![
            fact_id,
            label_json,
            label_digest,
            Utc::now().format("%Y-%m-%d %H:%M:%S%.6f").to_string(),
        ],
    )?;
    Ok(())
}

fn load_origin_record(
    conn: &Connection,
    fact_id: &str,
) -> Result<Option<OriginAuthorityRecordV1>, MemoryError> {
    let row: Option<(String, String, String)> = conn
        .query_row(
            "SELECT label_json, label_digest, recorded_at FROM origin_authority_labels
             WHERE fact_id = ?1",
            params![fact_id],
            |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
        )
        .optional()?;
    row.map(|(json, label_digest, recorded_at)| {
        let label = serde_json::from_str(&json).map_err(|error| MemoryError::CorruptData {
            table: "origin_authority_labels",
            row_id: fact_id.into(),
            detail: format!("invalid origin label: {error}"),
        })?;
        Ok(OriginAuthorityRecordV1 {
            fact_id: fact_id.into(),
            label,
            label_digest,
            recorded_at,
        })
    })
    .transpose()
}

fn governed_fact_access(
    conn: &Connection,
    fact_id: &str,
    request: &GovernedAccessRequestV1,
) -> Result<GovernedFactAccessV1, MemoryError> {
    let fact = crate::knowledge::get_fact(conn, fact_id)?;
    let origin = load_origin_record(conn, fact_id)?;
    let revocation_reference: Option<String> = conn
        .query_row(
            "SELECT revocation_reference FROM origin_authority_revocations
             WHERE fact_id = ?1 ORDER BY revoked_at DESC LIMIT 1",
            params![fact_id],
            |row| row.get(0),
        )
        .optional()?;
    let decision = decide(
        fact_id,
        fact.as_ref().map(|fact| fact.namespace.as_str()),
        origin.as_ref(),
        revocation_reference.as_deref(),
        request,
    );
    Ok(GovernedFactAccessV1 {
        fact: decision.allowed.then_some(fact).flatten(),
        decision,
        origin,
    })
}

fn mutation_content_digest(mutation: &Mutation) -> Result<String, MemoryError> {
    match mutation {
        Mutation::Append {
            namespace,
            content,
            source,
            metadata,
        } => digest_serialized(&(namespace, content, source, metadata)),
        Mutation::Supersede {
            target_fact_id,
            content,
            source,
        } => digest_serialized(&(target_fact_id, content, source)),
        Mutation::Redact {
            target_fact_id,
            reason,
        } => digest_serialized(&(REDACTED_CONTENT, target_fact_id, reason)),
    }
}

fn digest_serialized<T: Serialize>(value: &T) -> Result<String, MemoryError> {
    let bytes = serde_json::to_vec(value)
        .map_err(|e| MemoryError::DigestError(format!("serialize digest input: {e}")))?;
    Ok(blake3::hash(&bytes).to_hex().to_string())
}

fn receipt_without_digest(receipt: &AuthorityReceiptV1) -> Result<AuthorityReceiptV1, MemoryError> {
    let mut value = receipt.clone();
    value.receipt_digest.clear();
    Ok(value)
}

#[derive(Debug)]
struct ExistingOperation {
    payload_digest: String,
    principal: String,
    caller_id: String,
    operation_kind: String,
    receipt_json: String,
}

fn existing_operation(
    tx: &Transaction<'_>,
    key: &str,
) -> Result<Option<ExistingOperation>, MemoryError> {
    tx.query_row(
        "SELECT oj.payload_digest, oj.principal, oj.caller_id, oj.operation_kind,
                ar.receipt_json
         FROM operation_journal oj
         JOIN authority_receipts ar ON ar.operation_id = oj.operation_id
         WHERE oj.caller_idempotency_key = ?1",
        params![key],
        |row| {
            Ok(ExistingOperation {
                payload_digest: row.get(0)?,
                principal: row.get(1)?,
                caller_id: row.get(2)?,
                operation_kind: row.get(3)?,
                receipt_json: row.get(4)?,
            })
        },
    )
    .optional()
    .map_err(MemoryError::Database)
}

fn get_receipt(
    conn: &Connection,
    field: &str,
    value: &str,
) -> Result<Option<AuthorityReceiptV1>, MemoryError> {
    let sql = format!("SELECT receipt_json FROM authority_receipts WHERE {field} = ?1");
    let json: Option<String> = conn
        .query_row(&sql, params![value], |row| row.get(0))
        .optional()?;
    json.map(|raw| {
        serde_json::from_str(&raw).map_err(|e| MemoryError::CorruptData {
            table: "authority_receipts",
            row_id: value.to_string(),
            detail: format!("invalid stored receipt: {e}"),
        })
    })
    .transpose()
}

fn get_transition_record(
    conn: &Connection,
    key: &str,
) -> Result<Option<MemoryTransitionRecordV1>, MemoryError> {
    type TransitionRow = (
        String,
        String,
        String,
        String,
        String,
        String,
        String,
        Option<String>,
        String,
    );
    let row: Option<TransitionRow> = conn
        .query_row(
            "SELECT record_id, principal, caller_id, candidate_digest, candidate_json,
                    verification_json, disposition, authority_receipt_id, created_at
             FROM memory_transition_records WHERE caller_idempotency_key = ?1",
            params![key],
            |row| {
                Ok((
                    row.get(0)?,
                    row.get(1)?,
                    row.get(2)?,
                    row.get(3)?,
                    row.get(4)?,
                    row.get(5)?,
                    row.get(6)?,
                    row.get(7)?,
                    row.get(8)?,
                ))
            },
        )
        .optional()?;
    let Some((
        record_id,
        principal,
        caller_id,
        candidate_digest,
        candidate_json,
        verification_json,
        disposition,
        authority_receipt_id,
        created_at,
    )) = row
    else {
        return Ok(None);
    };
    let candidate =
        serde_json::from_str(&candidate_json).map_err(|error| MemoryError::CorruptData {
            table: "memory_transition_records",
            row_id: record_id.clone(),
            detail: format!("invalid candidate JSON: {error}"),
        })?;
    let verification =
        serde_json::from_str(&verification_json).map_err(|error| MemoryError::CorruptData {
            table: "memory_transition_records",
            row_id: record_id.clone(),
            detail: format!("invalid verification JSON: {error}"),
        })?;
    let disposition = match disposition.as_str() {
        "commit" => TransitionDisposition::Commit,
        "quarantine" => TransitionDisposition::Quarantine,
        other => {
            return Err(MemoryError::CorruptData {
                table: "memory_transition_records",
                row_id: record_id,
                detail: format!("invalid transition disposition '{other}'"),
            })
        }
    };
    Ok(Some(MemoryTransitionRecordV1 {
        schema_version: MEMORY_TRANSITION_RECORD_V1.into(),
        record_id,
        caller_idempotency_key: key.to_string(),
        principal,
        caller_id,
        candidate_digest,
        candidate,
        verification,
        disposition,
        authority_receipt_id,
        created_at,
    }))
}

fn fault_gate(
    fault: &Arc<Mutex<Option<AuthorityFaultStage>>>,
    stage: AuthorityFaultStage,
) -> Result<(), MemoryError> {
    let mut guard = fault
        .lock()
        .unwrap_or_else(|poisoned| poisoned.into_inner());
    if *guard == Some(stage) {
        *guard = None;
        return Err(MemoryError::AuthorityFaultInjected { stage });
    }
    Ok(())
}