use super::*;
use chio_core::capability::GovernedProvenanceEvidenceClass;
use chio_kernel::evidence_export::EvidenceLineageReferences;
use chio_kernel::operator_report::GovernedTransactionDiagnostics;
#[derive(Debug, Clone)]
struct GovernedTransactionProjection {
strong: GovernedTransactionReceiptMetadata,
diagnostics: Option<GovernedTransactionDiagnostics>,
}
#[derive(Debug, Clone, Default)]
struct PersistedLineageReferences {
statement_id: Option<String>,
session_anchor_id: Option<String>,
}
fn extract_governed_transaction_diagnostics(
receipt: &ChioReceipt,
) -> Option<GovernedTransactionDiagnostics> {
let diagnostics = receipt
.metadata
.as_ref()
.and_then(|metadata| metadata.get("governed_transaction_diagnostics"))
.cloned()
.and_then(|value| serde_json::from_value::<GovernedTransactionDiagnostics>(value).ok())
.unwrap_or_default();
(!diagnostics.is_empty()).then_some(diagnostics)
}
fn sanitize_governed_transaction_projection(
receipt: &ChioReceipt,
governed: &GovernedTransactionReceiptMetadata,
) -> GovernedTransactionProjection {
let mut strong = governed.clone();
let mut diagnostics = extract_governed_transaction_diagnostics(receipt).unwrap_or_default();
if let Some(call_chain) = strong
.call_chain
.clone()
.filter(|call_chain| call_chain.evidence_class == GovernedProvenanceEvidenceClass::Asserted)
{
strong.call_chain = None;
if diagnostics.asserted_call_chain.is_none() {
diagnostics.asserted_call_chain = Some(call_chain);
}
}
if diagnostics.lineage_references.session_anchor_id.is_none() {
diagnostics.lineage_references.session_anchor_id = strong
.call_chain
.as_ref()
.and_then(|call_chain| call_chain.session_anchor_id.clone())
.or_else(|| {
diagnostics
.asserted_call_chain
.as_ref()
.and_then(|call_chain| call_chain.session_anchor_id.clone())
});
}
if diagnostics
.lineage_references
.receipt_lineage_statement_id
.is_none()
{
diagnostics.lineage_references.receipt_lineage_statement_id = strong
.call_chain
.as_ref()
.and_then(|call_chain| call_chain.receipt_lineage_statement_id.clone())
.or_else(|| {
diagnostics
.asserted_call_chain
.as_ref()
.and_then(|call_chain| call_chain.receipt_lineage_statement_id.clone())
});
}
GovernedTransactionProjection {
strong,
diagnostics: (!diagnostics.is_empty()).then_some(diagnostics),
}
}
fn apply_persisted_lineage_references(
projection: &mut GovernedTransactionProjection,
persisted: &PersistedLineageReferences,
) {
if persisted.statement_id.is_none() && persisted.session_anchor_id.is_none() {
return;
}
if let Some(call_chain) = projection.strong.call_chain.as_mut() {
if call_chain.session_anchor_id.is_none() {
call_chain.session_anchor_id = persisted.session_anchor_id.clone();
}
if call_chain.receipt_lineage_statement_id.is_none() {
call_chain.receipt_lineage_statement_id = persisted.statement_id.clone();
}
}
let diagnostics = projection
.diagnostics
.get_or_insert_with(GovernedTransactionDiagnostics::default);
if let Some(call_chain) = diagnostics.asserted_call_chain.as_mut() {
if call_chain.session_anchor_id.is_none() {
call_chain.session_anchor_id = persisted.session_anchor_id.clone();
}
if call_chain.receipt_lineage_statement_id.is_none() {
call_chain.receipt_lineage_statement_id = persisted.statement_id.clone();
}
}
if diagnostics.lineage_references.session_anchor_id.is_none() {
diagnostics.lineage_references.session_anchor_id = persisted.session_anchor_id.clone();
}
if diagnostics
.lineage_references
.receipt_lineage_statement_id
.is_none()
{
diagnostics.lineage_references.receipt_lineage_statement_id =
persisted.statement_id.clone();
}
if diagnostics.is_empty() {
projection.diagnostics = None;
}
}
fn call_chain_evidence_class(
projection: &GovernedTransactionProjection,
) -> Option<GovernedProvenanceEvidenceClass> {
projection
.strong
.call_chain
.as_ref()
.map(|call_chain| call_chain.evidence_class)
.or_else(|| {
projection
.diagnostics
.as_ref()
.and_then(|diagnostics| diagnostics.asserted_call_chain.as_ref())
.map(|call_chain| call_chain.evidence_class)
})
}
fn lineage_references_from_projection(
projection: &GovernedTransactionProjection,
) -> Option<&EvidenceLineageReferences> {
projection
.diagnostics
.as_ref()
.map(|diagnostics| &diagnostics.lineage_references)
.filter(|references| !references.is_empty())
}
fn authorization_transaction_context_from_projection(
projection: &GovernedTransactionProjection,
) -> chio_kernel::operator_report::GovernedAuthorizationTransactionContext {
let mut transaction_context =
authorization_transaction_context_from_governed_metadata(&projection.strong);
if transaction_context.call_chain.is_none() {
transaction_context.call_chain = projection
.diagnostics
.as_ref()
.and_then(|diagnostics| diagnostics.asserted_call_chain.clone());
}
transaction_context
}
impl SqliteReceiptStore {
pub fn query_receipt_analytics(
&self,
query: &ReceiptAnalyticsQuery,
) -> Result<ReceiptAnalyticsResponse, ReceiptStoreError> {
let group_limit = query
.group_limit
.unwrap_or(50)
.clamp(1, MAX_ANALYTICS_GROUP_LIMIT);
let time_bucket = query.time_bucket.unwrap_or(AnalyticsTimeBucket::Day);
let bucket_width = time_bucket.width_secs() as i64;
let capability_id = query.capability_id.as_deref();
let tool_server = query.tool_server.as_deref();
let tool_name = query.tool_name.as_deref();
let since = query.since.map(|value| value as i64);
let until = query.until.map(|value| value as i64);
let agent_subject = query.agent_subject.as_deref();
let summary_sql = r#"
SELECT
COUNT(*) AS total_receipts,
COALESCE(SUM(CASE WHEN r.decision_kind = 'allow' THEN 1 ELSE 0 END), 0) AS allow_count,
COALESCE(SUM(CASE WHEN r.decision_kind = 'deny' THEN 1 ELSE 0 END), 0) AS deny_count,
COALESCE(SUM(CASE WHEN r.decision_kind = 'cancelled' THEN 1 ELSE 0 END), 0) AS cancelled_count,
COALESCE(SUM(CASE WHEN r.decision_kind = 'incomplete' THEN 1 ELSE 0 END), 0) AS incomplete_count,
COALESCE(SUM(CAST(COALESCE(json_extract(r.raw_json, '$.metadata.financial.cost_charged'), 0) AS INTEGER)), 0) AS total_cost_charged,
COALESCE(SUM(CAST(COALESCE(json_extract(r.raw_json, '$.metadata.financial.attempted_cost'), 0) AS INTEGER)), 0) AS total_attempted_cost
FROM chio_tool_receipts r
LEFT JOIN capability_lineage cl ON r.capability_id = cl.capability_id
WHERE (?1 IS NULL OR r.capability_id = ?1)
AND (?2 IS NULL OR r.tool_server = ?2)
AND (?3 IS NULL OR r.tool_name = ?3)
AND (?4 IS NULL OR r.timestamp >= ?4)
AND (?5 IS NULL OR r.timestamp <= ?5)
AND (?6 IS NULL OR COALESCE(r.subject_key, cl.subject_key) = ?6)
"#;
let summary = self.connection()?.query_row(
summary_sql,
params![
capability_id,
tool_server,
tool_name,
since,
until,
agent_subject
],
|row| {
Ok(ReceiptAnalyticsMetrics::from_raw(
row.get::<_, i64>(0)?.max(0) as u64,
row.get::<_, i64>(1)?.max(0) as u64,
row.get::<_, i64>(2)?.max(0) as u64,
row.get::<_, i64>(3)?.max(0) as u64,
row.get::<_, i64>(4)?.max(0) as u64,
row.get::<_, i64>(5)?.max(0) as u64,
row.get::<_, i64>(6)?.max(0) as u64,
))
},
)?;
let by_agent_sql = r#"
SELECT
COALESCE(r.subject_key, cl.subject_key) AS subject_key,
COUNT(*) AS total_receipts,
COALESCE(SUM(CASE WHEN r.decision_kind = 'allow' THEN 1 ELSE 0 END), 0) AS allow_count,
COALESCE(SUM(CASE WHEN r.decision_kind = 'deny' THEN 1 ELSE 0 END), 0) AS deny_count,
COALESCE(SUM(CASE WHEN r.decision_kind = 'cancelled' THEN 1 ELSE 0 END), 0) AS cancelled_count,
COALESCE(SUM(CASE WHEN r.decision_kind = 'incomplete' THEN 1 ELSE 0 END), 0) AS incomplete_count,
COALESCE(SUM(CAST(COALESCE(json_extract(r.raw_json, '$.metadata.financial.cost_charged'), 0) AS INTEGER)), 0) AS total_cost_charged,
COALESCE(SUM(CAST(COALESCE(json_extract(r.raw_json, '$.metadata.financial.attempted_cost'), 0) AS INTEGER)), 0) AS total_attempted_cost
FROM chio_tool_receipts r
LEFT JOIN capability_lineage cl ON r.capability_id = cl.capability_id
WHERE (?1 IS NULL OR r.capability_id = ?1)
AND (?2 IS NULL OR r.tool_server = ?2)
AND (?3 IS NULL OR r.tool_name = ?3)
AND (?4 IS NULL OR r.timestamp >= ?4)
AND (?5 IS NULL OR r.timestamp <= ?5)
AND (?6 IS NULL OR COALESCE(r.subject_key, cl.subject_key) = ?6)
AND COALESCE(r.subject_key, cl.subject_key) IS NOT NULL
GROUP BY COALESCE(r.subject_key, cl.subject_key)
ORDER BY total_receipts DESC, subject_key ASC
LIMIT ?7
"#;
let by_agent = self
.connection()?
.prepare(by_agent_sql)?
.query_map(
params![
capability_id,
tool_server,
tool_name,
since,
until,
agent_subject,
group_limit as i64
],
|row| {
Ok(AgentAnalyticsRow {
subject_key: row.get(0)?,
metrics: ReceiptAnalyticsMetrics::from_raw(
row.get::<_, i64>(1)?.max(0) as u64,
row.get::<_, i64>(2)?.max(0) as u64,
row.get::<_, i64>(3)?.max(0) as u64,
row.get::<_, i64>(4)?.max(0) as u64,
row.get::<_, i64>(5)?.max(0) as u64,
row.get::<_, i64>(6)?.max(0) as u64,
row.get::<_, i64>(7)?.max(0) as u64,
),
})
},
)?
.collect::<Result<Vec<_>, _>>()?;
let by_tool_sql = r#"
SELECT
r.tool_server,
r.tool_name,
COUNT(*) AS total_receipts,
COALESCE(SUM(CASE WHEN r.decision_kind = 'allow' THEN 1 ELSE 0 END), 0) AS allow_count,
COALESCE(SUM(CASE WHEN r.decision_kind = 'deny' THEN 1 ELSE 0 END), 0) AS deny_count,
COALESCE(SUM(CASE WHEN r.decision_kind = 'cancelled' THEN 1 ELSE 0 END), 0) AS cancelled_count,
COALESCE(SUM(CASE WHEN r.decision_kind = 'incomplete' THEN 1 ELSE 0 END), 0) AS incomplete_count,
COALESCE(SUM(CAST(COALESCE(json_extract(r.raw_json, '$.metadata.financial.cost_charged'), 0) AS INTEGER)), 0) AS total_cost_charged,
COALESCE(SUM(CAST(COALESCE(json_extract(r.raw_json, '$.metadata.financial.attempted_cost'), 0) AS INTEGER)), 0) AS total_attempted_cost
FROM chio_tool_receipts r
LEFT JOIN capability_lineage cl ON r.capability_id = cl.capability_id
WHERE (?1 IS NULL OR r.capability_id = ?1)
AND (?2 IS NULL OR r.tool_server = ?2)
AND (?3 IS NULL OR r.tool_name = ?3)
AND (?4 IS NULL OR r.timestamp >= ?4)
AND (?5 IS NULL OR r.timestamp <= ?5)
AND (?6 IS NULL OR COALESCE(r.subject_key, cl.subject_key) = ?6)
GROUP BY r.tool_server, r.tool_name
ORDER BY total_receipts DESC, r.tool_server ASC, r.tool_name ASC
LIMIT ?7
"#;
let by_tool = self
.connection()?
.prepare(by_tool_sql)?
.query_map(
params![
capability_id,
tool_server,
tool_name,
since,
until,
agent_subject,
group_limit as i64
],
|row| {
Ok(ToolAnalyticsRow {
tool_server: row.get(0)?,
tool_name: row.get(1)?,
metrics: ReceiptAnalyticsMetrics::from_raw(
row.get::<_, i64>(2)?.max(0) as u64,
row.get::<_, i64>(3)?.max(0) as u64,
row.get::<_, i64>(4)?.max(0) as u64,
row.get::<_, i64>(5)?.max(0) as u64,
row.get::<_, i64>(6)?.max(0) as u64,
row.get::<_, i64>(7)?.max(0) as u64,
row.get::<_, i64>(8)?.max(0) as u64,
),
})
},
)?
.collect::<Result<Vec<_>, _>>()?;
let by_time_sql = r#"
SELECT
CAST((r.timestamp / ?7) * ?7 AS INTEGER) AS bucket_start,
COUNT(*) AS total_receipts,
COALESCE(SUM(CASE WHEN r.decision_kind = 'allow' THEN 1 ELSE 0 END), 0) AS allow_count,
COALESCE(SUM(CASE WHEN r.decision_kind = 'deny' THEN 1 ELSE 0 END), 0) AS deny_count,
COALESCE(SUM(CASE WHEN r.decision_kind = 'cancelled' THEN 1 ELSE 0 END), 0) AS cancelled_count,
COALESCE(SUM(CASE WHEN r.decision_kind = 'incomplete' THEN 1 ELSE 0 END), 0) AS incomplete_count,
COALESCE(SUM(CAST(COALESCE(json_extract(r.raw_json, '$.metadata.financial.cost_charged'), 0) AS INTEGER)), 0) AS total_cost_charged,
COALESCE(SUM(CAST(COALESCE(json_extract(r.raw_json, '$.metadata.financial.attempted_cost'), 0) AS INTEGER)), 0) AS total_attempted_cost
FROM chio_tool_receipts r
LEFT JOIN capability_lineage cl ON r.capability_id = cl.capability_id
WHERE (?1 IS NULL OR r.capability_id = ?1)
AND (?2 IS NULL OR r.tool_server = ?2)
AND (?3 IS NULL OR r.tool_name = ?3)
AND (?4 IS NULL OR r.timestamp >= ?4)
AND (?5 IS NULL OR r.timestamp <= ?5)
AND (?6 IS NULL OR COALESCE(r.subject_key, cl.subject_key) = ?6)
GROUP BY bucket_start
ORDER BY bucket_start ASC
LIMIT ?8
"#;
let by_time = self
.connection()?
.prepare(by_time_sql)?
.query_map(
params![
capability_id,
tool_server,
tool_name,
since,
until,
agent_subject,
bucket_width,
group_limit as i64
],
|row| {
let bucket_start = row.get::<_, i64>(0)?.max(0) as u64;
Ok(TimeAnalyticsRow {
bucket_start,
bucket_end: bucket_start
.saturating_add(bucket_width.max(1) as u64)
.saturating_sub(1),
metrics: ReceiptAnalyticsMetrics::from_raw(
row.get::<_, i64>(1)?.max(0) as u64,
row.get::<_, i64>(2)?.max(0) as u64,
row.get::<_, i64>(3)?.max(0) as u64,
row.get::<_, i64>(4)?.max(0) as u64,
row.get::<_, i64>(5)?.max(0) as u64,
row.get::<_, i64>(6)?.max(0) as u64,
row.get::<_, i64>(7)?.max(0) as u64,
),
})
},
)?
.collect::<Result<Vec<_>, _>>()?;
Ok(ReceiptAnalyticsResponse {
summary,
by_agent,
by_tool,
by_time,
})
}
pub fn query_cost_attribution_report(
&self,
query: &CostAttributionQuery,
) -> Result<CostAttributionReport, ReceiptStoreError> {
let limit = query
.limit
.unwrap_or(100)
.clamp(1, MAX_COST_ATTRIBUTION_LIMIT);
let capability_id = query.capability_id.as_deref();
let tool_server = query.tool_server.as_deref();
let tool_name = query.tool_name.as_deref();
let since = query.since.map(|value| value as i64);
let until = query.until.map(|value| value as i64);
let agent_subject = query.agent_subject.as_deref();
let count_sql = r#"
SELECT COUNT(*)
FROM chio_tool_receipts r
LEFT JOIN capability_lineage cl ON r.capability_id = cl.capability_id
WHERE (?1 IS NULL OR r.capability_id = ?1)
AND (?2 IS NULL OR r.tool_server = ?2)
AND (?3 IS NULL OR r.tool_name = ?3)
AND (?4 IS NULL OR r.timestamp >= ?4)
AND (?5 IS NULL OR r.timestamp <= ?5)
AND (?6 IS NULL OR COALESCE(r.subject_key, cl.subject_key) = ?6)
AND json_type(r.raw_json, '$.metadata.financial') = 'object'
"#;
let matching_receipts = self
.connection()?
.query_row(
count_sql,
params![
capability_id,
tool_server,
tool_name,
since,
until,
agent_subject
],
|row| row.get::<_, i64>(0),
)
.map(|value| value.max(0) as u64)?;
let data_sql = r#"
SELECT r.seq, r.raw_json
FROM chio_tool_receipts r
LEFT JOIN capability_lineage cl ON r.capability_id = cl.capability_id
WHERE (?1 IS NULL OR r.capability_id = ?1)
AND (?2 IS NULL OR r.tool_server = ?2)
AND (?3 IS NULL OR r.tool_name = ?3)
AND (?4 IS NULL OR r.timestamp >= ?4)
AND (?5 IS NULL OR r.timestamp <= ?5)
AND (?6 IS NULL OR COALESCE(r.subject_key, cl.subject_key) = ?6)
AND json_type(r.raw_json, '$.metadata.financial') = 'object'
ORDER BY r.seq ASC
"#;
let rows = self
.connection()?
.prepare(data_sql)?
.query_map(
params![
capability_id,
tool_server,
tool_name,
since,
until,
agent_subject
],
|row| {
Ok((
row.get::<_, i64>(0)?.max(0) as u64,
row.get::<_, String>(1)?,
))
},
)?
.collect::<Result<Vec<_>, _>>()?;
let mut receipts = Vec::with_capacity(rows.len().min(limit));
let mut by_root = BTreeMap::<String, RootAggregate>::new();
let mut by_leaf = BTreeMap::<(String, String), LeafAggregate>::new();
let mut distinct_roots = BTreeSet::new();
let mut distinct_leaves = BTreeSet::new();
let mut total_cost_charged = 0_u64;
let mut total_attempted_cost = 0_u64;
let mut max_delegation_depth = 0_u64;
let mut lineage_gap_count = 0_u64;
for (seq, raw_json) in rows {
let receipt =
decode_verified_chio_receipt(&raw_json, "persisted tool receipt", Some(seq))?;
let Some(financial) = extract_financial_metadata(&receipt) else {
continue;
};
let attribution = extract_receipt_attribution(&receipt);
let chain_snapshots = self
.get_combined_delegation_chain(&receipt.capability_id)
.unwrap_or_default();
let lineage_complete = chain_is_complete(&receipt.capability_id, &chain_snapshots);
if !lineage_complete {
lineage_gap_count = lineage_gap_count.saturating_add(1);
}
let chain = chain_snapshots
.iter()
.map(|snapshot| CostAttributionChainHop {
capability_id: snapshot.capability_id.clone(),
subject_key: snapshot.subject_key.clone(),
issuer_key: snapshot.issuer_key.clone(),
delegation_depth: snapshot.delegation_depth,
parent_capability_id: snapshot.parent_capability_id.clone(),
})
.collect::<Vec<_>>();
let root_subject_key = chain_snapshots
.first()
.map(|snapshot| snapshot.subject_key.clone())
.or_else(|| Some(financial.root_budget_holder.clone()));
let leaf_subject_key = attribution.subject_key.clone().or_else(|| {
chain_snapshots
.last()
.map(|snapshot| snapshot.subject_key.clone())
});
let attempted_cost = financial.attempted_cost.unwrap_or(0);
let decision = decision_kind(&receipt.decision).to_string();
total_cost_charged = total_cost_charged.saturating_add(financial.cost_charged);
total_attempted_cost = total_attempted_cost.saturating_add(attempted_cost);
max_delegation_depth = max_delegation_depth.max(financial.delegation_depth as u64);
if let Some(root_key) = root_subject_key.clone() {
distinct_roots.insert(root_key.clone());
let root_entry = by_root.entry(root_key.clone()).or_default();
root_entry.receipt_count = root_entry.receipt_count.saturating_add(1);
root_entry.total_cost_charged = root_entry
.total_cost_charged
.saturating_add(financial.cost_charged);
root_entry.total_attempted_cost = root_entry
.total_attempted_cost
.saturating_add(attempted_cost);
root_entry.max_delegation_depth = root_entry
.max_delegation_depth
.max(financial.delegation_depth as u64);
if let Some(leaf_key) = leaf_subject_key.clone() {
root_entry.leaf_subjects.insert(leaf_key.clone());
let leaf_entry = by_leaf.entry((root_key, leaf_key)).or_default();
leaf_entry.receipt_count = leaf_entry.receipt_count.saturating_add(1);
leaf_entry.total_cost_charged = leaf_entry
.total_cost_charged
.saturating_add(financial.cost_charged);
leaf_entry.total_attempted_cost = leaf_entry
.total_attempted_cost
.saturating_add(attempted_cost);
leaf_entry.max_delegation_depth = leaf_entry
.max_delegation_depth
.max(financial.delegation_depth as u64);
}
}
if let Some(leaf_key) = leaf_subject_key.clone() {
distinct_leaves.insert(leaf_key);
}
if receipts.len() < limit {
receipts.push(CostAttributionReceiptRow {
seq,
receipt_id: receipt.id.clone(),
timestamp: receipt.timestamp,
capability_id: receipt.capability_id.clone(),
tool_server: receipt.tool_server.clone(),
tool_name: receipt.tool_name.clone(),
decision_kind: decision,
root_subject_key,
leaf_subject_key,
grant_index: Some(financial.grant_index),
delegation_depth: financial.delegation_depth as u64,
cost_charged: financial.cost_charged,
attempted_cost: financial.attempted_cost,
currency: financial.currency.clone(),
budget_total: Some(financial.budget_total),
budget_remaining: Some(financial.budget_remaining),
settlement_status: Some(financial.settlement_status),
payment_reference: financial.payment_reference.clone(),
budget_authority: receipt.financial_budget_authority_metadata(),
lineage_complete,
chain,
});
}
}
let mut by_root = by_root
.into_iter()
.map(|(root_subject_key, aggregate)| RootCostAttributionRow {
root_subject_key,
receipt_count: aggregate.receipt_count,
total_cost_charged: aggregate.total_cost_charged,
total_attempted_cost: aggregate.total_attempted_cost,
distinct_leaf_subjects: aggregate.leaf_subjects.len() as u64,
max_delegation_depth: aggregate.max_delegation_depth,
})
.collect::<Vec<_>>();
by_root.sort_by(|left, right| {
right
.total_cost_charged
.cmp(&left.total_cost_charged)
.then_with(|| right.receipt_count.cmp(&left.receipt_count))
.then_with(|| left.root_subject_key.cmp(&right.root_subject_key))
});
let mut by_leaf = by_leaf
.into_iter()
.map(
|((root_subject_key, leaf_subject_key), aggregate)| LeafCostAttributionRow {
root_subject_key,
leaf_subject_key,
receipt_count: aggregate.receipt_count,
total_cost_charged: aggregate.total_cost_charged,
total_attempted_cost: aggregate.total_attempted_cost,
max_delegation_depth: aggregate.max_delegation_depth,
},
)
.collect::<Vec<_>>();
by_leaf.sort_by(|left, right| {
right
.total_cost_charged
.cmp(&left.total_cost_charged)
.then_with(|| right.receipt_count.cmp(&left.receipt_count))
.then_with(|| left.root_subject_key.cmp(&right.root_subject_key))
.then_with(|| left.leaf_subject_key.cmp(&right.leaf_subject_key))
});
Ok(CostAttributionReport {
summary: CostAttributionSummary {
matching_receipts,
returned_receipts: receipts.len() as u64,
total_cost_charged,
total_attempted_cost,
max_delegation_depth,
distinct_root_subjects: distinct_roots.len() as u64,
distinct_leaf_subjects: distinct_leaves.len() as u64,
lineage_gap_count,
truncated: matching_receipts > receipts.len() as u64,
},
by_root,
by_leaf,
receipts,
})
}
pub fn query_shared_evidence_report(
&self,
query: &SharedEvidenceQuery,
) -> Result<SharedEvidenceReferenceReport, ReceiptStoreError> {
let limit = query.limit_or_default();
let capability_id = query.capability_id.as_deref();
let tool_server = query.tool_server.as_deref();
let tool_name = query.tool_name.as_deref();
let since = query.since.map(|value| value as i64);
let until = query.until.map(|value| value as i64);
let agent_subject = query.agent_subject.as_deref();
let issuer = query.issuer.as_deref();
let partner = query.partner.as_deref();
let rows = self
.connection()?
.prepare(
r#"
SELECT r.receipt_id, r.timestamp, r.capability_id, r.decision_kind
FROM chio_tool_receipts r
LEFT JOIN capability_lineage cl ON r.capability_id = cl.capability_id
WHERE (?1 IS NULL OR r.capability_id = ?1)
AND (?2 IS NULL OR r.tool_server = ?2)
AND (?3 IS NULL OR r.tool_name = ?3)
AND (?4 IS NULL OR r.timestamp >= ?4)
AND (?5 IS NULL OR r.timestamp <= ?5)
AND (?6 IS NULL OR COALESCE(r.subject_key, cl.subject_key) = ?6)
ORDER BY r.seq ASC
"#,
)?
.query_map(
params![
capability_id,
tool_server,
tool_name,
since,
until,
agent_subject
],
|row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, i64>(1)?.max(0) as u64,
row.get::<_, String>(2)?,
row.get::<_, String>(3)?,
))
},
)?
.collect::<Result<Vec<_>, _>>()?;
let mut share_cache = BTreeMap::<String, Option<FederatedEvidenceShareSummary>>::new();
let mut references = BTreeMap::<(String, String), SharedEvidenceReferenceRow>::new();
let mut matched_local_receipts = BTreeSet::<String>::new();
for (receipt_id, timestamp, local_capability_id, decision) in rows {
let chain = self.get_combined_delegation_chain(&local_capability_id)?;
if chain.is_empty() {
continue;
}
let mut matched_this_receipt = false;
for (index, snapshot) in chain.iter().enumerate() {
let share = match share_cache.get(&snapshot.capability_id) {
Some(cached) => cached.clone(),
None => {
let loaded = self
.get_federated_share_for_capability(&snapshot.capability_id)?
.map(|(share, _)| share);
share_cache.insert(snapshot.capability_id.clone(), loaded.clone());
loaded
}
};
let Some(share) = share else {
continue;
};
if issuer.is_some_and(|expected| share.issuer != expected) {
continue;
}
if partner.is_some_and(|expected| share.partner != expected) {
continue;
}
let local_anchor_capability_id =
chain.iter().skip(index + 1).find_map(|candidate| {
match share_cache.get(&candidate.capability_id) {
Some(Some(_)) => None,
Some(None) => Some(candidate.capability_id.clone()),
None => {
let loaded = self
.get_federated_share_for_capability(&candidate.capability_id)
.ok()
.and_then(|value| value.map(|(share, _)| share));
share_cache.insert(candidate.capability_id.clone(), loaded.clone());
if loaded.is_some() {
None
} else {
Some(candidate.capability_id.clone())
}
}
}
});
let key = (share.share_id.clone(), snapshot.capability_id.clone());
let entry = references
.entry(key)
.or_insert_with(|| SharedEvidenceReferenceRow {
share: share.clone(),
capability_id: snapshot.capability_id.clone(),
subject_key: snapshot.subject_key.clone(),
issuer_key: snapshot.issuer_key.clone(),
delegation_depth: snapshot.delegation_depth,
parent_capability_id: snapshot.parent_capability_id.clone(),
local_anchor_capability_id: local_anchor_capability_id.clone(),
matched_local_receipts: 0,
allow_count: 0,
deny_count: 0,
cancelled_count: 0,
incomplete_count: 0,
first_seen: Some(timestamp),
last_seen: Some(timestamp),
});
entry.local_anchor_capability_id = entry
.local_anchor_capability_id
.clone()
.or(local_anchor_capability_id);
entry.matched_local_receipts = entry.matched_local_receipts.saturating_add(1);
entry.first_seen = Some(
entry
.first_seen
.map_or(timestamp, |value| value.min(timestamp)),
);
entry.last_seen = Some(
entry
.last_seen
.map_or(timestamp, |value| value.max(timestamp)),
);
match decision.as_str() {
"allow" => entry.allow_count = entry.allow_count.saturating_add(1),
"deny" => entry.deny_count = entry.deny_count.saturating_add(1),
"cancelled" => entry.cancelled_count = entry.cancelled_count.saturating_add(1),
_ => entry.incomplete_count = entry.incomplete_count.saturating_add(1),
}
matched_this_receipt = true;
}
if matched_this_receipt {
matched_local_receipts.insert(receipt_id);
}
}
let mut returned_references = references.into_values().collect::<Vec<_>>();
returned_references.sort_by(|left, right| {
right
.matched_local_receipts
.cmp(&left.matched_local_receipts)
.then_with(|| right.last_seen.cmp(&left.last_seen))
.then_with(|| right.share.imported_at.cmp(&left.share.imported_at))
.then_with(|| left.share.share_id.cmp(&right.share.share_id))
.then_with(|| left.capability_id.cmp(&right.capability_id))
});
let mut distinct_shares = BTreeMap::<String, FederatedEvidenceShareSummary>::new();
let mut distinct_remote_subjects = BTreeSet::<String>::new();
for reference in &returned_references {
distinct_shares
.entry(reference.share.share_id.clone())
.or_insert_with(|| reference.share.clone());
distinct_remote_subjects.insert(reference.subject_key.clone());
}
let matching_references = returned_references.len() as u64;
let truncated = returned_references.len() > limit;
if truncated {
returned_references.truncate(limit);
}
Ok(SharedEvidenceReferenceReport {
summary: SharedEvidenceReferenceSummary {
matching_shares: distinct_shares.len() as u64,
matching_references,
matching_local_receipts: matched_local_receipts.len() as u64,
remote_tool_receipts: distinct_shares
.values()
.map(|share| share.tool_receipts)
.sum(),
remote_lineage_records: distinct_shares
.values()
.map(|share| share.capability_lineage)
.sum(),
distinct_remote_subjects: distinct_remote_subjects.len() as u64,
proof_required_shares: distinct_shares
.values()
.filter(|share| share.require_proofs)
.count() as u64,
truncated,
},
references: returned_references,
})
}
pub fn query_compliance_report(
&self,
query: &OperatorReportQuery,
) -> Result<ComplianceReport, ReceiptStoreError> {
let capability_id = query.capability_id.as_deref();
let tool_server = query.tool_server.as_deref();
let tool_name = query.tool_name.as_deref();
let since = query.since.map(|value| value as i64);
let until = query.until.map(|value| value as i64);
let agent_subject = query.agent_subject.as_deref();
let summary_sql = r#"
SELECT
COUNT(*) AS matching_receipts,
COALESCE(SUM(
CASE
WHEN EXISTS(
SELECT 1
FROM kernel_checkpoints kc
WHERE r.seq BETWEEN kc.batch_start_seq AND kc.batch_end_seq
) THEN 1
ELSE 0
END
), 0) AS evidence_ready_receipts,
COALESCE(SUM(CASE WHEN cl.capability_id IS NOT NULL THEN 1 ELSE 0 END), 0) AS lineage_covered_receipts,
COALESCE(SUM(
CASE
WHEN json_extract(r.raw_json, '$.metadata.financial.settlement_status') = 'pending' THEN 1
ELSE 0
END
), 0) AS pending_settlement_receipts,
COALESCE(SUM(
CASE
WHEN json_extract(r.raw_json, '$.metadata.financial.settlement_status') = 'failed' THEN 1
ELSE 0
END
), 0) AS failed_settlement_receipts
FROM chio_tool_receipts r
LEFT JOIN capability_lineage cl ON r.capability_id = cl.capability_id
WHERE (?1 IS NULL OR r.capability_id = ?1)
AND (?2 IS NULL OR r.tool_server = ?2)
AND (?3 IS NULL OR r.tool_name = ?3)
AND (?4 IS NULL OR r.timestamp >= ?4)
AND (?5 IS NULL OR r.timestamp <= ?5)
AND (?6 IS NULL OR COALESCE(r.subject_key, cl.subject_key) = ?6)
"#;
let (
matching_receipts,
evidence_ready_receipts,
lineage_covered_receipts,
pending_settlement_receipts,
failed_settlement_receipts,
) = self.connection()?.query_row(
summary_sql,
params![
capability_id,
tool_server,
tool_name,
since,
until,
agent_subject
],
|row| {
Ok((
row.get::<_, i64>(0)?.max(0) as u64,
row.get::<_, i64>(1)?.max(0) as u64,
row.get::<_, i64>(2)?.max(0) as u64,
row.get::<_, i64>(3)?.max(0) as u64,
row.get::<_, i64>(4)?.max(0) as u64,
))
},
)?;
let uncheckpointed_receipts = matching_receipts.saturating_sub(evidence_ready_receipts);
let lineage_gap_receipts = matching_receipts.saturating_sub(lineage_covered_receipts);
let export_query = query.to_evidence_export_query();
Ok(ComplianceReport {
matching_receipts,
evidence_ready_receipts,
uncheckpointed_receipts,
checkpoint_coverage_rate: ratio_option(evidence_ready_receipts, matching_receipts),
lineage_covered_receipts,
lineage_gap_receipts,
lineage_coverage_rate: ratio_option(lineage_covered_receipts, matching_receipts),
pending_settlement_receipts,
failed_settlement_receipts,
direct_evidence_export_supported: query.direct_evidence_export_supported(),
child_receipt_scope: export_query.child_receipt_scope(),
proofs_complete: uncheckpointed_receipts == 0,
export_query: export_query.clone(),
export_scope_note: compliance_export_scope_note(query, &export_query),
})
}
pub fn upsert_settlement_reconciliation(
&self,
receipt_id: &str,
reconciliation_state: SettlementReconciliationState,
note: Option<&str>,
) -> Result<i64, ReceiptStoreError> {
let exists = self
.connection()?
.query_row(
"SELECT 1 FROM chio_tool_receipts WHERE receipt_id = ?1",
params![receipt_id],
|row| row.get::<_, i64>(0),
)
.optional()?;
if exists.is_none() {
return Err(ReceiptStoreError::NotFound(format!(
"receipt {receipt_id} does not exist"
)));
}
let updated_at = unix_timestamp_now_i64();
self.connection()?.execute(
r#"
INSERT INTO settlement_reconciliations (
receipt_id,
reconciliation_state,
note,
updated_at
) VALUES (?1, ?2, ?3, ?4)
ON CONFLICT(receipt_id) DO UPDATE SET
reconciliation_state = excluded.reconciliation_state,
note = excluded.note,
updated_at = excluded.updated_at
"#,
params![
receipt_id,
settlement_reconciliation_state_text(reconciliation_state),
note,
updated_at
],
)?;
Ok(updated_at)
}
pub fn upsert_metered_billing_reconciliation(
&self,
receipt_id: &str,
evidence: &MeteredBillingEvidenceRecord,
reconciliation_state: MeteredBillingReconciliationState,
note: Option<&str>,
) -> Result<i64, ReceiptStoreError> {
let (seq, raw_json) = self
.connection()?
.query_row(
"SELECT seq, raw_json FROM chio_tool_receipts WHERE receipt_id = ?1",
params![receipt_id],
|row| Ok((row.get::<_, i64>(0)?, row.get::<_, String>(1)?)),
)
.optional()?
.ok_or_else(|| {
ReceiptStoreError::NotFound(format!("receipt {receipt_id} does not exist"))
})?;
let receipt = decode_verified_chio_receipt(
&raw_json,
"persisted tool receipt",
Some(seq.max(0) as u64),
)?;
let governed = extract_governed_transaction_metadata(&receipt).ok_or_else(|| {
ReceiptStoreError::Conflict(format!(
"receipt {receipt_id} does not carry governed transaction metadata"
))
})?;
if governed.metered_billing.is_none() {
return Err(ReceiptStoreError::Conflict(format!(
"receipt {receipt_id} does not carry metered billing context"
)));
}
let existing_receipt = self
.connection()?
.query_row(
r#"
SELECT receipt_id
FROM metered_billing_reconciliations
WHERE adapter_kind = ?1 AND evidence_id = ?2
"#,
params![
&evidence.usage_evidence.evidence_kind,
&evidence.usage_evidence.evidence_id
],
|row| row.get::<_, String>(0),
)
.optional()?;
if let Some(existing_receipt) = existing_receipt {
if existing_receipt != receipt_id {
return Err(ReceiptStoreError::Conflict(format!(
"metered billing evidence {}/{} is already attached to receipt {}",
evidence.usage_evidence.evidence_kind,
evidence.usage_evidence.evidence_id,
existing_receipt
)));
}
}
let updated_at = unix_timestamp_now_i64();
self.connection()?.execute(
r#"
INSERT INTO metered_billing_reconciliations (
receipt_id,
adapter_kind,
evidence_id,
observed_units,
billed_cost_units,
billed_cost_currency,
evidence_sha256,
recorded_at,
reconciliation_state,
note,
updated_at
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11)
ON CONFLICT(receipt_id) DO UPDATE SET
adapter_kind = excluded.adapter_kind,
evidence_id = excluded.evidence_id,
observed_units = excluded.observed_units,
billed_cost_units = excluded.billed_cost_units,
billed_cost_currency = excluded.billed_cost_currency,
evidence_sha256 = excluded.evidence_sha256,
recorded_at = excluded.recorded_at,
reconciliation_state = excluded.reconciliation_state,
note = excluded.note,
updated_at = excluded.updated_at
"#,
params![
receipt_id,
&evidence.usage_evidence.evidence_kind,
&evidence.usage_evidence.evidence_id,
evidence.usage_evidence.observed_units as i64,
evidence.billed_cost.units as i64,
&evidence.billed_cost.currency,
evidence.usage_evidence.evidence_sha256.as_deref(),
evidence.recorded_at as i64,
metered_billing_reconciliation_state_text(reconciliation_state),
note,
updated_at
],
)?;
Ok(updated_at)
}
pub fn query_metered_billing_reconciliation_report(
&self,
query: &OperatorReportQuery,
) -> Result<MeteredBillingReconciliationReport, ReceiptStoreError> {
let capability_id = query.capability_id.as_deref();
let tool_server = query.tool_server.as_deref();
let tool_name = query.tool_name.as_deref();
let since = query.since.map(|value| value as i64);
let until = query.until.map(|value| value as i64);
let agent_subject = query.agent_subject.as_deref();
let row_limit = query.metered_limit_or_default();
let summary = self.query_metered_billing_summary(query)?;
let rows_sql = r#"
SELECT
r.seq,
r.raw_json,
COALESCE(r.subject_key, cl.subject_key),
mbr.adapter_kind,
mbr.evidence_id,
mbr.observed_units,
mbr.billed_cost_units,
mbr.billed_cost_currency,
mbr.evidence_sha256,
mbr.recorded_at,
COALESCE(mbr.reconciliation_state, 'open'),
mbr.note,
mbr.updated_at
FROM chio_tool_receipts r
LEFT JOIN capability_lineage cl ON r.capability_id = cl.capability_id
LEFT JOIN metered_billing_reconciliations mbr ON r.receipt_id = mbr.receipt_id
WHERE json_type(r.raw_json, '$.metadata.governed_transaction.metered_billing') = 'object'
AND (?1 IS NULL OR r.capability_id = ?1)
AND (?2 IS NULL OR r.tool_server = ?2)
AND (?3 IS NULL OR r.tool_name = ?3)
AND (?4 IS NULL OR r.timestamp >= ?4)
AND (?5 IS NULL OR r.timestamp <= ?5)
AND (?6 IS NULL OR COALESCE(r.subject_key, cl.subject_key) = ?6)
ORDER BY r.timestamp DESC, r.seq DESC
LIMIT ?7
"#;
let connection = self.connection()?;
let mut stmt = connection.prepare(rows_sql)?;
let rows = stmt.query_map(
params![
capability_id,
tool_server,
tool_name,
since,
until,
agent_subject,
row_limit as i64
],
|row| {
Ok((
row.get::<_, i64>(0)?,
row.get::<_, String>(1)?,
row.get::<_, Option<String>>(2)?,
row.get::<_, Option<String>>(3)?,
row.get::<_, Option<String>>(4)?,
row.get::<_, Option<i64>>(5)?,
row.get::<_, Option<i64>>(6)?,
row.get::<_, Option<String>>(7)?,
row.get::<_, Option<String>>(8)?,
row.get::<_, Option<i64>>(9)?,
row.get::<_, String>(10)?,
row.get::<_, Option<String>>(11)?,
row.get::<_, Option<i64>>(12)?,
))
},
)?;
let mut receipts = Vec::new();
for row in rows {
let (
seq,
raw_json,
subject_key,
adapter_kind,
evidence_id,
observed_units,
billed_cost_units,
billed_cost_currency,
evidence_sha256,
recorded_at,
reconciliation_state_text,
note,
updated_at,
) = row?;
let receipt = decode_verified_chio_receipt(
&raw_json,
"persisted tool receipt",
Some(seq.max(0) as u64),
)?;
let governed = extract_governed_transaction_metadata(&receipt).ok_or_else(|| {
ReceiptStoreError::Canonical(format!(
"receipt {} is missing governed transaction metadata",
receipt.id
))
})?;
let metered = governed.metered_billing.ok_or_else(|| {
ReceiptStoreError::Canonical(format!(
"receipt {} is missing metered billing metadata",
receipt.id
))
})?;
let financial = extract_financial_metadata(&receipt);
let evidence = metered_billing_evidence_record_from_columns(
adapter_kind,
evidence_id,
observed_units,
billed_cost_units,
billed_cost_currency,
evidence_sha256,
recorded_at,
);
let reconciliation_state =
parse_metered_billing_reconciliation_state(&reconciliation_state_text)?;
let analysis = analyze_metered_billing_reconciliation(
&metered,
financial.as_ref(),
evidence.as_ref(),
reconciliation_state,
);
let budget_authority = receipt.financial_budget_authority_metadata();
receipts.push(MeteredBillingReconciliationRow {
receipt_id: receipt.id,
timestamp: receipt.timestamp,
capability_id: receipt.capability_id,
subject_key,
tool_server: receipt.tool_server,
tool_name: receipt.tool_name,
settlement_mode: metered.settlement_mode,
provider: metered.quote.provider.clone(),
quote_id: metered.quote.quote_id.clone(),
billing_unit: metered.quote.billing_unit.clone(),
quoted_units: metered.quote.quoted_units,
quoted_cost: metered.quote.quoted_cost.clone(),
max_billed_units: metered.max_billed_units,
financial_cost_charged: financial.as_ref().map(|value| value.cost_charged),
financial_currency: financial.as_ref().map(|value| value.currency.clone()),
budget_authority,
evidence,
reconciliation_state,
action_required: analysis.action_required,
evidence_missing: analysis.evidence_missing,
exceeds_quoted_units: analysis.exceeds_quoted_units,
exceeds_max_billed_units: analysis.exceeds_max_billed_units,
exceeds_quoted_cost: analysis.exceeds_quoted_cost,
financial_mismatch: analysis.financial_mismatch,
note,
updated_at: updated_at.map(|value| value.max(0) as u64),
});
}
Ok(MeteredBillingReconciliationReport {
summary: MeteredBillingReconciliationSummary {
matching_receipts: summary.metered_receipts,
returned_receipts: receipts.len() as u64,
evidence_attached_receipts: summary.evidence_attached_receipts,
missing_evidence_receipts: summary.missing_evidence_receipts,
over_quoted_units_receipts: summary.over_quoted_units_receipts,
over_max_billed_units_receipts: summary.over_max_billed_units_receipts,
over_quoted_cost_receipts: summary.over_quoted_cost_receipts,
financial_mismatch_receipts: summary.financial_mismatch_receipts,
actionable_receipts: summary.actionable_receipts,
reconciled_receipts: summary.reconciled_receipts,
truncated: summary.metered_receipts > receipts.len() as u64,
},
receipts,
})
}
pub fn query_economic_receipt_projection_report(
&self,
query: &OperatorReportQuery,
) -> Result<EconomicReceiptProjectionReport, ReceiptStoreError> {
let capability_id = query.capability_id.as_deref();
let tool_server = query.tool_server.as_deref();
let tool_name = query.tool_name.as_deref();
let since = query.since.map(|value| value as i64);
let until = query.until.map(|value| value as i64);
let agent_subject = query.agent_subject.as_deref();
let row_limit = query.economic_limit_or_default();
let matching_receipts = self.connection()?.query_row(
r#"
SELECT COUNT(*)
FROM chio_tool_receipts r
LEFT JOIN capability_lineage cl ON r.capability_id = cl.capability_id
WHERE json_type(r.raw_json, '$.metadata.governed_transaction.economic_authorization') = 'object'
AND (?1 IS NULL OR r.capability_id = ?1)
AND (?2 IS NULL OR r.tool_server = ?2)
AND (?3 IS NULL OR r.tool_name = ?3)
AND (?4 IS NULL OR r.timestamp >= ?4)
AND (?5 IS NULL OR r.timestamp <= ?5)
AND (?6 IS NULL OR COALESCE(r.subject_key, cl.subject_key) = ?6)
"#,
params![
capability_id,
tool_server,
tool_name,
since,
until,
agent_subject
],
|row| row.get::<_, i64>(0),
)?
.max(0) as u64;
let rows_sql = r#"
SELECT
r.seq,
r.receipt_id,
r.timestamp,
r.capability_id,
COALESCE(r.subject_key, cl.subject_key),
r.tool_server,
r.tool_name,
COALESCE(sr.reconciliation_state, 'open'),
sr.note,
sr.updated_at,
COALESCE(mbr.reconciliation_state, 'open'),
mbr.note,
mbr.updated_at,
mbr.adapter_kind,
mbr.evidence_id,
mbr.observed_units,
mbr.billed_cost_units,
mbr.billed_cost_currency,
mbr.evidence_sha256,
mbr.recorded_at,
r.raw_json
FROM chio_tool_receipts r
LEFT JOIN capability_lineage cl ON r.capability_id = cl.capability_id
LEFT JOIN settlement_reconciliations sr ON r.receipt_id = sr.receipt_id
LEFT JOIN metered_billing_reconciliations mbr ON r.receipt_id = mbr.receipt_id
WHERE json_type(r.raw_json, '$.metadata.governed_transaction.economic_authorization') = 'object'
AND (?1 IS NULL OR r.capability_id = ?1)
AND (?2 IS NULL OR r.tool_server = ?2)
AND (?3 IS NULL OR r.tool_name = ?3)
AND (?4 IS NULL OR r.timestamp >= ?4)
AND (?5 IS NULL OR r.timestamp <= ?5)
AND (?6 IS NULL OR COALESCE(r.subject_key, cl.subject_key) = ?6)
ORDER BY r.timestamp DESC, r.seq DESC
LIMIT ?7
"#;
let connection = self.connection()?;
let mut stmt = connection.prepare(rows_sql)?;
let rows = stmt.query_map(
params![
capability_id,
tool_server,
tool_name,
since,
until,
agent_subject,
row_limit as i64
],
|row| {
Ok((
row.get::<_, i64>(0)?,
row.get::<_, String>(1)?,
row.get::<_, i64>(2)?,
row.get::<_, String>(3)?,
row.get::<_, Option<String>>(4)?,
row.get::<_, String>(5)?,
row.get::<_, String>(6)?,
row.get::<_, String>(7)?,
row.get::<_, Option<String>>(8)?,
row.get::<_, Option<i64>>(9)?,
row.get::<_, String>(10)?,
row.get::<_, Option<String>>(11)?,
row.get::<_, Option<i64>>(12)?,
row.get::<_, Option<String>>(13)?,
row.get::<_, Option<String>>(14)?,
row.get::<_, Option<i64>>(15)?,
row.get::<_, Option<i64>>(16)?,
row.get::<_, Option<String>>(17)?,
row.get::<_, Option<String>>(18)?,
row.get::<_, Option<i64>>(19)?,
row.get::<_, String>(20)?,
))
},
)?;
let mut receipts = Vec::new();
let mut metered_receipts = 0_u64;
let mut pending_settlement_receipts = 0_u64;
let mut failed_settlement_receipts = 0_u64;
let mut settlement_actionable_receipts = 0_u64;
let mut metering_actionable_receipts = 0_u64;
let mut metering_evidence_missing_receipts = 0_u64;
let mut metering_financial_mismatch_receipts = 0_u64;
for row in rows {
let (
seq,
receipt_id,
timestamp,
capability_id,
subject_key,
tool_server,
tool_name,
settlement_reconciliation_state_text,
settlement_note,
settlement_updated_at,
metering_reconciliation_state_text,
metering_note,
metering_updated_at,
adapter_kind,
evidence_id,
observed_units,
billed_cost_units,
billed_cost_currency,
evidence_sha256,
recorded_at,
raw_json,
) = row?;
let receipt = decode_verified_chio_receipt(
&raw_json,
"persisted tool receipt",
Some(seq.max(0) as u64),
)?;
let governed = extract_governed_transaction_metadata(&receipt).ok_or_else(|| {
ReceiptStoreError::Canonical(format!(
"receipt {} is missing governed transaction metadata",
receipt.id
))
})?;
let economic_authorization = governed
.economic_authorization
.clone()
.or_else(|| extract_economic_authorization_metadata(&receipt))
.ok_or_else(|| {
ReceiptStoreError::Canonical(format!(
"receipt {} is missing economic authorization metadata",
receipt.id
))
})?;
let financial = extract_financial_metadata(&receipt);
let settlement_reconciliation_state =
parse_settlement_reconciliation_state(&settlement_reconciliation_state_text)?;
let settlement = EconomicReceiptSettlementProjection {
settlement_status: economic_authorization.settlement.settlement_status.clone(),
reconciliation_state: settlement_reconciliation_state,
action_required: settlement_reconciliation_action_required(
economic_authorization.settlement.settlement_status.clone(),
settlement_reconciliation_state,
),
note: settlement_note,
updated_at: settlement_updated_at.map(|value| value.max(0) as u64),
};
if settlement.settlement_status == SettlementStatus::Pending {
pending_settlement_receipts = pending_settlement_receipts.saturating_add(1);
}
if settlement.settlement_status == SettlementStatus::Failed {
failed_settlement_receipts = failed_settlement_receipts.saturating_add(1);
}
if settlement.action_required {
settlement_actionable_receipts = settlement_actionable_receipts.saturating_add(1);
}
let metering = if economic_authorization.metering.is_some() {
metered_receipts = metered_receipts.saturating_add(1);
let governed_metering = governed.metered_billing.as_ref().ok_or_else(|| {
ReceiptStoreError::Canonical(format!(
"receipt {} has economic metering metadata without governed metered billing context",
receipt.id
))
})?;
let evidence = metered_billing_evidence_record_from_columns(
adapter_kind,
evidence_id,
observed_units,
billed_cost_units,
billed_cost_currency,
evidence_sha256,
recorded_at,
);
let reconciliation_state = parse_metered_billing_reconciliation_state(
&metering_reconciliation_state_text,
)?;
let analysis = analyze_metered_billing_reconciliation(
governed_metering,
financial.as_ref(),
evidence.as_ref(),
reconciliation_state,
);
if analysis.action_required {
metering_actionable_receipts = metering_actionable_receipts.saturating_add(1);
}
if analysis.evidence_missing {
metering_evidence_missing_receipts =
metering_evidence_missing_receipts.saturating_add(1);
}
if analysis.financial_mismatch {
metering_financial_mismatch_receipts =
metering_financial_mismatch_receipts.saturating_add(1);
}
Some(EconomicReceiptMeteringProjection {
reconciliation_state,
action_required: analysis.action_required,
evidence_missing: analysis.evidence_missing,
exceeds_quoted_units: analysis.exceeds_quoted_units,
exceeds_max_billed_units: analysis.exceeds_max_billed_units,
exceeds_quoted_cost: analysis.exceeds_quoted_cost,
financial_mismatch: analysis.financial_mismatch,
evidence,
note: metering_note,
updated_at: metering_updated_at.map(|value| value.max(0) as u64),
})
} else {
None
};
receipts.push(EconomicReceiptProjectionRow {
receipt_id,
timestamp: timestamp.max(0) as u64,
capability_id,
subject_key,
tool_server,
tool_name,
economic_authorization,
budget_authority: receipt.financial_budget_authority_metadata(),
settlement,
metering,
});
}
Ok(EconomicReceiptProjectionReport {
summary: EconomicReceiptProjectionSummary {
matching_receipts,
returned_receipts: receipts.len() as u64,
metered_receipts,
pending_settlement_receipts,
failed_settlement_receipts,
settlement_actionable_receipts,
metering_actionable_receipts,
metering_evidence_missing_receipts,
metering_financial_mismatch_receipts,
truncated: matching_receipts > receipts.len() as u64,
},
receipts,
})
}
pub fn query_economic_completion_flow_report(
&self,
query: &ExposureLedgerQuery,
) -> Result<EconomicCompletionFlowReport, ReceiptStoreError> {
let normalized = query.normalized();
if let Err(message) = normalized.validate() {
return Err(ReceiptStoreError::Conflict(message));
}
let economic_receipts =
self.query_economic_receipt_projection_report(&OperatorReportQuery {
capability_id: normalized.capability_id.clone(),
agent_subject: normalized.agent_subject.clone(),
tool_server: normalized.tool_server.clone(),
tool_name: normalized.tool_name.clone(),
since: normalized.since,
until: normalized.until,
economic_limit: normalized.receipt_limit,
..OperatorReportQuery::default()
})?;
let underwriting_decisions =
self.query_underwriting_decisions(&UnderwritingDecisionQuery {
decision_id: None,
capability_id: normalized.capability_id.clone(),
agent_subject: normalized.agent_subject.clone(),
tool_server: normalized.tool_server.clone(),
tool_name: normalized.tool_name.clone(),
outcome: None,
lifecycle_state: None,
appeal_status: None,
limit: normalized.decision_limit,
})?;
let credit_facilities = self.query_credit_facilities(&CreditFacilityListQuery {
facility_id: None,
capability_id: normalized.capability_id.clone(),
agent_subject: normalized.agent_subject.clone(),
tool_server: normalized.tool_server.clone(),
tool_name: normalized.tool_name.clone(),
disposition: None,
lifecycle_state: None,
limit: normalized.decision_limit,
})?;
let credit_bonds = self.query_credit_bonds(&CreditBondListQuery {
bond_id: None,
facility_id: None,
capability_id: normalized.capability_id.clone(),
agent_subject: normalized.agent_subject.clone(),
tool_server: normalized.tool_server.clone(),
tool_name: normalized.tool_name.clone(),
disposition: None,
lifecycle_state: None,
limit: normalized.decision_limit,
})?;
let latest_underwriting = underwriting_decisions
.decisions
.iter()
.find(|row| row.lifecycle_state == UnderwritingDecisionLifecycleState::Active)
.or_else(|| underwriting_decisions.decisions.first());
let latest_credit_facility = credit_facilities
.facilities
.iter()
.find(|row| row.lifecycle_state == CreditFacilityLifecycleState::Active)
.or_else(|| credit_facilities.facilities.first());
let latest_credit_bond = credit_bonds
.bonds
.iter()
.find(|row| row.lifecycle_state == CreditBondLifecycleState::Active)
.or_else(|| credit_bonds.bonds.first());
Ok(EconomicCompletionFlowReport {
schema: ECONOMIC_COMPLETION_FLOW_SCHEMA.to_string(),
generated_at: unix_now(),
filters: normalized,
summary: EconomicCompletionFlowSummary {
matching_receipts: economic_receipts.summary.matching_receipts,
returned_receipts: economic_receipts.summary.returned_receipts,
matching_underwriting_decisions: underwriting_decisions.summary.matching_decisions,
returned_underwriting_decisions: underwriting_decisions.summary.returned_decisions,
matching_credit_facilities: credit_facilities.summary.matching_facilities,
returned_credit_facilities: credit_facilities.summary.returned_facilities,
matching_credit_bonds: credit_bonds.summary.matching_bonds,
returned_credit_bonds: credit_bonds.summary.returned_bonds,
pending_settlement_receipts: economic_receipts.summary.pending_settlement_receipts,
failed_settlement_receipts: economic_receipts.summary.failed_settlement_receipts,
metering_actionable_receipts: economic_receipts
.summary
.metering_actionable_receipts,
latest_underwriting_decision_id: latest_underwriting
.map(|row| row.decision.body.decision_id.clone()),
latest_underwriting_outcome: latest_underwriting
.map(|row| row.decision.body.evaluation.outcome),
latest_credit_facility_id: latest_credit_facility
.map(|row| row.facility.body.facility_id.clone()),
latest_credit_facility_disposition: latest_credit_facility
.map(|row| row.facility.body.report.disposition),
latest_credit_bond_id: latest_credit_bond.map(|row| row.bond.body.bond_id.clone()),
latest_credit_bond_disposition: latest_credit_bond
.map(|row| row.bond.body.report.disposition),
},
economic_receipts,
underwriting_decisions,
credit_facilities,
credit_bonds,
})
}
pub fn query_settlement_reconciliation_report(
&self,
query: &OperatorReportQuery,
) -> Result<SettlementReconciliationReport, ReceiptStoreError> {
let capability_id = query.capability_id.as_deref();
let tool_server = query.tool_server.as_deref();
let tool_name = query.tool_name.as_deref();
let since = query.since.map(|value| value as i64);
let until = query.until.map(|value| value as i64);
let agent_subject = query.agent_subject.as_deref();
let row_limit = query.settlement_limit_or_default();
let summary_sql = r#"
SELECT
COUNT(*) AS matching_receipts,
COALESCE(SUM(
CASE
WHEN json_extract(r.raw_json, '$.metadata.financial.settlement_status') = 'pending' THEN 1
ELSE 0
END
), 0) AS pending_receipts,
COALESCE(SUM(
CASE
WHEN json_extract(r.raw_json, '$.metadata.financial.settlement_status') = 'failed' THEN 1
ELSE 0
END
), 0) AS failed_receipts,
COALESCE(SUM(
CASE
WHEN COALESCE(sr.reconciliation_state, 'open') NOT IN ('reconciled', 'ignored') THEN 1
ELSE 0
END
), 0) AS actionable_receipts,
COALESCE(SUM(
CASE
WHEN COALESCE(sr.reconciliation_state, 'open') = 'reconciled' THEN 1
ELSE 0
END
), 0) AS reconciled_receipts
FROM chio_tool_receipts r
LEFT JOIN capability_lineage cl ON r.capability_id = cl.capability_id
LEFT JOIN settlement_reconciliations sr ON r.receipt_id = sr.receipt_id
WHERE json_extract(r.raw_json, '$.metadata.financial.settlement_status') IN ('pending', 'failed')
AND (?1 IS NULL OR r.capability_id = ?1)
AND (?2 IS NULL OR r.tool_server = ?2)
AND (?3 IS NULL OR r.tool_name = ?3)
AND (?4 IS NULL OR r.timestamp >= ?4)
AND (?5 IS NULL OR r.timestamp <= ?5)
AND (?6 IS NULL OR COALESCE(r.subject_key, cl.subject_key) = ?6)
"#;
let (
matching_receipts,
pending_receipts,
failed_receipts,
actionable_receipts,
reconciled_receipts,
) = self.connection()?.query_row(
summary_sql,
params![
capability_id,
tool_server,
tool_name,
since,
until,
agent_subject
],
|row| {
Ok((
row.get::<_, i64>(0)?.max(0) as u64,
row.get::<_, i64>(1)?.max(0) as u64,
row.get::<_, i64>(2)?.max(0) as u64,
row.get::<_, i64>(3)?.max(0) as u64,
row.get::<_, i64>(4)?.max(0) as u64,
))
},
)?;
let rows_sql = r#"
SELECT
r.seq,
r.receipt_id,
r.timestamp,
r.capability_id,
COALESCE(r.subject_key, cl.subject_key),
r.tool_server,
r.tool_name,
json_extract(r.raw_json, '$.metadata.financial.payment_reference'),
json_extract(r.raw_json, '$.metadata.financial.settlement_status'),
CAST(json_extract(r.raw_json, '$.metadata.financial.cost_charged') AS INTEGER),
json_extract(r.raw_json, '$.metadata.financial.currency'),
COALESCE(sr.reconciliation_state, 'open'),
sr.note,
sr.updated_at,
r.raw_json
FROM chio_tool_receipts r
LEFT JOIN capability_lineage cl ON r.capability_id = cl.capability_id
LEFT JOIN settlement_reconciliations sr ON r.receipt_id = sr.receipt_id
WHERE json_extract(r.raw_json, '$.metadata.financial.settlement_status') IN ('pending', 'failed')
AND (?1 IS NULL OR r.capability_id = ?1)
AND (?2 IS NULL OR r.tool_server = ?2)
AND (?3 IS NULL OR r.tool_name = ?3)
AND (?4 IS NULL OR r.timestamp >= ?4)
AND (?5 IS NULL OR r.timestamp <= ?5)
AND (?6 IS NULL OR COALESCE(r.subject_key, cl.subject_key) = ?6)
ORDER BY r.timestamp DESC, r.seq DESC
LIMIT ?7
"#;
let connection = self.connection()?;
let mut stmt = connection.prepare(rows_sql)?;
let rows = stmt.query_map(
params![
capability_id,
tool_server,
tool_name,
since,
until,
agent_subject,
row_limit as i64
],
|row| {
Ok((
row.get::<_, i64>(0)?,
row.get::<_, String>(1)?,
row.get::<_, i64>(2)?,
row.get::<_, String>(3)?,
row.get::<_, Option<String>>(4)?,
row.get::<_, String>(5)?,
row.get::<_, String>(6)?,
row.get::<_, Option<String>>(7)?,
row.get::<_, String>(8)?,
row.get::<_, Option<i64>>(9)?,
row.get::<_, Option<String>>(10)?,
row.get::<_, String>(11)?,
row.get::<_, Option<String>>(12)?,
row.get::<_, Option<i64>>(13)?,
row.get::<_, String>(14)?,
))
},
)?;
let mut receipts = Vec::new();
for row in rows {
let (
seq,
receipt_id,
timestamp,
capability_id,
subject_key,
tool_server,
tool_name,
payment_reference,
settlement_status_text,
cost_charged,
currency,
reconciliation_state_text,
note,
updated_at,
raw_json,
) = row?;
let receipt = decode_verified_chio_receipt(
&raw_json,
"persisted tool receipt",
Some(seq.max(0) as u64),
)?;
let settlement_status = parse_settlement_status(&settlement_status_text)?;
let reconciliation_state =
parse_settlement_reconciliation_state(&reconciliation_state_text)?;
let action_required = settlement_reconciliation_action_required(
settlement_status.clone(),
reconciliation_state,
);
receipts.push(SettlementReconciliationRow {
receipt_id,
timestamp: timestamp.max(0) as u64,
capability_id,
subject_key,
tool_server,
tool_name,
payment_reference,
settlement_status,
cost_charged: cost_charged.map(|value| value.max(0) as u64),
currency,
budget_authority: receipt.financial_budget_authority_metadata(),
reconciliation_state,
action_required,
note,
updated_at: updated_at.map(|value| value.max(0) as u64),
});
}
Ok(SettlementReconciliationReport {
summary: SettlementReconciliationSummary {
matching_receipts,
returned_receipts: receipts.len() as u64,
pending_receipts,
failed_receipts,
actionable_receipts,
reconciled_receipts,
truncated: matching_receipts > receipts.len() as u64,
},
receipts,
})
}
pub fn query_authorization_context_report(
&self,
query: &OperatorReportQuery,
) -> Result<AuthorizationContextReport, ReceiptStoreError> {
let capability_id = query.capability_id.as_deref();
let tool_server = query.tool_server.as_deref();
let tool_name = query.tool_name.as_deref();
let since = query.since.map(|value| value as i64);
let until = query.until.map(|value| value as i64);
let agent_subject = query.agent_subject.as_deref();
let row_limit = query.authorization_limit_or_default();
let summary_sql = r#"
SELECT
COUNT(*),
COALESCE(SUM(
CASE
WHEN json_type(r.raw_json, '$.metadata.governed_transaction.approval') = 'object' THEN 1
ELSE 0
END
), 0),
COALESCE(SUM(
CASE
WHEN json_extract(r.raw_json, '$.metadata.governed_transaction.approval.approved') = 1 THEN 1
ELSE 0
END
), 0),
COALESCE(SUM(
CASE
WHEN json_type(r.raw_json, '$.metadata.governed_transaction.commerce') = 'object' THEN 1
ELSE 0
END
), 0),
COALESCE(SUM(
CASE
WHEN json_type(r.raw_json, '$.metadata.governed_transaction.metered_billing') = 'object' THEN 1
ELSE 0
END
), 0),
COALESCE(SUM(
CASE
WHEN json_type(r.raw_json, '$.metadata.governed_transaction.runtime_assurance') = 'object' THEN 1
ELSE 0
END
), 0),
COALESCE(SUM(
CASE
WHEN json_type(r.raw_json, '$.metadata.governed_transaction.call_chain') = 'object' THEN 1
ELSE 0
END
), 0),
COALESCE(SUM(
CASE
WHEN json_type(r.raw_json, '$.metadata.governed_transaction.max_amount') = 'object' THEN 1
ELSE 0
END
), 0)
FROM chio_tool_receipts r
LEFT JOIN capability_lineage cl ON r.capability_id = cl.capability_id
WHERE json_type(r.raw_json, '$.metadata.governed_transaction') = 'object'
AND (?1 IS NULL OR r.capability_id = ?1)
AND (?2 IS NULL OR r.tool_server = ?2)
AND (?3 IS NULL OR r.tool_name = ?3)
AND (?4 IS NULL OR r.timestamp >= ?4)
AND (?5 IS NULL OR r.timestamp <= ?5)
AND (?6 IS NULL OR COALESCE(r.subject_key, cl.subject_key) = ?6)
"#;
let (
matching_receipts,
approval_receipts,
approved_receipts,
commerce_receipts,
metered_billing_receipts,
runtime_assurance_receipts,
_call_chain_receipts,
max_amount_receipts,
) = self.connection()?.query_row(
summary_sql,
params![
capability_id,
tool_server,
tool_name,
since,
until,
agent_subject
],
|row| {
Ok((
row.get::<_, i64>(0)?.max(0) as u64,
row.get::<_, i64>(1)?.max(0) as u64,
row.get::<_, i64>(2)?.max(0) as u64,
row.get::<_, i64>(3)?.max(0) as u64,
row.get::<_, i64>(4)?.max(0) as u64,
row.get::<_, i64>(5)?.max(0) as u64,
row.get::<_, i64>(6)?.max(0) as u64,
row.get::<_, i64>(7)?.max(0) as u64,
))
},
)?;
let rows_sql = r#"
SELECT
r.seq,
r.raw_json,
r.subject_key,
r.issuer_key,
cl.subject_key,
cl.issuer_key,
r.grant_index,
cl.grants_json,
rls.statement_id,
rls.session_anchor_id
FROM chio_tool_receipts r
LEFT JOIN capability_lineage cl ON r.capability_id = cl.capability_id
LEFT JOIN receipt_lineage_statements rls ON r.receipt_id = rls.receipt_id
WHERE json_type(r.raw_json, '$.metadata.governed_transaction') = 'object'
AND (?1 IS NULL OR r.capability_id = ?1)
AND (?2 IS NULL OR r.tool_server = ?2)
AND (?3 IS NULL OR r.tool_name = ?3)
AND (?4 IS NULL OR r.timestamp >= ?4)
AND (?5 IS NULL OR r.timestamp <= ?5)
AND (?6 IS NULL OR COALESCE(r.subject_key, cl.subject_key) = ?6)
ORDER BY r.timestamp DESC, r.seq DESC
"#;
let connection = self.connection()?;
let mut stmt = connection.prepare(rows_sql)?;
let rows = stmt.query_map(
params![
capability_id,
tool_server,
tool_name,
since,
until,
agent_subject
],
|row| {
Ok((
row.get::<_, i64>(0)?,
row.get::<_, String>(1)?,
row.get::<_, Option<String>>(2)?,
row.get::<_, Option<String>>(3)?,
row.get::<_, Option<String>>(4)?,
row.get::<_, Option<String>>(5)?,
row.get::<_, Option<i64>>(6)?,
row.get::<_, Option<String>>(7)?,
row.get::<_, Option<String>>(8)?,
row.get::<_, Option<String>>(9)?,
))
},
)?;
let mut sender_bound_receipts = 0_u64;
let mut dpop_bound_receipts = 0_u64;
let mut runtime_assurance_bound_receipts = 0_u64;
let mut delegated_sender_bound_receipts = 0_u64;
let mut call_chain_receipts = 0_u64;
let mut asserted_call_chain_receipts = 0_u64;
let mut observed_call_chain_receipts = 0_u64;
let mut verified_call_chain_receipts = 0_u64;
let mut session_anchor_receipts = 0_u64;
let mut request_lineage_receipts = 0_u64;
let mut receipt_lineage_statement_receipts = 0_u64;
let mut receipts = Vec::new();
for row in rows {
let (
seq,
raw_json,
receipt_subject_key,
receipt_issuer_key,
lineage_subject_key,
lineage_issuer_key,
persisted_grant_index,
grants_json,
persisted_statement_id,
persisted_session_anchor_id,
) = row?;
let receipt = decode_verified_chio_receipt(
&raw_json,
"persisted tool receipt",
Some(seq.max(0) as u64),
)?;
let governed = extract_governed_transaction_metadata(&receipt).ok_or_else(|| {
ReceiptStoreError::Canonical(format!(
"receipt {} is missing governed transaction metadata",
receipt.id
))
})?;
let mut projection = sanitize_governed_transaction_projection(&receipt, &governed);
apply_persisted_lineage_references(
&mut projection,
&PersistedLineageReferences {
statement_id: persisted_statement_id,
session_anchor_id: persisted_session_anchor_id,
},
);
let attribution = extract_receipt_attribution(&receipt);
let transaction_context =
authorization_transaction_context_from_projection(&projection);
let sender_constraint = derive_authorization_sender_constraint(
&receipt.id,
AuthorizationSenderConstraintArgs {
tool_server: &receipt.tool_server,
tool_name: &receipt.tool_name,
receipt_subject_key: receipt_subject_key.as_deref(),
receipt_issuer_key: receipt_issuer_key.as_deref(),
lineage_subject_key: lineage_subject_key.as_deref(),
lineage_issuer_key: lineage_issuer_key.as_deref(),
grant_index: attribution
.grant_index
.or_else(|| persisted_grant_index.map(|value| value.max(0) as u32)),
grants_json: grants_json.as_deref(),
},
&transaction_context,
)?;
let authorization_row = AuthorizationContextRow {
receipt_id: receipt.id,
timestamp: receipt.timestamp,
capability_id: receipt.capability_id,
subject_key: Some(sender_constraint.subject_key.clone()),
tool_server: receipt.tool_server,
tool_name: receipt.tool_name,
decision: receipt.decision,
authorization_details: authorization_details_from_governed_metadata(
&projection.strong,
),
transaction_context,
governed_transaction_diagnostics: projection.diagnostics.clone(),
sender_constraint,
};
validate_chio_oauth_authorization_row(&authorization_row)?;
if let Some(evidence_class) = call_chain_evidence_class(&projection) {
call_chain_receipts += 1;
match evidence_class {
GovernedProvenanceEvidenceClass::Asserted => {
asserted_call_chain_receipts += 1;
}
GovernedProvenanceEvidenceClass::Observed => {
observed_call_chain_receipts += 1;
}
GovernedProvenanceEvidenceClass::Verified => {
verified_call_chain_receipts += 1;
}
}
}
if let Some(lineage_references) = lineage_references_from_projection(&projection) {
if lineage_references.session_anchor_id.is_some() {
session_anchor_receipts += 1;
}
if lineage_references.request_lineage_id.is_some() {
request_lineage_receipts += 1;
}
if lineage_references.receipt_lineage_statement_id.is_some() {
receipt_lineage_statement_receipts += 1;
}
}
sender_bound_receipts += 1;
if authorization_row.sender_constraint.proof_required {
dpop_bound_receipts += 1;
}
if authorization_row.sender_constraint.runtime_assurance_bound {
runtime_assurance_bound_receipts += 1;
}
if authorization_row
.sender_constraint
.delegated_call_chain_bound
{
delegated_sender_bound_receipts += 1;
}
if receipts.len() < row_limit {
receipts.push(authorization_row);
}
}
Ok(AuthorizationContextReport {
schema: CHIO_OAUTH_AUTHORIZATION_CONTEXT_REPORT_SCHEMA.to_string(),
profile: ChioOAuthAuthorizationProfile::default(),
summary: AuthorizationContextSummary {
matching_receipts,
returned_receipts: receipts.len() as u64,
approval_receipts,
approved_receipts,
commerce_receipts,
metered_billing_receipts,
runtime_assurance_receipts,
call_chain_receipts,
asserted_call_chain_receipts,
observed_call_chain_receipts,
verified_call_chain_receipts,
max_amount_receipts,
sender_bound_receipts,
dpop_bound_receipts,
runtime_assurance_bound_receipts,
delegated_sender_bound_receipts,
session_anchor_receipts,
request_lineage_receipts,
receipt_lineage_statement_receipts,
truncated: matching_receipts > receipts.len() as u64,
},
receipts,
})
}
pub fn authorization_profile_metadata_report(&self) -> ChioOAuthAuthorizationMetadataReport {
ChioOAuthAuthorizationMetadataReport {
schema: CHIO_OAUTH_AUTHORIZATION_METADATA_SCHEMA.to_string(),
generated_at: unix_now(),
profile: ChioOAuthAuthorizationProfile::default(),
report_schema: CHIO_OAUTH_AUTHORIZATION_CONTEXT_REPORT_SCHEMA.to_string(),
discovery: ChioOAuthAuthorizationDiscoveryMetadata {
protected_resource_metadata_paths: vec![
"/.well-known/oauth-protected-resource".to_string(),
"/.well-known/oauth-protected-resource/mcp".to_string(),
],
authorization_server_metadata_path_template:
"/.well-known/oauth-authorization-server/{issuer-path}".to_string(),
discovery_informational_only: true,
},
support_boundary: ChioOAuthAuthorizationSupportBoundary {
governed_receipts_authoritative: true,
hosted_request_time_authorization_supported: true,
resource_indicator_binding_supported: true,
sender_constrained_projection: true,
runtime_assurance_projection: true,
delegated_call_chain_projection: true,
generic_token_issuance_supported: false,
oidc_identity_assertions_supported: false,
mtls_transport_binding_in_profile: false,
approval_tokens_runtime_authorization_supported: false,
capabilities_runtime_authorization_supported: false,
reviewer_evidence_runtime_authorization_supported: false,
},
example_mapping: ChioOAuthAuthorizationExampleMapping {
authorization_detail_types: vec![
"type".to_string(),
"locations".to_string(),
"actions".to_string(),
"purpose".to_string(),
"maxAmount".to_string(),
"commerce".to_string(),
"meteredBilling".to_string(),
],
transaction_context_fields: ChioOAuthAuthorizationProfile::default()
.transaction_context_fields,
sender_constraint_fields: vec![
"subjectKey".to_string(),
"subjectKeySource".to_string(),
"issuerKey".to_string(),
"issuerKeySource".to_string(),
"matchedGrantIndex".to_string(),
"proofRequired".to_string(),
"proofType".to_string(),
"proofSchema".to_string(),
"runtimeAssuranceBound".to_string(),
"delegatedCallChainBound".to_string(),
],
},
}
}
pub fn query_authorization_review_pack(
&self,
query: &OperatorReportQuery,
) -> Result<ChioOAuthAuthorizationReviewPack, ReceiptStoreError> {
let authorization_context = self.query_authorization_context_report(query)?;
let metadata = self.authorization_profile_metadata_report();
let mut records = Vec::with_capacity(authorization_context.receipts.len());
for row in authorization_context.receipts {
let (seq, raw_json) = self.connection()?.query_row(
"SELECT seq, raw_json FROM chio_tool_receipts WHERE receipt_id = ?1",
params![row.receipt_id.as_str()],
|db_row| Ok((db_row.get::<_, i64>(0)?, db_row.get::<_, String>(1)?)),
)?;
let signed_receipt = decode_verified_chio_receipt(
&raw_json,
"persisted tool receipt",
Some(seq.max(0) as u64),
)?;
let governed_transaction = extract_governed_transaction_metadata(&signed_receipt)
.ok_or_else(|| {
ReceiptStoreError::Canonical(format!(
"receipt {} is missing governed transaction metadata",
signed_receipt.id
))
})?;
let mut projection =
sanitize_governed_transaction_projection(&signed_receipt, &governed_transaction);
if let Some(diagnostics) = row.governed_transaction_diagnostics.as_ref() {
apply_persisted_lineage_references(
&mut projection,
&PersistedLineageReferences {
statement_id: diagnostics
.lineage_references
.receipt_lineage_statement_id
.clone(),
session_anchor_id: diagnostics.lineage_references.session_anchor_id.clone(),
},
);
}
records.push(ChioOAuthAuthorizationReviewPackRecord {
receipt_id: row.receipt_id.clone(),
capability_id: row.capability_id.clone(),
authorization_context: row,
governed_transaction: projection.strong,
governed_transaction_diagnostics: projection.diagnostics,
signed_receipt,
});
}
Ok(ChioOAuthAuthorizationReviewPack {
schema: CHIO_OAUTH_AUTHORIZATION_REVIEW_PACK_SCHEMA.to_string(),
generated_at: unix_now(),
filters: query.clone(),
metadata,
summary: ChioOAuthAuthorizationReviewPackSummary {
matching_receipts: authorization_context.summary.matching_receipts,
returned_receipts: records.len() as u64,
dpop_required_receipts: authorization_context.summary.dpop_bound_receipts,
runtime_assurance_receipts: authorization_context
.summary
.runtime_assurance_bound_receipts,
delegated_call_chain_receipts: authorization_context.summary.call_chain_receipts,
asserted_call_chain_receipts: authorization_context
.summary
.asserted_call_chain_receipts,
observed_call_chain_receipts: authorization_context
.summary
.observed_call_chain_receipts,
verified_call_chain_receipts: authorization_context
.summary
.verified_call_chain_receipts,
session_anchor_receipts: authorization_context.summary.session_anchor_receipts,
request_lineage_receipts: authorization_context.summary.request_lineage_receipts,
receipt_lineage_statement_receipts: authorization_context
.summary
.receipt_lineage_statement_receipts,
truncated: authorization_context.summary.truncated,
},
records,
})
}
pub fn query_behavioral_feed_receipts(
&self,
query: &BehavioralFeedQuery,
) -> Result<
(
BehavioralFeedSettlementSummary,
BehavioralFeedGovernedActionSummary,
BehavioralFeedMeteredBillingSummary,
BehavioralFeedReceiptSelection,
),
ReceiptStoreError,
> {
let operator_query = query.to_operator_report_query();
let capability_id = operator_query.capability_id.as_deref();
let tool_server = operator_query.tool_server.as_deref();
let tool_name = operator_query.tool_name.as_deref();
let since = operator_query.since.map(|value| value as i64);
let until = operator_query.until.map(|value| value as i64);
let agent_subject = operator_query.agent_subject.as_deref();
let summary_sql = r#"
SELECT
COALESCE(SUM(
CASE
WHEN COALESCE(json_extract(r.raw_json, '$.metadata.financial.settlement_status'), 'not_applicable') = 'pending' THEN 1
ELSE 0
END
), 0),
COALESCE(SUM(
CASE
WHEN COALESCE(json_extract(r.raw_json, '$.metadata.financial.settlement_status'), 'not_applicable') = 'settled' THEN 1
ELSE 0
END
), 0),
COALESCE(SUM(
CASE
WHEN COALESCE(json_extract(r.raw_json, '$.metadata.financial.settlement_status'), 'not_applicable') = 'failed' THEN 1
ELSE 0
END
), 0),
COALESCE(SUM(
CASE
WHEN COALESCE(json_extract(r.raw_json, '$.metadata.financial.settlement_status'), 'not_applicable') = 'not_applicable' THEN 1
ELSE 0
END
), 0),
COALESCE(SUM(
CASE
WHEN COALESCE(json_extract(r.raw_json, '$.metadata.financial.settlement_status'), 'not_applicable') IN ('pending', 'failed')
AND COALESCE(sr.reconciliation_state, 'open') NOT IN ('reconciled', 'ignored')
THEN 1
ELSE 0
END
), 0),
COALESCE(SUM(
CASE
WHEN COALESCE(sr.reconciliation_state, 'open') = 'reconciled' THEN 1
ELSE 0
END
), 0),
COALESCE(SUM(
CASE
WHEN json_type(r.raw_json, '$.metadata.governed_transaction') IS NOT NULL THEN 1
ELSE 0
END
), 0),
COALESCE(SUM(
CASE
WHEN json_type(r.raw_json, '$.metadata.governed_transaction.approval') IS NOT NULL THEN 1
ELSE 0
END
), 0),
COALESCE(SUM(
CASE
WHEN json_extract(r.raw_json, '$.metadata.governed_transaction.approval.approved') = 1 THEN 1
ELSE 0
END
), 0),
COALESCE(SUM(
CASE
WHEN json_type(r.raw_json, '$.metadata.governed_transaction.commerce') IS NOT NULL THEN 1
ELSE 0
END
), 0),
COALESCE(SUM(
CASE
WHEN json_type(r.raw_json, '$.metadata.governed_transaction.max_amount') IS NOT NULL THEN 1
ELSE 0
END
), 0)
FROM chio_tool_receipts r
LEFT JOIN capability_lineage cl ON r.capability_id = cl.capability_id
LEFT JOIN settlement_reconciliations sr ON r.receipt_id = sr.receipt_id
WHERE (?1 IS NULL OR r.capability_id = ?1)
AND (?2 IS NULL OR r.tool_server = ?2)
AND (?3 IS NULL OR r.tool_name = ?3)
AND (?4 IS NULL OR r.timestamp >= ?4)
AND (?5 IS NULL OR r.timestamp <= ?5)
AND (?6 IS NULL OR COALESCE(r.subject_key, cl.subject_key) = ?6)
"#;
let (settlements, governed_actions) = self.connection()?.query_row(
summary_sql,
params![
capability_id,
tool_server,
tool_name,
since,
until,
agent_subject
],
|row| {
Ok((
BehavioralFeedSettlementSummary {
pending_receipts: row.get::<_, i64>(0)?.max(0) as u64,
settled_receipts: row.get::<_, i64>(1)?.max(0) as u64,
failed_receipts: row.get::<_, i64>(2)?.max(0) as u64,
not_applicable_receipts: row.get::<_, i64>(3)?.max(0) as u64,
actionable_receipts: row.get::<_, i64>(4)?.max(0) as u64,
reconciled_receipts: row.get::<_, i64>(5)?.max(0) as u64,
},
BehavioralFeedGovernedActionSummary {
governed_receipts: row.get::<_, i64>(6)?.max(0) as u64,
approval_receipts: row.get::<_, i64>(7)?.max(0) as u64,
approved_receipts: row.get::<_, i64>(8)?.max(0) as u64,
commerce_receipts: row.get::<_, i64>(9)?.max(0) as u64,
max_amount_receipts: row.get::<_, i64>(10)?.max(0) as u64,
},
))
},
)?;
let metered_billing = self.query_metered_billing_summary(&operator_query)?;
let row_limit = query.receipt_limit_or_default();
let count_sql = r#"
SELECT COUNT(*)
FROM chio_tool_receipts r
LEFT JOIN capability_lineage cl ON r.capability_id = cl.capability_id
WHERE (?1 IS NULL OR r.capability_id = ?1)
AND (?2 IS NULL OR r.tool_server = ?2)
AND (?3 IS NULL OR r.tool_name = ?3)
AND (?4 IS NULL OR r.timestamp >= ?4)
AND (?5 IS NULL OR r.timestamp <= ?5)
AND (?6 IS NULL OR COALESCE(r.subject_key, cl.subject_key) = ?6)
"#;
let matching_receipts = self
.connection()?
.query_row(
count_sql,
params![
capability_id,
tool_server,
tool_name,
since,
until,
agent_subject
],
|row| row.get::<_, i64>(0),
)
.map(|value| value.max(0) as u64)?;
let rows_sql = r#"
SELECT r.seq, r.raw_json
FROM chio_tool_receipts r
LEFT JOIN capability_lineage cl ON r.capability_id = cl.capability_id
WHERE (?1 IS NULL OR r.capability_id = ?1)
AND (?2 IS NULL OR r.tool_server = ?2)
AND (?3 IS NULL OR r.tool_name = ?3)
AND (?4 IS NULL OR r.timestamp >= ?4)
AND (?5 IS NULL OR r.timestamp <= ?5)
AND (?6 IS NULL OR COALESCE(r.subject_key, cl.subject_key) = ?6)
ORDER BY r.timestamp DESC, r.seq DESC
LIMIT ?7
"#;
let connection = self.connection()?;
let mut stmt = connection.prepare(rows_sql)?;
let rows = stmt.query_map(
params![
capability_id,
tool_server,
tool_name,
since,
until,
agent_subject,
row_limit as i64,
],
|row| Ok((row.get::<_, i64>(0)?, row.get::<_, String>(1)?)),
)?;
let mut receipts = Vec::with_capacity(row_limit);
for row in rows {
let (seq, raw_json) = row?;
let receipt = decode_verified_chio_receipt(
&raw_json,
"persisted tool receipt",
Some(seq.max(0) as u64),
)?;
receipts.push(self.behavioral_feed_receipt_row_from_receipt(receipt)?);
}
Ok((
settlements,
governed_actions,
metered_billing,
BehavioralFeedReceiptSelection {
matching_receipts,
receipts,
},
))
}
pub fn query_recent_credit_loss_receipts(
&self,
query: &BehavioralFeedQuery,
limit: usize,
) -> Result<(u64, Vec<BehavioralFeedReceiptRow>), ReceiptStoreError> {
let operator_query = query.to_operator_report_query();
let capability_id = operator_query.capability_id.as_deref();
let tool_server = operator_query.tool_server.as_deref();
let tool_name = operator_query.tool_name.as_deref();
let since = operator_query.since.map(|value| value as i64);
let until = operator_query.until.map(|value| value as i64);
let agent_subject = operator_query.agent_subject.as_deref();
let row_limit = limit.max(1);
let count_sql = r#"
SELECT COUNT(*)
FROM chio_tool_receipts r
LEFT JOIN capability_lineage cl ON r.capability_id = cl.capability_id
LEFT JOIN settlement_reconciliations sr ON r.receipt_id = sr.receipt_id
WHERE (?1 IS NULL OR r.capability_id = ?1)
AND (?2 IS NULL OR r.tool_server = ?2)
AND (?3 IS NULL OR r.tool_name = ?3)
AND (?4 IS NULL OR r.timestamp >= ?4)
AND (?5 IS NULL OR r.timestamp <= ?5)
AND (?6 IS NULL OR COALESCE(r.subject_key, cl.subject_key) = ?6)
AND (
COALESCE(json_extract(r.raw_json, '$.metadata.financial.settlement_status'), 'not_applicable') = 'failed'
OR (
COALESCE(json_extract(r.raw_json, '$.metadata.financial.settlement_status'), 'not_applicable') IN ('pending', 'failed')
AND COALESCE(sr.reconciliation_state, 'open') NOT IN ('reconciled', 'ignored')
)
)
"#;
let matching_loss_events = self
.connection()?
.query_row(
count_sql,
params![
capability_id,
tool_server,
tool_name,
since,
until,
agent_subject
],
|row| row.get::<_, i64>(0),
)
.map(|value| value.max(0) as u64)?;
let rows_sql = r#"
SELECT r.seq, r.raw_json
FROM chio_tool_receipts r
LEFT JOIN capability_lineage cl ON r.capability_id = cl.capability_id
LEFT JOIN settlement_reconciliations sr ON r.receipt_id = sr.receipt_id
WHERE (?1 IS NULL OR r.capability_id = ?1)
AND (?2 IS NULL OR r.tool_server = ?2)
AND (?3 IS NULL OR r.tool_name = ?3)
AND (?4 IS NULL OR r.timestamp >= ?4)
AND (?5 IS NULL OR r.timestamp <= ?5)
AND (?6 IS NULL OR COALESCE(r.subject_key, cl.subject_key) = ?6)
AND (
COALESCE(json_extract(r.raw_json, '$.metadata.financial.settlement_status'), 'not_applicable') = 'failed'
OR (
COALESCE(json_extract(r.raw_json, '$.metadata.financial.settlement_status'), 'not_applicable') IN ('pending', 'failed')
AND COALESCE(sr.reconciliation_state, 'open') NOT IN ('reconciled', 'ignored')
)
)
ORDER BY r.timestamp DESC, r.seq DESC
LIMIT ?7
"#;
let connection = self.connection()?;
let mut stmt = connection.prepare(rows_sql)?;
let rows = stmt.query_map(
params![
capability_id,
tool_server,
tool_name,
since,
until,
agent_subject,
row_limit as i64
],
|row| Ok((row.get::<_, i64>(0)?, row.get::<_, String>(1)?)),
)?;
let mut receipts = Vec::new();
for row in rows {
let (seq, raw_json) = row?;
let receipt = decode_verified_chio_receipt(
&raw_json,
"persisted tool receipt",
Some(seq.max(0) as u64),
)?;
receipts.push(self.behavioral_feed_receipt_row_from_receipt(receipt)?);
}
Ok((matching_loss_events, receipts))
}
pub(crate) fn behavioral_feed_receipt_row_from_receipt(
&self,
receipt: ChioReceipt,
) -> Result<BehavioralFeedReceiptRow, ReceiptStoreError> {
let attribution = extract_receipt_attribution(&receipt);
let lineage = if attribution.subject_key.is_none() || attribution.issuer_key.is_none() {
self.get_combined_lineage(&receipt.capability_id)?
} else {
None
};
let financial = extract_financial_metadata(&receipt);
let governed = extract_governed_transaction_metadata(&receipt);
let governed_projection = governed
.as_ref()
.map(|metadata| sanitize_governed_transaction_projection(&receipt, metadata));
let metered_reconciliation = governed
.as_ref()
.and_then(|metadata| metadata.metered_billing.as_ref())
.map(|metered| {
let evidence = self.load_metered_billing_evidence_record(&receipt.id)?;
let reconciliation_state = self
.connection()?
.query_row(
r#"
SELECT COALESCE(reconciliation_state, 'open')
FROM metered_billing_reconciliations
WHERE receipt_id = ?1
"#,
params![&receipt.id],
|row| row.get::<_, String>(0),
)
.optional()?
.map(|value| parse_metered_billing_reconciliation_state(&value))
.transpose()?
.unwrap_or(MeteredBillingReconciliationState::Open);
let analysis = analyze_metered_billing_reconciliation(
metered,
financial.as_ref(),
evidence.as_ref(),
reconciliation_state,
);
Ok::<BehavioralFeedMeteredBillingRow, ReceiptStoreError>(
BehavioralFeedMeteredBillingRow {
reconciliation_state,
action_required: analysis.action_required,
evidence_missing: analysis.evidence_missing,
exceeds_quoted_units: analysis.exceeds_quoted_units,
exceeds_max_billed_units: analysis.exceeds_max_billed_units,
exceeds_quoted_cost: analysis.exceeds_quoted_cost,
financial_mismatch: analysis.financial_mismatch,
evidence,
},
)
})
.transpose()?;
let settlement_status = financial
.as_ref()
.map(|metadata| metadata.settlement_status.clone())
.unwrap_or(SettlementStatus::NotApplicable);
let reconciliation_state = self
.connection()?
.query_row(
r#"
SELECT COALESCE(reconciliation_state, 'open')
FROM settlement_reconciliations
WHERE receipt_id = ?1
"#,
params![receipt.id],
|row| row.get::<_, String>(0),
)
.optional()?
.map(|value| parse_settlement_reconciliation_state(&value))
.transpose()?
.unwrap_or(SettlementReconciliationState::Open);
let action_required = settlement_reconciliation_action_required(
settlement_status.clone(),
reconciliation_state,
);
let budget_authority = receipt.financial_budget_authority_metadata();
Ok(BehavioralFeedReceiptRow {
receipt_id: receipt.id,
timestamp: receipt.timestamp,
capability_id: receipt.capability_id,
subject_key: attribution.subject_key.or_else(|| {
lineage
.as_ref()
.map(|snapshot| snapshot.subject_key.clone())
}),
issuer_key: attribution
.issuer_key
.or_else(|| lineage.as_ref().map(|snapshot| snapshot.issuer_key.clone())),
tool_server: receipt.tool_server,
tool_name: receipt.tool_name,
decision: receipt.decision,
settlement_status,
reconciliation_state,
action_required,
cost_charged: financial.as_ref().map(|metadata| metadata.cost_charged),
attempted_cost: financial
.as_ref()
.and_then(|metadata| metadata.attempted_cost),
currency: financial.as_ref().map(|metadata| metadata.currency.clone()),
budget_authority,
governed: governed_projection
.as_ref()
.map(|projection| projection.strong.clone()),
governed_transaction_diagnostics: governed_projection
.and_then(|projection| projection.diagnostics),
metered_reconciliation,
})
}
pub(crate) fn query_metered_billing_summary(
&self,
query: &OperatorReportQuery,
) -> Result<BehavioralFeedMeteredBillingSummary, ReceiptStoreError> {
let capability_id = query.capability_id.as_deref();
let tool_server = query.tool_server.as_deref();
let tool_name = query.tool_name.as_deref();
let since = query.since.map(|value| value as i64);
let until = query.until.map(|value| value as i64);
let agent_subject = query.agent_subject.as_deref();
let summary_sql = r#"
SELECT
COUNT(*) AS matching_receipts,
COALESCE(SUM(CASE WHEN mbr.receipt_id IS NOT NULL THEN 1 ELSE 0 END), 0),
COALESCE(SUM(CASE WHEN mbr.receipt_id IS NULL THEN 1 ELSE 0 END), 0),
COALESCE(SUM(
CASE
WHEN mbr.receipt_id IS NOT NULL
AND mbr.observed_units > CAST(json_extract(r.raw_json, '$.metadata.governed_transaction.metered_billing.quote.quotedUnits') AS INTEGER)
THEN 1
ELSE 0
END
), 0),
COALESCE(SUM(
CASE
WHEN mbr.receipt_id IS NOT NULL
AND json_extract(r.raw_json, '$.metadata.governed_transaction.metered_billing.maxBilledUnits') IS NOT NULL
AND mbr.observed_units > CAST(json_extract(r.raw_json, '$.metadata.governed_transaction.metered_billing.maxBilledUnits') AS INTEGER)
THEN 1
ELSE 0
END
), 0),
COALESCE(SUM(
CASE
WHEN mbr.receipt_id IS NOT NULL
AND (
mbr.billed_cost_currency != json_extract(r.raw_json, '$.metadata.governed_transaction.metered_billing.quote.quotedCost.currency')
OR mbr.billed_cost_units > CAST(json_extract(r.raw_json, '$.metadata.governed_transaction.metered_billing.quote.quotedCost.units') AS INTEGER)
)
THEN 1
ELSE 0
END
), 0),
COALESCE(SUM(
CASE
WHEN mbr.receipt_id IS NOT NULL
AND json_type(r.raw_json, '$.metadata.financial') = 'object'
AND (
mbr.billed_cost_currency != json_extract(r.raw_json, '$.metadata.financial.currency')
OR mbr.billed_cost_units != CAST(json_extract(r.raw_json, '$.metadata.financial.cost_charged') AS INTEGER)
)
THEN 1
ELSE 0
END
), 0),
COALESCE(SUM(
CASE
WHEN COALESCE(mbr.reconciliation_state, 'open') = 'reconciled' THEN 1
ELSE 0
END
), 0),
COALESCE(SUM(
CASE
WHEN COALESCE(mbr.reconciliation_state, 'open') NOT IN ('reconciled', 'ignored')
AND (
mbr.receipt_id IS NULL
OR mbr.observed_units > CAST(json_extract(r.raw_json, '$.metadata.governed_transaction.metered_billing.quote.quotedUnits') AS INTEGER)
OR (
json_extract(r.raw_json, '$.metadata.governed_transaction.metered_billing.maxBilledUnits') IS NOT NULL
AND mbr.observed_units > CAST(json_extract(r.raw_json, '$.metadata.governed_transaction.metered_billing.maxBilledUnits') AS INTEGER)
)
OR mbr.billed_cost_currency != json_extract(r.raw_json, '$.metadata.governed_transaction.metered_billing.quote.quotedCost.currency')
OR mbr.billed_cost_units > CAST(json_extract(r.raw_json, '$.metadata.governed_transaction.metered_billing.quote.quotedCost.units') AS INTEGER)
OR (
json_type(r.raw_json, '$.metadata.financial') = 'object'
AND (
mbr.billed_cost_currency != json_extract(r.raw_json, '$.metadata.financial.currency')
OR mbr.billed_cost_units != CAST(json_extract(r.raw_json, '$.metadata.financial.cost_charged') AS INTEGER)
)
)
)
THEN 1
ELSE 0
END
), 0)
FROM chio_tool_receipts r
LEFT JOIN capability_lineage cl ON r.capability_id = cl.capability_id
LEFT JOIN metered_billing_reconciliations mbr ON r.receipt_id = mbr.receipt_id
WHERE json_type(r.raw_json, '$.metadata.governed_transaction.metered_billing') = 'object'
AND (?1 IS NULL OR r.capability_id = ?1)
AND (?2 IS NULL OR r.tool_server = ?2)
AND (?3 IS NULL OR r.tool_name = ?3)
AND (?4 IS NULL OR r.timestamp >= ?4)
AND (?5 IS NULL OR r.timestamp <= ?5)
AND (?6 IS NULL OR COALESCE(r.subject_key, cl.subject_key) = ?6)
"#;
let (
metered_receipts,
evidence_attached_receipts,
missing_evidence_receipts,
over_quoted_units_receipts,
over_max_billed_units_receipts,
over_quoted_cost_receipts,
financial_mismatch_receipts,
reconciled_receipts,
actionable_receipts,
) = self.connection()?.query_row(
summary_sql,
params![
capability_id,
tool_server,
tool_name,
since,
until,
agent_subject
],
|row| {
Ok((
row.get::<_, i64>(0)?.max(0) as u64,
row.get::<_, i64>(1)?.max(0) as u64,
row.get::<_, i64>(2)?.max(0) as u64,
row.get::<_, i64>(3)?.max(0) as u64,
row.get::<_, i64>(4)?.max(0) as u64,
row.get::<_, i64>(5)?.max(0) as u64,
row.get::<_, i64>(6)?.max(0) as u64,
row.get::<_, i64>(7)?.max(0) as u64,
row.get::<_, i64>(8)?.max(0) as u64,
))
},
)?;
Ok(BehavioralFeedMeteredBillingSummary {
metered_receipts,
evidence_attached_receipts,
missing_evidence_receipts,
over_quoted_units_receipts,
over_max_billed_units_receipts,
over_quoted_cost_receipts,
financial_mismatch_receipts,
actionable_receipts,
reconciled_receipts,
})
}
pub(crate) fn load_metered_billing_evidence_record(
&self,
receipt_id: &str,
) -> Result<Option<MeteredBillingEvidenceRecord>, ReceiptStoreError> {
self.connection()?
.query_row(
r#"
SELECT
adapter_kind,
evidence_id,
observed_units,
billed_cost_units,
billed_cost_currency,
evidence_sha256,
recorded_at
FROM metered_billing_reconciliations
WHERE receipt_id = ?1
"#,
params![receipt_id],
|row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, String>(1)?,
row.get::<_, i64>(2)?,
row.get::<_, i64>(3)?,
row.get::<_, String>(4)?,
row.get::<_, Option<String>>(5)?,
row.get::<_, i64>(6)?,
))
},
)
.optional()?
.map(
|(
adapter_kind,
evidence_id,
observed_units,
billed_cost_units,
billed_cost_currency,
evidence_sha256,
recorded_at,
)| {
Ok(MeteredBillingEvidenceRecord {
usage_evidence: chio_core::receipt::MeteredUsageEvidenceReceiptMetadata {
evidence_kind: adapter_kind,
evidence_id,
observed_units: observed_units.max(0) as u64,
evidence_sha256,
},
billed_cost: chio_core::capability::MonetaryAmount {
units: billed_cost_units.max(0) as u64,
currency: billed_cost_currency,
},
recorded_at: recorded_at.max(0) as u64,
})
},
)
.transpose()
}
}