use chio_core::canonical::CanonicalBytes;
use chio_core::capability::token::CapabilityToken;
use chio_core::credit::CreditBondRow;
use chio_core::crypto::Keypair;
use chio_core::receipt::{body::ChioReceipt, lineage::ChildRequestReceipt};
use chio_log_redact::redacted;
use crate::capability_lineage::CapabilitySnapshot;
use crate::checkpoint::KernelCheckpoint;
#[derive(Debug, Clone)]
pub struct RetentionConfig {
pub retention_days: u64,
pub max_size_bytes: u64,
pub archive_path: String,
pub tenant_id: Option<String>,
pub check_interval_secs: u64,
pub explicit_cutoff_unix_secs: Option<u64>,
}
impl Default for RetentionConfig {
fn default() -> Self {
Self {
retention_days: 90,
max_size_bytes: 10_737_418_240,
archive_path: "receipts-archive.sqlite3".to_string(),
tenant_id: None,
check_interval_secs: 3_600,
explicit_cutoff_unix_secs: None,
}
}
}
pub struct RetentionMaintenanceHandle {
stop: std::sync::Arc<std::sync::atomic::AtomicBool>,
join: Option<std::thread::JoinHandle<()>>,
}
impl RetentionMaintenanceHandle {
pub(crate) fn spawn(store: std::sync::Arc<dyn ReceiptStore>, config: RetentionConfig) -> Self {
let stop = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
let worker_stop = std::sync::Arc::clone(&stop);
let interval = std::time::Duration::from_secs(config.check_interval_secs.max(1));
let join = std::thread::spawn(move || {
while !worker_stop.load(std::sync::atomic::Ordering::SeqCst) {
let mut waited = std::time::Duration::ZERO;
let slice = std::time::Duration::from_millis(200);
while waited < interval && !worker_stop.load(std::sync::atomic::Ordering::SeqCst) {
std::thread::sleep(slice);
waited += slice;
}
if worker_stop.load(std::sync::atomic::Ordering::SeqCst) {
break;
}
let outcome = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
store.rotate_receipts(&config)
}));
match outcome {
Ok(Ok(_archived)) => {
store.record_retention_rotation_outcome(None);
}
Ok(Err(error)) => {
store.record_retention_rotation_outcome(Some(&error.to_string()));
tracing::warn!(
target: "chio::retention",
error = %redacted!(&error),
"receipt rotation failed; will retry next interval"
);
}
Err(_panic) => {
store.record_retention_rotation_outcome(Some("receipt rotation panicked"));
tracing::warn!(
target: "chio::retention",
"receipt rotation panicked; will retry next interval"
);
}
}
}
});
Self {
stop,
join: Some(join),
}
}
}
impl Drop for RetentionMaintenanceHandle {
fn drop(&mut self) {
self.stop.store(true, std::sync::atomic::Ordering::SeqCst);
if let Some(join) = self.join.take() {
let _ = join.join();
}
}
}
#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct ReceiptWriterCounters {
pub accepted_total: u64,
pub committed_total: u64,
pub failed_total: u64,
pub saturated_total: u64,
pub inflight: u64,
#[serde(default)]
pub timed_out_total: u64,
#[serde(default)]
pub timed_out_inflight: u64,
#[serde(default)]
pub queue_depth: u64,
#[serde(default)]
pub last_commit_unix_ms: Option<u64>,
#[serde(default)]
pub first_accept_unix_ms: Option<u64>,
#[serde(default)]
pub last_error: Option<String>,
}
#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct ReceiptWalCheckpointReport {
pub busy: u64,
pub log_frames: u64,
pub checkpointed_frames: u64,
}
#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct ReceiptFlushReport {
pub writer: ReceiptWriterCounters,
pub latest_committed_entry_seq: u64,
#[serde(default)]
pub latest_checkpoint_seq: Option<u64>,
pub latest_checkpointed_entry_seq: u64,
#[serde(default)]
pub uncheckpointed_start_seq: Option<u64>,
#[serde(default)]
pub uncheckpointed_end_seq: Option<u64>,
#[serde(default)]
pub wal_checkpoint: Option<ReceiptWalCheckpointReport>,
#[serde(default)]
pub db_size_bytes: Option<u64>,
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct ReceiptCheckpointRange {
pub start_seq: u64,
pub end_seq: u64,
}
#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct ReceiptCheckpointStatusReport {
pub healthy: bool,
pub latest_committed_entry_seq: u64,
#[serde(default)]
pub latest_checkpoint_seq: Option<u64>,
pub latest_checkpointed_entry_seq: u64,
#[serde(default)]
pub next_range: Option<ReceiptCheckpointRange>,
#[serde(default)]
pub checkpoint_error: Option<String>,
#[serde(default)]
pub retention_watermark_entry_seq: Option<u64>,
}
#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct ReceiptStoreHealthReport {
pub healthy: bool,
pub writer: ReceiptWriterCounters,
#[serde(default = "receipt_writer_liveness_unknown_label")]
pub writer_liveness: String,
pub latest_committed_entry_seq: u64,
#[serde(default)]
pub latest_checkpoint_seq: Option<u64>,
pub latest_checkpointed_entry_seq: u64,
#[serde(default)]
pub uncheckpointed_start_seq: Option<u64>,
#[serde(default)]
pub uncheckpointed_end_seq: Option<u64>,
#[serde(default)]
pub checkpoint_error: Option<String>,
#[serde(default)]
pub db_size_bytes: Option<u64>,
#[serde(default)]
pub retention_watermark_entry_seq: Option<u64>,
#[serde(default)]
pub retention_error: Option<String>,
#[serde(default)]
pub writer_level: chio_supervisor::HealthLevel,
#[serde(default)]
pub writer_restart_total: u64,
}
#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct ReceiptCheckpointCreateReport {
pub created: bool,
#[serde(default)]
pub checkpoint_seq: Option<u64>,
#[serde(default)]
pub batch_start_seq: Option<u64>,
#[serde(default)]
pub batch_end_seq: Option<u64>,
pub latest_committed_entry_seq: u64,
pub latest_checkpointed_entry_seq: u64,
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct AuthorizationReceiptConsumption {
pub authorization_receipt_id: String,
pub consumer_receipt_id: String,
pub request_id: String,
pub session_id: String,
pub tool_call_id: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub tenant_id: Option<String>,
pub parameter_hash: String,
pub consumed_at_unix_ms: u64,
}
#[derive(Debug, thiserror::Error)]
pub enum ReceiptStoreError {
#[error("sqlite error: {0}")]
Sqlite(#[from] rusqlite::Error),
#[error("sqlite pool error: {0}")]
Pool(String),
#[error("{operation} timed out after {timeout_ms}ms")]
Timeout { operation: String, timeout_ms: u64 },
#[error("serialization error: {0}")]
Json(#[from] serde_json::Error),
#[error("failed to prepare receipt store directory: {0}")]
Io(#[from] std::io::Error),
#[error("crypto decode error: {0}")]
CryptoDecode(String),
#[error("canonical json error: {0}")]
Canonical(String),
#[error("invalid outcome filter: {0}")]
InvalidOutcome(String),
#[error("receipt read boundary error: {0}")]
ReadBoundary(String),
#[error("conflict: {0}")]
Conflict(String),
#[error("not found: {0}")]
NotFound(String),
#[error("unsupported receipt-store operation: {0}")]
Unsupported(String),
#[error("receipt-store mutation was fenced")]
Fenced,
#[error("receipt-store durable outcome is unknown: {0}")]
OutcomeUnknown(String),
#[error("retention co-archival incomplete for {table}: {live} live rows, {archived} archived; aborting delete to preserve inclusion-proof integrity")]
RetentionArchiveIncomplete {
table: &'static str,
live: u64,
archived: u64,
},
#[error(
"retention watermark regression: attempted {attempted}, current high-water mark {current}"
)]
RetentionWatermarkRegression { attempted: u64, current: u64 },
#[error("claim receipt log projection is missing over a checkpointed or archived range (watermark {watermark}); the entry ordering cannot be safely regenerated to match committed checkpoint boundaries; restore the claim_receipt_log_entries projection from a backup taken before it was lost")]
ArchivedRangeProjection { watermark: u64 },
#[error("tenant-scoped retention is not expressible as a prefix watermark and is unsupported; no data was modified")]
RetentionTenantScopeUnsupported,
#[error("receipt commit writer is not serving after {restarts} restart(s): {last_error}")]
WriterDead { restarts: u64, last_error: String },
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AtomicReceiptProjection {
Unsupported,
SettlementObservationV1,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct PendingSettlementObservation {
pub next_visible_at_ms: u64,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum ReceiptWriterLiveness {
Healthy,
Saturated,
Wedged,
Dead,
Unknown,
}
impl ReceiptWriterLiveness {
pub fn healthy(self) -> bool {
matches!(self, Self::Healthy | Self::Unknown)
}
pub fn as_label(self) -> &'static str {
match self {
Self::Healthy => "healthy",
Self::Saturated => "saturated",
Self::Wedged => "wedged",
Self::Dead => "dead",
Self::Unknown => "unknown",
}
}
}
fn receipt_writer_liveness_unknown_label() -> String {
ReceiptWriterLiveness::Unknown.as_label().to_string()
}
pub trait ReceiptStore: Send + Sync {
fn append_chio_receipt(&self, receipt: &ChioReceipt) -> Result<(), ReceiptStoreError>;
fn admission_projection_capabilities(
&self,
) -> crate::admission_operation::AdmissionProjectionCapabilities {
crate::admission_operation::AdmissionProjectionCapabilities::default()
}
fn commit_admission_projection(
&self,
_projection: &crate::admission_operation::AdmissionTerminalProjection,
) -> Result<crate::admission_operation::AdmissionTerminal, ReceiptStoreError> {
Err(ReceiptStoreError::Unsupported(
"atomic admission terminal projection".to_string(),
))
}
fn settlement_store_binding(&self) -> Option<chio_settle::SettlementStoreBinding> {
None
}
fn atomic_receipt_projection(&self) -> AtomicReceiptProjection {
AtomicReceiptProjection::Unsupported
}
fn supports_atomic_receipt_projection_with_timeout(&self) -> bool {
false
}
fn append_chio_receipt_with_pending_observation(
&self,
_receipt: &ChioReceipt,
_pending: &PendingSettlementObservation,
) -> Result<(), ReceiptStoreError> {
Err(ReceiptStoreError::Unsupported(
"atomic settlement observation projection".to_string(),
))
}
fn append_chio_receipt_with_pending_observation_and_timeout(
&self,
_receipt: &ChioReceipt,
_pending: &PendingSettlementObservation,
_budget: std::time::Duration,
) -> Result<Option<u64>, ReceiptStoreError> {
Err(ReceiptStoreError::Unsupported(
"timeout-aware atomic settlement observation projection".to_string(),
))
}
fn load_chio_receipt(
&self,
_receipt_id: &str,
) -> Result<Option<ChioReceipt>, ReceiptStoreError> {
Ok(None)
}
fn load_child_receipt(
&self,
_receipt_id: &str,
) -> Result<Option<ChildRequestReceipt>, ReceiptStoreError> {
Ok(None)
}
fn append_chio_receipt_canonical(
&self,
receipt: &ChioReceipt,
_canonical: &CanonicalBytes,
) -> Result<(), ReceiptStoreError> {
self.append_chio_receipt(receipt)
}
fn append_chio_receipt_returning_seq(
&self,
receipt: &ChioReceipt,
) -> Result<Option<u64>, ReceiptStoreError> {
self.append_chio_receipt(receipt)?;
Ok(None)
}
fn append_chio_receipt_with_timeout(
&self,
receipt: &ChioReceipt,
_budget: std::time::Duration,
) -> Result<Option<u64>, ReceiptStoreError> {
self.append_chio_receipt_returning_seq(receipt)
}
fn writer_liveness(&self, _stall_threshold: std::time::Duration) -> ReceiptWriterLiveness {
ReceiptWriterLiveness::Unknown
}
fn append_chio_receipt_consuming_authorization(
&self,
_receipt: &ChioReceipt,
_consumption: &AuthorizationReceiptConsumption,
) -> Result<(), ReceiptStoreError> {
Err(ReceiptStoreError::Conflict(
"durable authorization receipt consumption is not supported by this receipt store"
.to_string(),
))
}
fn append_child_receipt(&self, receipt: &ChildRequestReceipt) -> Result<(), ReceiptStoreError>;
fn append_child_receipt_returning_seq(
&self,
receipt: &ChildRequestReceipt,
) -> Result<Option<u64>, ReceiptStoreError> {
self.append_child_receipt(receipt)?;
Ok(None)
}
fn append_child_receipt_with_timeout(
&self,
receipt: &ChildRequestReceipt,
_budget: std::time::Duration,
) -> Result<Option<u64>, ReceiptStoreError> {
self.append_child_receipt_returning_seq(receipt)
}
fn receipts_canonical_bytes_range(
&self,
_start_seq: u64,
_end_seq: u64,
) -> Result<Vec<(u64, Vec<u8>)>, ReceiptStoreError> {
Err(ReceiptStoreError::Conflict(
"receipt canonical byte ranges are not supported by this receipt store".to_string(),
))
}
fn flush_receipt_writes(&self) -> Result<ReceiptFlushReport, ReceiptStoreError> {
Err(ReceiptStoreError::Conflict(
"receipt writer flush is not supported by this receipt store".to_string(),
))
}
fn flush_receipt_writes_with_timeout(
&self,
_timeout: std::time::Duration,
) -> Result<ReceiptFlushReport, ReceiptStoreError> {
self.flush_receipt_writes()
}
fn receipt_store_health(&self) -> Result<ReceiptStoreHealthReport, ReceiptStoreError> {
Err(ReceiptStoreError::Conflict(
"receipt store health is not supported by this receipt store".to_string(),
))
}
fn writer_serving_closed(&self) -> bool {
false
}
fn latest_committed_entry_seq(&self) -> Result<u64, ReceiptStoreError> {
Err(ReceiptStoreError::Conflict(
"receipt committed sequence reporting is not supported by this receipt store"
.to_string(),
))
}
fn latest_checkpointed_entry_seq(&self) -> Result<u64, ReceiptStoreError> {
Err(ReceiptStoreError::Conflict(
"receipt checkpoint sequence reporting is not supported by this receipt store"
.to_string(),
))
}
fn next_checkpoint_range(
&self,
_max_batch: u64,
) -> Result<Option<ReceiptCheckpointRange>, ReceiptStoreError> {
Err(ReceiptStoreError::Conflict(
"receipt checkpoint ranges are not supported by this receipt store".to_string(),
))
}
fn receipt_checkpoint_status(
&self,
_max_batch: Option<u64>,
) -> Result<ReceiptCheckpointStatusReport, ReceiptStoreError> {
Err(ReceiptStoreError::Conflict(
"receipt checkpoint status is not supported by this receipt store".to_string(),
))
}
fn store_checkpoint(&self, _checkpoint: &KernelCheckpoint) -> Result<(), ReceiptStoreError> {
Err(ReceiptStoreError::Conflict(
"receipt checkpoint storage is not supported by this receipt store".to_string(),
))
}
fn create_next_receipt_checkpoint(
&self,
_max_batch: u64,
_keypair: &Keypair,
) -> Result<ReceiptCheckpointCreateReport, ReceiptStoreError> {
Err(ReceiptStoreError::Conflict(
"receipt checkpoint creation is not supported by this receipt store".to_string(),
))
}
fn load_checkpoint_by_seq(
&self,
_checkpoint_seq: u64,
) -> Result<Option<KernelCheckpoint>, ReceiptStoreError> {
Ok(None)
}
fn load_latest_checkpoint(&self) -> Result<Option<KernelCheckpoint>, ReceiptStoreError> {
let mut checkpoint_seq = 1;
let mut latest = None;
loop {
let Some(checkpoint) = self.load_checkpoint_by_seq(checkpoint_seq)? else {
return Ok(latest);
};
if checkpoint.body.checkpoint_seq != checkpoint_seq {
return Err(ReceiptStoreError::Conflict(format!(
"checkpoint loader returned checkpoint {} for requested sequence {}",
checkpoint.body.checkpoint_seq, checkpoint_seq
)));
}
checkpoint_seq = checkpoint
.body
.checkpoint_seq
.checked_add(1)
.ok_or_else(|| {
ReceiptStoreError::Conflict(
"checkpoint_seq overflow while loading latest".to_string(),
)
})?;
latest = Some(checkpoint);
}
}
fn supports_kernel_signed_checkpoints(&self) -> bool {
false
}
fn enable_background_checkpoints(
&self,
_keypair: Keypair,
_max_batch: u64,
) -> Result<bool, ReceiptStoreError> {
Ok(false)
}
fn supports_retention(&self) -> bool {
false
}
fn supports_tenant_scoped_retention(&self) -> bool {
false
}
fn rotate_receipts(&self, _config: &RetentionConfig) -> Result<u64, ReceiptStoreError> {
Err(ReceiptStoreError::Conflict(
"receipt retention is not supported by this receipt store".to_string(),
))
}
fn record_retention_rotation_outcome(&self, _failure: Option<&str>) {}
fn record_capability_snapshot(
&self,
_token: &CapabilityToken,
_parent_capability_id: Option<&str>,
) -> Result<(), ReceiptStoreError> {
Ok(())
}
fn record_capability_snapshot_with_timeout(
&self,
token: &CapabilityToken,
parent_capability_id: Option<&str>,
_budget: std::time::Duration,
) -> Result<(), ReceiptStoreError> {
self.record_capability_snapshot(token, parent_capability_id)
}
fn get_capability_snapshot(
&self,
_capability_id: &str,
) -> Result<Option<CapabilitySnapshot>, ReceiptStoreError> {
Ok(None)
}
fn get_capability_delegation_chain(
&self,
_capability_id: &str,
) -> Result<Vec<CapabilitySnapshot>, ReceiptStoreError> {
Ok(Vec::new())
}
fn resolve_credit_bond(
&self,
_bond_id: &str,
) -> Result<Option<CreditBondRow>, ReceiptStoreError> {
Ok(None)
}
fn record_session_anchor(
&self,
_session_id: &str,
_anchor_id: &str,
_auth_context_fingerprint: &str,
_issued_at: u64,
_supersedes_anchor_id: Option<&str>,
_anchor_json: &serde_json::Value,
) -> Result<(), ReceiptStoreError> {
Ok(())
}
#[allow(clippy::too_many_arguments)]
fn record_request_lineage(
&self,
_session_id: &str,
_request_id: &str,
_parent_request_id: Option<&str>,
_session_anchor_id: Option<&str>,
_recorded_at: u64,
_request_fingerprint: Option<&str>,
_lineage_json: &serde_json::Value,
) -> Result<(), ReceiptStoreError> {
Ok(())
}
#[allow(clippy::too_many_arguments)]
fn record_receipt_lineage_statement(
&self,
_child_receipt_id: &str,
_request_id: Option<&str>,
_session_id: Option<&str>,
_session_anchor_id: Option<&str>,
_parent_request_id: Option<&str>,
_parent_receipt_id: Option<&str>,
_chain_id: Option<&str>,
_recorded_at: u64,
_statement_json: &serde_json::Value,
) -> Result<(), ReceiptStoreError> {
Ok(())
}
fn get_receipt_lineage_verification(
&self,
_receipt_id: &str,
) -> Result<Option<ReceiptLineageVerification>, ReceiptStoreError> {
Ok(None)
}
fn list_receipt_lineage_statement_links(
&self,
_receipt_id: &str,
) -> Result<Vec<ReceiptLineageStatementLink>, ReceiptStoreError> {
Ok(Vec::new())
}
fn as_any_mut(&self) -> Option<&dyn std::any::Any> {
None
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct AdmissionBudgetAuthorization {
pub decision: crate::budget_store::BudgetAuthorizeHoldDecision,
pub operation: crate::admission_operation::AdmissionOperationV1,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct AdmissionBudgetCapture {
pub decision: crate::budget_store::BudgetInvocationCaptureDecision,
pub operation: crate::admission_operation::AdmissionOperationV1,
}
#[derive(Debug, Clone, Copy)]
pub struct AdmissionPaymentJournalAdvance<'a> {
pub operation: &'a crate::admission_operation::AdmissionOperationV1,
pub recovery_lease: &'a crate::admission_operation::AdmissionRecoveryLease,
pub expected: &'a crate::payment::PaymentJournalRecord,
pub transition: &'a crate::payment::PaymentJournalTransition,
pub release_evidence: Option<&'a crate::tool_outcome::MonetaryReleaseEvidenceV1>,
pub active_fence: &'a crate::admission_operation::StoreMutationFence,
pub trusted_now_unix_ms: u64,
}
#[derive(Debug, Clone)]
pub struct AdmissionPaymentSettlementBegin<'a> {
pub operation: &'a crate::admission_operation::AdmissionOperationV1,
pub recovery_lease: &'a crate::admission_operation::AdmissionRecoveryLease,
pub expected: &'a crate::payment::PaymentJournalRecord,
pub transition: Option<&'a crate::payment::PaymentJournalTransition>,
pub release_evidence: Option<&'a crate::tool_outcome::MonetaryReleaseEvidenceV1>,
pub budget_reconcile: crate::budget_store::BudgetReconcileHoldRequest,
pub active_fence: &'a crate::admission_operation::StoreMutationFence,
pub trusted_now_unix_ms: u64,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct AdmissionPaymentSettlement {
pub journal: crate::payment::PaymentJournalRecord,
pub budget: crate::budget_store::BudgetReconcileHoldDecision,
pub budget_already_reconciled: bool,
}
#[derive(Debug, thiserror::Error)]
pub enum AdmissionBudgetAuthorizationError {
#[error("combined admission budget authorization is unavailable: {0}")]
Unavailable(String),
#[error("combined admission budget authorization was fenced")]
Fenced,
#[error("combined admission budget authorization durable outcome is unknown: {0}")]
OutcomeUnknown(String),
#[error("combined admission budget authorization invariant failed: {0}")]
Invariant(String),
#[error(transparent)]
Operation(#[from] crate::admission_operation::AdmissionOperationError),
}
#[derive(Debug, thiserror::Error)]
pub enum AdmissionPaymentJournalError {
#[error("qualified payment journal is unavailable: {0}")]
Unavailable(String),
#[error("qualified payment journal mutation was fenced")]
Fenced,
#[error("qualified payment journal compare-and-set conflicted: {0}")]
Conflict(String),
#[error("qualified payment journal durable outcome is unknown: {0}")]
OutcomeUnknown(String),
#[error("qualified payment journal invariant failed: {0}")]
Invariant(String),
}
pub const ADMISSION_TERMINAL_PROJECTION_DESCRIPTOR_KIND: &str =
"chio.admission.terminal-projection.v1";
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ThresholdApprovalReplayReservationV1 {
proposal: chio_core::capability::governance::ThresholdApprovalProposal,
tokens: Vec<chio_core::capability::governance::GovernedApprovalToken>,
verified_set: chio_core::capability::governance::VerifiedApprovalSetBody,
}
impl ThresholdApprovalReplayReservationV1 {
pub fn new(
proposal: chio_core::capability::governance::ThresholdApprovalProposal,
mut tokens: Vec<chio_core::capability::governance::GovernedApprovalToken>,
verified_set: chio_core::capability::governance::VerifiedApprovalSetBody,
) -> Result<Self, crate::admission_operation::AdmissionOperationStoreError> {
use std::collections::HashSet;
if tokens.is_empty()
|| tokens.len()
> chio_core::capability::threshold_approval::MAX_THRESHOLD_APPROVAL_TOKENS
{
return Err(
crate::admission_operation::AdmissionOperationStoreError::Invariant(format!(
"threshold approval replay reservation must contain between 1 and {} tokens",
chio_core::capability::threshold_approval::MAX_THRESHOLD_APPROVAL_TOKENS
)),
);
}
if !proposal.verify_signature().map_err(|error| {
crate::admission_operation::AdmissionOperationStoreError::Invariant(error.to_string())
})? {
return Err(
crate::admission_operation::AdmissionOperationStoreError::Invariant(
"threshold approval replay proposal signature is invalid".to_owned(),
),
);
}
let proposal_hash = proposal.artifact_digest().map_err(|error| {
crate::admission_operation::AdmissionOperationStoreError::Invariant(error.to_string())
})?;
let mut token_ids = HashSet::new();
let mut approvers = HashSet::new();
let mut tokens_with_digests = Vec::with_capacity(tokens.len());
for token in tokens.drain(..) {
if token.id.is_empty()
|| token.id.trim() != token.id
|| token.threshold_proposal_hash.as_deref() != Some(proposal_hash.as_str())
|| token.request_id != proposal.body.request_id
|| token.governed_intent_hash != proposal.body.governed_intent_hash
|| token.subject != proposal.body.subject
|| token.decision
!= chio_core::capability::governance::GovernedApprovalDecision::Approved
|| token.issued_at < proposal.body.proposal_created_at
|| token.issued_at >= proposal.body.proposal_deadline
|| token.expires_at > proposal.body.proposal_deadline
|| !token_ids.insert(token.id.clone())
|| !approvers.insert(token.approver.to_hex())
{
return Err(
crate::admission_operation::AdmissionOperationStoreError::Invariant(
"threshold approval replay tokens do not form a distinct proposal set"
.to_owned(),
),
);
}
if !token.verify_signature().map_err(|error| {
crate::admission_operation::AdmissionOperationStoreError::Invariant(
error.to_string(),
)
})? {
return Err(
crate::admission_operation::AdmissionOperationStoreError::Invariant(
"threshold approval replay token signature is invalid".to_owned(),
),
);
}
let digest = token.artifact_digest().map_err(|error| {
crate::admission_operation::AdmissionOperationStoreError::Invariant(
error.to_string(),
)
})?;
tokens_with_digests.push((digest, token));
}
tokens_with_digests.sort_by(|left, right| left.0.cmp(&right.0));
if tokens_with_digests
.windows(2)
.any(|pair| pair[0].0 == pair[1].0)
|| tokens_with_digests
.iter()
.map(|(digest, _)| digest)
.ne(verified_set.token_digests.iter())
{
return Err(
crate::admission_operation::AdmissionOperationStoreError::Invariant(
"threshold approval replay token digests do not match the verified set"
.to_owned(),
),
);
}
let reconstructed = chio_core::capability::governance::VerifiedApprovalSetBody::new(
verified_set.token_digests.clone(),
&proposal,
)
.map_err(|error| {
crate::admission_operation::AdmissionOperationStoreError::Invariant(error.to_string())
})?;
if reconstructed != verified_set {
return Err(
crate::admission_operation::AdmissionOperationStoreError::Invariant(
"threshold approval replay set does not match its signed proposal".to_owned(),
),
);
}
Ok(Self {
proposal,
tokens: tokens_with_digests
.into_iter()
.map(|(_, token)| token)
.collect(),
verified_set,
})
}
#[must_use]
pub const fn proposal(&self) -> &chio_core::capability::governance::ThresholdApprovalProposal {
&self.proposal
}
#[must_use]
pub fn tokens(&self) -> &[chio_core::capability::governance::GovernedApprovalToken] {
&self.tokens
}
#[must_use]
pub const fn verified_set(
&self,
) -> &chio_core::capability::governance::VerifiedApprovalSetBody {
&self.verified_set
}
}
pub trait QualifiedAdmissionProjectionStore:
ReceiptStore + crate::admission_operation::QualifiedAdmissionOperationStore
{
fn load_payment_journal(
&self,
operation_id: &str,
active_fence: &crate::admission_operation::StoreMutationFence,
) -> Result<Option<crate::payment::PaymentJournalRecord>, AdmissionPaymentJournalError>;
fn advance_payment_journal(
&self,
advance: AdmissionPaymentJournalAdvance<'_>,
) -> Result<crate::payment::PaymentJournalRecord, AdmissionPaymentJournalError>;
fn begin_payment_settlement(
&self,
begin: AdmissionPaymentSettlementBegin<'_>,
) -> Result<AdmissionPaymentSettlement, AdmissionPaymentJournalError>;
#[allow(clippy::too_many_arguments)]
fn authorize_budget_and_commit_admission(
&self,
operation: &crate::admission_operation::AdmissionOperationV1,
recovery_lease: &crate::admission_operation::AdmissionRecoveryLease,
request: crate::budget_store::BudgetAuthorizeHoldRequest,
payment_journal: Option<crate::payment::PaymentJournalRecord>,
credit_exposure: Option<chio_credit::obligation::CreditExposureReservationRequest>,
active_fence: &crate::admission_operation::StoreMutationFence,
trusted_now_unix_ms: u64,
) -> Result<AdmissionBudgetAuthorization, AdmissionBudgetAuthorizationError>;
fn capture_invocation_and_commit_dispatch(
&self,
operation: &crate::admission_operation::AdmissionOperationV1,
recovery_lease: &crate::admission_operation::AdmissionRecoveryLease,
request: crate::budget_store::BudgetCaptureInvocationRequest,
active_fence: &crate::admission_operation::StoreMutationFence,
trusted_now_unix_ms: u64,
) -> Result<AdmissionBudgetCapture, crate::admission_operation::AdmissionCaptureError>;
fn reserve_threshold_approval_and_commit_admission(
&self,
_command: &crate::admission_operation::AdmissionOperationCommand,
_reservation: &ThresholdApprovalReplayReservationV1,
_trusted_now_unix_ms: u64,
) -> Result<
crate::admission_operation::AdmissionCommandResult,
crate::admission_operation::AdmissionOperationStoreError,
> {
Err(
crate::admission_operation::AdmissionOperationStoreError::Unavailable(
"durable threshold approval replay reservation is unsupported".to_owned(),
),
)
}
fn list_admission_receipts_after(
&self,
after_receipt_id: Option<&str>,
limit: usize,
) -> Result<Vec<ChioReceipt>, ReceiptStoreError>;
}
pub trait AnchoredAdmissionProjectionStore: QualifiedAdmissionProjectionStore {
fn stage_anchored_terminal_projection(
&self,
advance: &chio_core::economic_continuity::VerifiedEconomicStateBatchAdvance,
recovery_lease: &crate::admission_operation::AdmissionRecoveryLease,
envelope: &crate::admission_operation::SignedAdmissionTerminalProjectionV1,
active_fence: &crate::admission_operation::StoreMutationFence,
trusted_now_unix_ms: u64,
) -> Result<(), ReceiptStoreError>;
fn qualify_anchored_terminal_projection(
&self,
batch_id: &str,
active_fence: &crate::admission_operation::StoreMutationFence,
trusted_now_unix_ms: u64,
) -> Result<(), ReceiptStoreError>;
fn record_anchored_terminal_projection(
&self,
advance: &chio_core::economic_continuity::VerifiedEconomicStateBatchAdvance,
committed: &chio_core::economic_continuity::VerifiedEconomicStateView,
pins: &chio_core::economic_continuity::EconomicStateAnchorPins,
active_fence: &crate::admission_operation::StoreMutationFence,
trusted_now_unix_ms: u64,
) -> Result<(), ReceiptStoreError>;
fn commit_anchored_terminal_projection(
&self,
batch_id: &str,
active_fence: &crate::admission_operation::StoreMutationFence,
trusted_now_unix_ms: u64,
) -> Result<crate::admission_operation::AdmissionTerminal, ReceiptStoreError>;
}
#[cfg(test)]
mod tests {
use super::*;
use std::sync::atomic::{AtomicUsize, Ordering};
use chio_core::receipt::{
body::ChioReceiptBody, decision::Decision, decision::ToolCallAction, kinds::TrustLevel,
};
struct AppendOnlyStore;
impl ReceiptStore for AppendOnlyStore {
fn append_chio_receipt(&self, _receipt: &ChioReceipt) -> Result<(), ReceiptStoreError> {
Ok(())
}
fn append_child_receipt(
&self,
_receipt: &ChildRequestReceipt,
) -> Result<(), ReceiptStoreError> {
Ok(())
}
}
struct CountingAppendStore {
append_calls: AtomicUsize,
}
impl ReceiptStore for CountingAppendStore {
fn append_chio_receipt(&self, _receipt: &ChioReceipt) -> Result<(), ReceiptStoreError> {
self.append_calls.fetch_add(1, Ordering::SeqCst);
Ok(())
}
fn append_child_receipt(
&self,
_receipt: &ChildRequestReceipt,
) -> Result<(), ReceiptStoreError> {
Ok(())
}
}
struct LegacyAtomicOnlyStore {
atomic_append_calls: AtomicUsize,
}
impl ReceiptStore for LegacyAtomicOnlyStore {
fn append_chio_receipt(&self, _receipt: &ChioReceipt) -> Result<(), ReceiptStoreError> {
Ok(())
}
fn atomic_receipt_projection(&self) -> AtomicReceiptProjection {
AtomicReceiptProjection::SettlementObservationV1
}
fn append_chio_receipt_with_pending_observation(
&self,
_receipt: &ChioReceipt,
_pending: &PendingSettlementObservation,
) -> Result<(), ReceiptStoreError> {
self.atomic_append_calls.fetch_add(1, Ordering::SeqCst);
Ok(())
}
fn append_child_receipt(
&self,
_receipt: &ChildRequestReceipt,
) -> Result<(), ReceiptStoreError> {
Ok(())
}
}
fn signed_receipt() -> ChioReceipt {
let keypair = Keypair::generate();
let action = match ToolCallAction::from_parameters(serde_json::json!({})) {
Ok(action) => action,
Err(error) => panic!("test action construction failed: {error}"),
};
let body = ChioReceiptBody {
id: "receipt-1".to_string(),
timestamp: 1,
capability_id: "capability-1".to_string(),
tool_server: "server".to_string(),
tool_name: "tool".to_string(),
action,
decision: Some(Decision::Allow),
receipt_kind: Default::default(),
boundary_class: Default::default(),
observation_outcome: None,
tool_origin: Default::default(),
redaction_mode: Default::default(),
actor_chain: Vec::new(),
content_hash: "content".to_string(),
policy_hash: "policy".to_string(),
evidence: Vec::new(),
metadata: None,
trust_level: TrustLevel::default(),
tenant_id: None,
kernel_key: keypair.public_key(),
bbs_projection_version: None,
};
match ChioReceipt::sign(body, &keypair) {
Ok(receipt) => receipt,
Err(error) => panic!("test receipt signing failed: {error}"),
}
}
#[test]
fn unsupported_atomic_projection_never_falls_back_to_receipt_only_append() {
let store = CountingAppendStore {
append_calls: AtomicUsize::new(0),
};
let receipt = signed_receipt();
assert_eq!(
store.atomic_receipt_projection(),
AtomicReceiptProjection::Unsupported
);
assert!(!store.supports_atomic_receipt_projection_with_timeout());
assert_eq!(store.settlement_store_binding(), None);
let pending = PendingSettlementObservation {
next_visible_at_ms: 1,
};
assert!(matches!(
store.append_chio_receipt_with_pending_observation(&receipt, &pending),
Err(ReceiptStoreError::Unsupported(_))
));
assert!(matches!(
store.append_chio_receipt_with_pending_observation_and_timeout(
&receipt,
&pending,
std::time::Duration::from_millis(1),
),
Err(ReceiptStoreError::Unsupported(_))
));
assert_eq!(store.append_calls.load(Ordering::SeqCst), 0);
}
#[test]
fn timed_atomic_default_never_calls_the_legacy_unbounded_projection() {
let store = LegacyAtomicOnlyStore {
atomic_append_calls: AtomicUsize::new(0),
};
let receipt = signed_receipt();
let pending = PendingSettlementObservation {
next_visible_at_ms: 1,
};
assert_eq!(
store.atomic_receipt_projection(),
AtomicReceiptProjection::SettlementObservationV1
);
assert!(!store.supports_atomic_receipt_projection_with_timeout());
assert!(matches!(
store.append_chio_receipt_with_pending_observation_and_timeout(
&receipt,
&pending,
std::time::Duration::from_millis(1),
),
Err(ReceiptStoreError::Unsupported(_))
));
assert_eq!(store.atomic_append_calls.load(Ordering::SeqCst), 0);
}
#[test]
fn unsupported_durability_surfaces_fail_closed() -> Result<(), Box<dyn std::error::Error>> {
let store = AppendOnlyStore;
let checkpoint = crate::checkpoint::build_checkpoint(
1,
1,
1,
&[b"receipt".to_vec()],
&Keypair::generate(),
)?;
assert!(matches!(
store.receipts_canonical_bytes_range(1, 1),
Err(ReceiptStoreError::Conflict(message))
if message.contains("receipt canonical byte ranges are not supported")
));
assert!(matches!(
store.flush_receipt_writes(),
Err(ReceiptStoreError::Conflict(message))
if message.contains("receipt writer flush is not supported")
));
assert!(matches!(
store.flush_receipt_writes_with_timeout(std::time::Duration::from_millis(1)),
Err(ReceiptStoreError::Conflict(message))
if message.contains("receipt writer flush is not supported")
));
assert!(matches!(
store.receipt_store_health(),
Err(ReceiptStoreError::Conflict(message))
if message.contains("receipt store health is not supported")
));
assert!(matches!(
store.latest_committed_entry_seq(),
Err(ReceiptStoreError::Conflict(message))
if message.contains("receipt committed sequence reporting is not supported")
));
assert!(matches!(
store.latest_checkpointed_entry_seq(),
Err(ReceiptStoreError::Conflict(message))
if message.contains("receipt checkpoint sequence reporting is not supported")
));
assert!(matches!(
store.next_checkpoint_range(1),
Err(ReceiptStoreError::Conflict(message))
if message.contains("receipt checkpoint ranges are not supported")
));
assert!(matches!(
store.receipt_checkpoint_status(Some(1)),
Err(ReceiptStoreError::Conflict(message))
if message.contains("receipt checkpoint status is not supported")
));
assert!(matches!(
store.store_checkpoint(&checkpoint),
Err(ReceiptStoreError::Conflict(message))
if message.contains("receipt checkpoint storage is not supported")
));
assert!(matches!(
store.rotate_receipts(&RetentionConfig::default()),
Err(ReceiptStoreError::Conflict(message))
if message.contains("receipt retention is not supported")
));
Ok(())
}
#[derive(Default)]
struct FailingRetentionStore {
retention_error: std::sync::Mutex<Option<String>>,
rotations: std::sync::atomic::AtomicU64,
}
impl ReceiptStore for FailingRetentionStore {
fn append_chio_receipt(&self, _receipt: &ChioReceipt) -> Result<(), ReceiptStoreError> {
Ok(())
}
fn append_child_receipt(
&self,
_receipt: &ChildRequestReceipt,
) -> Result<(), ReceiptStoreError> {
Ok(())
}
fn supports_retention(&self) -> bool {
true
}
fn rotate_receipts(&self, _config: &RetentionConfig) -> Result<u64, ReceiptStoreError> {
self.rotations
.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
Err(ReceiptStoreError::Conflict(
"archive path is unwritable".to_string(),
))
}
fn record_retention_rotation_outcome(&self, failure: Option<&str>) {
if let Ok(mut guard) = self.retention_error.lock() {
*guard = failure.map(ToString::to_string);
}
}
fn receipt_store_health(&self) -> Result<ReceiptStoreHealthReport, ReceiptStoreError> {
let retention_error = self.retention_error.lock().ok().and_then(|g| g.clone());
Ok(ReceiptStoreHealthReport {
healthy: retention_error.is_none(),
retention_error,
..ReceiptStoreHealthReport::default()
})
}
}
#[test]
fn background_retention_failure_surfaces_in_health() {
let store = std::sync::Arc::new(FailingRetentionStore::default());
assert!(store.receipt_store_health().expect("health report").healthy);
let config = RetentionConfig {
check_interval_secs: 1,
..RetentionConfig::default()
};
let handle = RetentionMaintenanceHandle::spawn(store.clone(), config);
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
while std::time::Instant::now() < deadline
&& store
.receipt_store_health()
.expect("health report")
.retention_error
.is_none()
{
std::thread::sleep(std::time::Duration::from_millis(50));
}
let report = store.receipt_store_health().expect("health report");
drop(handle);
assert!(
store.rotations.load(std::sync::atomic::Ordering::SeqCst) > 0,
"the maintenance worker never attempted a rotation"
);
assert!(
!report.healthy,
"a persistently failing background rotation must mark the store unhealthy"
);
let message = report
.retention_error
.expect("the background rotation failure must surface in health");
assert!(
message.contains("archive path is unwritable"),
"unexpected retention error: {message}"
);
}
}
#[derive(Debug, Clone)]
pub struct StoredToolReceipt {
pub seq: u64,
pub receipt: ChioReceipt,
}
#[derive(Debug, Clone)]
pub struct StoredChildReceipt {
pub seq: u64,
pub receipt: ChildRequestReceipt,
}
#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct ReceiptLineageVerification {
pub receipt_id: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub request_id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub session_id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub session_anchor_id: Option<String>,
pub session_anchor_verified: bool,
pub parent_request_verified: bool,
pub parent_receipt_verified: bool,
pub replay_protected: bool,
}
impl ReceiptLineageVerification {
#[must_use]
pub fn delegated_call_chain_bound(&self) -> bool {
self.parent_receipt_verified
|| (self.session_anchor_verified && self.parent_request_verified)
}
}
#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct ReceiptLineageStatementLink {
#[serde(default, skip_serializing_if = "Option::is_none")]
pub statement_id: Option<String>,
pub child_receipt_id: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub child_request_id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub parent_receipt_id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub parent_request_id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub session_id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub session_anchor_id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub chain_id: Option<String>,
pub recorded_at: u64,
}
#[derive(Debug, Clone)]
pub struct FederatedEvidenceShareImport {
pub share_id: String,
pub manifest_hash: String,
pub exported_at: u64,
pub issuer: String,
pub partner: String,
pub signer_public_key: String,
pub require_proofs: bool,
pub query_json: String,
pub tool_receipts: Vec<StoredToolReceipt>,
pub capability_lineage: Vec<CapabilitySnapshot>,
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct FederatedEvidenceShareSummary {
pub share_id: String,
pub manifest_hash: String,
pub imported_at: u64,
pub exported_at: u64,
pub issuer: String,
pub partner: String,
pub signer_public_key: String,
pub require_proofs: bool,
pub tool_receipts: u64,
pub capability_lineage: u64,
}