use std::sync::Arc;
use agentstategraph::{CommitOptions, RepoError, Repository};
use agentstategraph_core::IntentCategory;
use chrono::Utc;
use crate::error::PolicyError;
use crate::evaluator;
use crate::external::ExternalEvaluatorRegistry;
use crate::paths;
use crate::selector::Situation;
use crate::types::{ChangeProposal, Decision, ExternalEvaluatorRef, Policy, PolicySignature};
use crate::verifier::SignatureVerifier;
const MAX_CAS_RETRIES: u32 = 32;
macro_rules! evaluation_span {
($name:literal $(, $($fields:tt)*)?) => {
tracing::info_span!(
$name,
asg.policy.decision = tracing::field::Empty,
error.type = tracing::field::Empty,
otel.status_code = tracing::field::Empty,
$($($fields)*)?
)
};
}
fn describe_metrics() {
static DESCRIBED: std::sync::Once = std::sync::Once::new();
DESCRIBED.call_once(|| {
metrics::describe_counter!(
"asg.policy.decisions",
"Policy evaluations, by the decision they reached."
);
metrics::describe_counter!(
"asg.policy.evaluator.errors",
"External policy evaluations that failed, so that their policies were skipped."
);
});
}
fn evaluation(
span: tracing::Span,
evaluate: impl FnOnce() -> Result<Decision, PolicyError>,
) -> Result<Decision, PolicyError> {
let _entered = span.enter();
let result = evaluate();
describe_metrics();
match &result {
Ok(decision) => {
span.record("asg.policy.decision", decision.kind());
metrics::counter!("asg.policy.decisions", "asg.policy.decision" => decision.kind())
.increment(1);
}
Err(e) => {
span.record("error.type", e.code());
span.record("otel.status_code", "error");
}
}
result
}
pub struct PolicyStore {
repo: Arc<Repository>,
prefix: String,
agent_id: String,
verifier: Option<Arc<dyn SignatureVerifier>>,
require_signed: bool,
external_evaluators: Option<Arc<ExternalEvaluatorRegistry>>,
}
impl PolicyStore {
pub fn new(
repo: Arc<Repository>,
prefix: impl Into<String>,
agent_id: impl Into<String>,
) -> Self {
let mut prefix = prefix.into();
if prefix.ends_with('/') {
prefix.pop();
}
Self {
repo,
prefix,
agent_id: agent_id.into(),
verifier: None,
require_signed: false,
external_evaluators: None,
}
}
pub fn with_verifier(mut self, verifier: Arc<dyn SignatureVerifier>) -> Self {
self.verifier = Some(verifier);
self
}
pub fn with_require_signed(mut self, require_signed: bool) -> Self {
self.require_signed = require_signed;
self
}
pub fn with_external_evaluators(mut self, registry: Arc<ExternalEvaluatorRegistry>) -> Self {
self.external_evaluators = Some(registry);
self
}
pub fn prefix(&self) -> &str {
&self.prefix
}
pub fn agent_id(&self) -> &str {
&self.agent_id
}
pub fn propose(&self, ref_name: &str, mut policy: Policy) -> Result<String, PolicyError> {
let normalized = paths::normalize(&policy.path)?;
policy.path = normalized.clone();
policy.version = 1;
policy.proposed_by = self.agent_id.clone();
policy.proposed_at = Utc::now();
policy.ratified_by = None;
policy.ratified_at = None;
policy.ratification_reasoning = None;
policy.supersedes = None;
if policy.active_from.timestamp() == 0 {
policy.active_from = Utc::now();
}
let path = paths::active(&self.prefix, &normalized);
let value = serde_json::to_value(&policy)?;
for _ in 0..MAX_CAS_RETRIES {
let head = self.repo.head(ref_name)?;
if self.exists(ref_name, &normalized)? {
return Err(PolicyError::AlreadyExists(normalized));
}
match self.repo.set_json_cas(
ref_name,
head,
&path,
&value,
self.commit_opts(
IntentCategory::PolicyPropose,
format!("Propose policy {}", policy.handle()),
),
) {
Ok(_) => return Ok(policy.handle()),
Err(RepoError::WriteConflict) => continue,
Err(e) => return Err(e.into()),
}
}
Err(RepoError::WriteConflict.into())
}
pub fn ratify(
&self,
ref_name: &str,
path: &str,
ratifier: &str,
reasoning: &str,
) -> Result<(), PolicyError> {
if ratifier.trim().is_empty() {
return Err(PolicyError::Invalid("ratifier required".into()));
}
let normalized = paths::normalize(path)?;
self.update_active(ref_name, &normalized, |policy| {
if policy.is_ratified() {
return Err(PolicyError::AlreadyRatified(policy.handle()));
}
policy.ratified_by = Some(ratifier.to_string());
policy.ratified_at = Some(Utc::now());
policy.ratification_reasoning = if reasoning.is_empty() {
None
} else {
Some(reasoning.to_string())
};
Ok((
IntentCategory::PolicyRatify,
format!("Ratify policy {} by {}", policy.handle(), ratifier),
))
})
}
pub fn supersede(
&self,
ref_name: &str,
path: &str,
mut new_policy: Policy,
) -> Result<String, PolicyError> {
let normalized = paths::normalize(path)?;
if new_policy.active_from.timestamp() == 0 {
new_policy.active_from = Utc::now();
}
let active_path = paths::active(&self.prefix, &normalized);
for _ in 0..MAX_CAS_RETRIES {
let head = self.repo.head(ref_name)?;
let old = self.load_active(ref_name, &normalized)?;
new_policy.path = normalized.clone();
new_policy.version = old.version + 1;
new_policy.proposed_by = self.agent_id.clone();
new_policy.proposed_at = Utc::now();
new_policy.supersedes = Some(old.handle());
let history_path = paths::historical(&self.prefix, &normalized, old.version);
let old_value = serde_json::to_value(&old)?;
let new_value = serde_json::to_value(&new_policy)?;
let handle = self
.repo
.speculate(ref_name, Some(format!("Supersede {}", normalized)))?;
self.repo.spec_set_json(handle, &history_path, &old_value)?;
self.repo.spec_set_json(handle, &active_path, &new_value)?;
match self.repo.commit_speculation_cas(
handle,
head,
self.commit_opts(
IntentCategory::PolicySupersede,
format!("Supersede {} → {}", old.handle(), new_policy.handle()),
),
) {
Ok(_) => return Ok(new_policy.handle()),
Err(RepoError::WriteConflict) => continue,
Err(e) => return Err(e.into()),
}
}
Err(RepoError::WriteConflict.into())
}
pub fn set_signature(
&self,
ref_name: &str,
path: &str,
signature: PolicySignature,
) -> Result<(), PolicyError> {
let normalized = paths::normalize(path)?;
self.update_active(ref_name, &normalized, |policy| {
policy.signature = Some(signature.clone());
Ok((
IntentCategory::PolicySign,
format!("Sign policy {}", policy.handle()),
))
})
}
pub fn get(
&self,
ref_name: &str,
path: &str,
version: Option<u64>,
) -> Result<Policy, PolicyError> {
let normalized = paths::normalize(path)?;
match version {
None => self.load_active(ref_name, &normalized),
Some(v) => {
let current = self.load_active(ref_name, &normalized)?;
if current.version == v {
return Ok(current);
}
let p = paths::historical(&self.prefix, &normalized, v);
let value = self.repo.get_json(ref_name, &p).map_err(|e| {
if is_path_not_found(&e) {
PolicyError::NotFound(format!("{}@{}", normalized, v))
} else {
e.into()
}
})?;
Ok(serde_json::from_value(value)?)
}
}
}
pub fn list(
&self,
ref_name: &str,
prefix_filter: Option<&str>,
) -> Result<Vec<Policy>, PolicyError> {
self.list_scoped(ref_name, prefix_filter, None)
}
pub fn list_scoped(
&self,
ref_name: &str,
prefix_filter: Option<&str>,
tenant_filter: Option<&str>,
) -> Result<Vec<Policy>, PolicyError> {
let leaves = match self.repo.list_paths(ref_name, &self.prefix, None) {
Ok(v) => v,
Err(e) if is_path_not_found(&e) => return Ok(Vec::new()),
Err(e) => return Err(e.into()),
};
let store_prefix_with_slash = format!("{}/", self.prefix);
let meta_marker = format!("/{}/", paths::META_KEY);
let history_marker = format!("/{}/", paths::HISTORY_KEY);
let mut policy_paths: Vec<String> = Vec::new();
for leaf in &leaves {
let Some(rel) = leaf.strip_prefix(&store_prefix_with_slash) else {
continue;
};
if rel.contains(&history_marker) {
continue;
}
let Some(idx) = rel.find(&meta_marker) else {
continue;
};
let policy_rel = &rel[..idx];
if policy_rel.is_empty() {
continue;
}
if let Some(filter) = prefix_filter {
let filter = filter.trim_start_matches('/');
if !policy_rel.starts_with(filter) {
continue;
}
}
policy_paths.push(policy_rel.to_string());
}
policy_paths.sort();
policy_paths.dedup();
let mut out = Vec::with_capacity(policy_paths.len());
for p in policy_paths {
match self.load_active(ref_name, &p) {
Ok(policy) => {
if !tenant_matches(&policy, tenant_filter) {
continue;
}
out.push(policy);
}
Err(PolicyError::NotFound(_)) => continue,
Err(e) => return Err(e),
}
}
Ok(out)
}
pub fn active(
&self,
ref_name: &str,
prefix_filter: Option<&str>,
) -> Result<Vec<Policy>, PolicyError> {
self.active_scoped(ref_name, prefix_filter, None)
}
pub fn active_scoped(
&self,
ref_name: &str,
prefix_filter: Option<&str>,
tenant_filter: Option<&str>,
) -> Result<Vec<Policy>, PolicyError> {
let now = Utc::now();
Ok(self
.list_scoped(ref_name, prefix_filter, tenant_filter)?
.into_iter()
.filter(|p| p.is_currently_active(now))
.filter(|p| self.signature_passes(p))
.collect())
}
fn signature_passes(&self, policy: &Policy) -> bool {
let Some(verifier) = self.verifier.as_ref() else {
return true;
};
match policy.signature.as_ref() {
None => !self.require_signed,
Some(_) => verifier.verify_policy(policy).is_ok(),
}
}
pub fn history(&self, ref_name: &str, path: &str) -> Result<Vec<Policy>, PolicyError> {
let normalized = paths::normalize(path)?;
let current = self.load_active(ref_name, &normalized)?;
let mut versions = vec![current.clone()];
let mut v = current.version;
while v > 1 {
v -= 1;
let hp = paths::historical(&self.prefix, &normalized, v);
let value = self.repo.get_json(ref_name, &hp).map_err(|e| {
if is_path_not_found(&e) {
PolicyError::NotFound(format!("{}@{}", normalized, v))
} else {
PolicyError::from(e)
}
})?;
let policy: Policy = serde_json::from_value(value)?;
versions.push(policy);
}
versions.reverse();
Ok(versions)
}
pub fn policies_for_situation(
&self,
ref_name: &str,
situation: &Situation,
) -> Result<Vec<Policy>, PolicyError> {
self.policies_for_situation_scoped(ref_name, situation, None)
}
pub fn policies_for_situation_scoped(
&self,
ref_name: &str,
situation: &Situation,
tenant_filter: Option<&str>,
) -> Result<Vec<Policy>, PolicyError> {
Ok(self
.active_scoped(ref_name, None, tenant_filter)?
.into_iter()
.filter(|p| p.situation_selector.matches(situation))
.collect())
}
pub fn evaluate(
&self,
ref_name: &str,
situation: &Situation,
action: &str,
agent_id: &str,
) -> Result<Decision, PolicyError> {
self.evaluate_scoped(ref_name, situation, action, agent_id, None)
}
pub fn evaluate_scoped(
&self,
ref_name: &str,
situation: &Situation,
action: &str,
agent_id: &str,
tenant_filter: Option<&str>,
) -> Result<Decision, PolicyError> {
evaluation(evaluation_span!("asg.policy.evaluate"), || {
let matched = self.policies_for_situation_scoped(ref_name, situation, tenant_filter)?;
let (locals, externals) = self.partition_for_dispatch(&matched);
let external_decisions = self.run_externals(&externals, situation, action, agent_id);
let local_decision = evaluator::evaluate_matched(&locals, action, agent_id);
Ok(combine_decisions(external_decisions, local_decision))
})
}
pub fn evaluate_change(
&self,
ref_name: &str,
proposal: &ChangeProposal,
) -> Result<Decision, PolicyError> {
self.evaluate_change_scoped(ref_name, proposal, None)
}
pub fn evaluate_change_scoped(
&self,
ref_name: &str,
proposal: &ChangeProposal,
tenant_filter: Option<&str>,
) -> Result<Decision, PolicyError> {
evaluation(evaluation_span!("asg.policy.evaluate_change"), || {
let actives = self.active_scoped(ref_name, None, tenant_filter)?;
let (locals, externals) = self.partition_for_dispatch(&actives);
let empty_situation = Situation::new();
let external_decisions = self.run_externals(
&externals,
&empty_situation,
&proposal.action,
&proposal.agent_id,
);
let local_decision = evaluator::evaluate_change(&locals, proposal);
Ok(combine_decisions(external_decisions, local_decision))
})
}
fn partition_for_dispatch<'a>(
&self,
policies: &'a [Policy],
) -> (Vec<&'a Policy>, Vec<&'a Policy>) {
let Some(registry) = self.external_evaluators.as_ref() else {
return (policies.iter().collect(), Vec::new());
};
let mut locals = Vec::with_capacity(policies.len());
let mut externals = Vec::new();
for p in policies {
match p.external_evaluator.as_ref() {
None => locals.push(p),
Some(ref_) => {
if registry.get(external_kind(ref_)).is_some() {
externals.push(p);
}
}
}
}
(locals, externals)
}
fn run_externals(
&self,
policies: &[&Policy],
situation: &Situation,
action: &str,
agent_id: &str,
) -> Vec<Decision> {
let Some(registry) = self.external_evaluators.as_ref() else {
return Vec::new();
};
let mut out = Vec::with_capacity(policies.len());
for p in policies {
let Some(ref_) = p.external_evaluator.as_ref() else {
continue;
};
let Some(runner) = registry.get(external_kind(ref_)) else {
continue;
};
let source = external_source(ref_);
let kind = runner.kind();
let span = evaluation_span!("asg.policy.external", asg.policy.evaluator = kind);
let _entered = span.enter();
match runner.evaluate(source, situation, action, agent_id) {
Ok(d) => {
span.record("asg.policy.decision", d.kind());
out.push(d);
}
Err(e) => {
span.record("error.type", e.code());
span.record("otel.status_code", "error");
describe_metrics();
metrics::counter!(
"asg.policy.evaluator.errors",
"asg.policy.evaluator" => kind,
"error.type" => e.code()
)
.increment(1);
tracing::warn!(
asg.policy.evaluator = kind,
asg.policy.id = %p.handle(),
error.type = e.code(),
"an external policy evaluator failed, so its policy was skipped"
);
}
}
}
out
}
fn exists(&self, ref_name: &str, normalized: &str) -> Result<bool, PolicyError> {
match self.load_active(ref_name, normalized) {
Ok(_) => Ok(true),
Err(PolicyError::NotFound(_)) => Ok(false),
Err(e) => Err(e),
}
}
fn update_active<F>(
&self,
ref_name: &str,
normalized: &str,
mut change: F,
) -> Result<(), PolicyError>
where
F: FnMut(&mut Policy) -> Result<(IntentCategory, String), PolicyError>,
{
let active_path = paths::active(&self.prefix, normalized);
for _ in 0..MAX_CAS_RETRIES {
let head = self.repo.head(ref_name)?;
let mut policy = self.load_active(ref_name, normalized)?;
let (category, description) = change(&mut policy)?;
let value = serde_json::to_value(&policy)?;
match self.repo.set_json_cas(
ref_name,
head,
&active_path,
&value,
self.commit_opts(category, description),
) {
Ok(_) => return Ok(()),
Err(RepoError::WriteConflict) => continue,
Err(e) => return Err(e.into()),
}
}
Err(RepoError::WriteConflict.into())
}
fn load_active(&self, ref_name: &str, normalized: &str) -> Result<Policy, PolicyError> {
let path = paths::active(&self.prefix, normalized);
let value = self.repo.get_json(ref_name, &path).map_err(|e| {
if is_path_not_found(&e) {
PolicyError::NotFound(normalized.to_string())
} else {
PolicyError::from(e)
}
})?;
Ok(serde_json::from_value(value)?)
}
fn commit_opts(
&self,
category: IntentCategory,
description: impl Into<String>,
) -> CommitOptions {
CommitOptions::new(&self.agent_id, category, description)
}
}
fn is_path_not_found(e: &agentstategraph::RepoError) -> bool {
matches!(e, agentstategraph::RepoError::Tree(_))
}
fn external_kind(ref_: &ExternalEvaluatorRef) -> &'static str {
match ref_ {
ExternalEvaluatorRef::Rego { .. } => "rego",
ExternalEvaluatorRef::Cedar { .. } => "cedar",
ExternalEvaluatorRef::Wasm { .. } => "wasm",
}
}
fn external_source(ref_: &ExternalEvaluatorRef) -> &crate::types::EvaluatorSource {
match ref_ {
ExternalEvaluatorRef::Rego { source }
| ExternalEvaluatorRef::Cedar { source }
| ExternalEvaluatorRef::Wasm { source } => source,
}
}
fn combine_decisions(externals: Vec<Decision>, local: Decision) -> Decision {
let all: Vec<Decision> = externals
.into_iter()
.chain(std::iter::once(local))
.collect();
for d in &all {
if matches!(d, Decision::Deny { .. }) {
return d.clone();
}
}
for d in &all {
if matches!(d, Decision::RequireApproval { .. }) {
return d.clone();
}
}
for d in &all {
if matches!(d, Decision::Allow { .. }) {
return d.clone();
}
}
Decision::NoPolicyMatch
}
fn tenant_matches(policy: &Policy, tenant_filter: Option<&str>) -> bool {
match tenant_filter {
None => true,
Some(tid) => match policy.tenant_id.as_deref() {
None => true,
Some(p_tid) => p_tid == tid,
},
}
}