use std::collections::{BTreeMap, BTreeSet};
use serde::{Deserialize, Serialize};
use thiserror::Error;
use uuid::Uuid;
use crate::{
ActionClass, AgentAuthority, AgentBudget, AgentError, AgentRole, DelegationPlan,
SCHEMA_VERSION, TaskContract,
agent::{scope_contains, valid_digest, validate_scope_path},
hash_json,
};
pub const AGENT_GRAPH_PROTOCOL_VERSION: &str = "proofborne.agent-graph.v1";
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct AgentGraphBudget {
pub max_nodes: usize,
pub max_concurrency: usize,
pub max_provider_turns: usize,
pub max_tool_calls: u64,
pub max_duration_ms: u64,
}
impl AgentGraphBudget {
pub fn validate(&self) -> Result<(), AgentGraphError> {
if self.max_nodes == 0
|| self.max_concurrency == 0
|| self.max_concurrency > self.max_nodes
|| self.max_provider_turns == 0
|| self.max_tool_calls == 0
|| self.max_duration_ms == 0
{
return Err(AgentGraphError::InvalidGraphBudget);
}
Ok(())
}
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct AgentAuthorityTemplate {
pub profile: String,
pub provider: String,
pub model: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub credential_id: Option<String>,
}
impl AgentAuthorityTemplate {
fn bind(&self, role: AgentRole) -> Result<AgentAuthority, AgentGraphError> {
AgentAuthority::new(
role,
self.profile.clone(),
self.provider.clone(),
self.model.clone(),
self.credential_id.clone(),
)
.map_err(AgentGraphError::Delegation)
}
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct AgentGraphNodeTemplate {
pub node_id: String,
pub role: AgentRole,
#[serde(default)]
pub dependencies: BTreeSet<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub review_target_node_id: Option<String>,
#[serde(default)]
pub delegated_criterion_ids: BTreeSet<String>,
#[serde(default)]
pub added_constraints: Vec<String>,
pub authority: AgentAuthorityTemplate,
pub read_paths: BTreeSet<String>,
#[serde(default)]
pub write_paths: BTreeSet<String>,
pub budget: AgentBudget,
pub allowed_action_classes: BTreeSet<ActionClass>,
}
impl AgentGraphNodeTemplate {
fn bind(&self) -> Result<AgentGraphNode, AgentGraphError> {
Ok(AgentGraphNode {
node_id: self.node_id.clone(),
role: self.role,
dependencies: self.dependencies.clone(),
review_target_node_id: self.review_target_node_id.clone(),
delegated_criterion_ids: self.delegated_criterion_ids.clone(),
added_constraints: self.added_constraints.clone(),
authority: self.authority.bind(self.role)?,
read_paths: self.read_paths.clone(),
write_paths: self.write_paths.clone(),
budget: self.budget.clone(),
allowed_action_classes: self.allowed_action_classes.clone(),
})
}
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct AgentGraphTemplate {
pub schema_version: String,
pub protocol_version: String,
pub coordinator_authority: AgentAuthorityTemplate,
pub budget: AgentGraphBudget,
pub nodes: Vec<AgentGraphNodeTemplate>,
}
impl AgentGraphTemplate {
pub fn bind(
&self,
parent_session_id: Uuid,
parent: &TaskContract,
parent_state_binding: impl Into<String>,
) -> Result<AgentGraphPlan, AgentGraphError> {
if self.schema_version != SCHEMA_VERSION {
return Err(AgentGraphError::UnsupportedSchema(
self.schema_version.clone(),
));
}
if self.protocol_version != AGENT_GRAPH_PROTOCOL_VERSION {
return Err(AgentGraphError::UnsupportedProtocol(
self.protocol_version.clone(),
));
}
let nodes = self
.nodes
.iter()
.map(AgentGraphNodeTemplate::bind)
.collect::<Result<Vec<_>, _>>()?;
AgentGraphPlan::new(
parent_session_id,
parent,
parent_state_binding,
self.coordinator_authority.bind(AgentRole::Coordinator)?,
self.budget.clone(),
nodes,
)
}
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct AgentGraphNode {
pub node_id: String,
pub role: AgentRole,
#[serde(default)]
pub dependencies: BTreeSet<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub review_target_node_id: Option<String>,
#[serde(default)]
pub delegated_criterion_ids: BTreeSet<String>,
#[serde(default)]
pub added_constraints: Vec<String>,
pub authority: AgentAuthority,
pub read_paths: BTreeSet<String>,
#[serde(default)]
pub write_paths: BTreeSet<String>,
pub budget: AgentBudget,
pub allowed_action_classes: BTreeSet<ActionClass>,
}
impl AgentGraphNode {
pub fn derive_delegation(
&self,
parent_session_id: Uuid,
parent: &TaskContract,
input_workspace_generation: impl Into<String>,
leased_input_hash: impl Into<String>,
) -> Result<DelegationPlan, AgentGraphError> {
if self.role != AgentRole::Worker {
return Err(AgentGraphError::NodeRole(self.node_id.clone()));
}
let input_workspace_generation = input_workspace_generation.into();
let lease = crate::WorkspaceLease {
input_workspace_generation: input_workspace_generation.clone(),
leased_input_hash: leased_input_hash.into(),
read_paths: self.read_paths.clone(),
write_paths: self.write_paths.clone(),
};
DelegationPlan::derive(
parent_session_id,
parent,
self.delegated_criterion_ids.clone(),
self.added_constraints.clone(),
self.authority.clone(),
lease,
self.budget.clone(),
self.allowed_action_classes.clone(),
input_workspace_generation,
)
.map_err(AgentGraphError::Delegation)
}
fn validate(&self) -> Result<(), AgentGraphError> {
if !valid_node_id(&self.node_id) {
return Err(AgentGraphError::InvalidNodeId(self.node_id.clone()));
}
self.authority
.validate()
.map_err(AgentGraphError::Delegation)?;
self.budget
.validate()
.map_err(AgentGraphError::Delegation)?;
if self.authority.role != self.role || self.role == AgentRole::Coordinator {
return Err(AgentGraphError::NodeRole(self.node_id.clone()));
}
if self.read_paths.is_empty() || self.allowed_action_classes.is_empty() {
return Err(AgentGraphError::NodeAuthority(self.node_id.clone()));
}
for path in self.read_paths.iter().chain(&self.write_paths) {
validate_scope_path(path).map_err(AgentGraphError::Delegation)?;
}
for write_path in &self.write_paths {
if !self
.read_paths
.iter()
.any(|read_path| scope_contains(read_path, write_path))
{
return Err(AgentGraphError::WriteOutsideReadScope {
node_id: self.node_id.clone(),
path: write_path.clone(),
});
}
}
match self.role {
AgentRole::Worker => {
if self.review_target_node_id.is_some()
|| self.delegated_criterion_ids.is_empty()
|| self.allowed_action_classes.contains(&ActionClass::Delegate)
|| self
.allowed_action_classes
.contains(&ActionClass::Sensitive)
{
return Err(AgentGraphError::NodeAuthority(self.node_id.clone()));
}
}
AgentRole::Reviewer | AgentRole::Adversary => {
if self.review_target_node_id.is_none()
|| !self.delegated_criterion_ids.is_empty()
|| !self.added_constraints.is_empty()
|| !self.write_paths.is_empty()
|| !self.allowed_action_classes.contains(&ActionClass::Read)
|| self
.allowed_action_classes
.iter()
.any(|action| !matches!(action, ActionClass::Read | ActionClass::Diagnose))
{
return Err(AgentGraphError::NodeAuthority(self.node_id.clone()));
}
}
AgentRole::Coordinator => {
return Err(AgentGraphError::NodeRole(self.node_id.clone()));
}
}
Ok(())
}
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct AgentGraphPlan {
pub schema_version: String,
pub protocol_version: String,
pub graph_id: Uuid,
pub parent_session_id: Uuid,
pub parent_contract_id: Uuid,
pub parent_contract_hash: String,
pub parent_state_binding: String,
pub coordinator_authority: AgentAuthority,
pub budget: AgentGraphBudget,
pub nodes: Vec<AgentGraphNode>,
}
impl AgentGraphPlan {
#[allow(clippy::too_many_arguments)]
pub fn new(
parent_session_id: Uuid,
parent: &TaskContract,
parent_state_binding: impl Into<String>,
coordinator_authority: AgentAuthority,
budget: AgentGraphBudget,
mut nodes: Vec<AgentGraphNode>,
) -> Result<Self, AgentGraphError> {
nodes.sort_by(|left, right| left.node_id.cmp(&right.node_id));
let plan = Self {
schema_version: SCHEMA_VERSION.to_owned(),
protocol_version: AGENT_GRAPH_PROTOCOL_VERSION.to_owned(),
graph_id: Uuid::now_v7(),
parent_session_id,
parent_contract_id: parent.id,
parent_contract_hash: hash_json(
&serde_json::to_value(parent).map_err(|_| AgentGraphError::Serialization)?,
),
parent_state_binding: parent_state_binding.into(),
coordinator_authority,
budget,
nodes,
};
plan.validate(parent)?;
Ok(plan)
}
pub fn validate(&self, parent: &TaskContract) -> Result<(), AgentGraphError> {
if self.schema_version != SCHEMA_VERSION {
return Err(AgentGraphError::UnsupportedSchema(
self.schema_version.clone(),
));
}
if self.protocol_version != AGENT_GRAPH_PROTOCOL_VERSION {
return Err(AgentGraphError::UnsupportedProtocol(
self.protocol_version.clone(),
));
}
parent
.validate()
.map_err(|_| AgentGraphError::InvalidParentContract)?;
if !parent.confirmed
|| self.parent_contract_id != parent.id
|| self.parent_contract_hash
!= hash_json(
&serde_json::to_value(parent).map_err(|_| AgentGraphError::Serialization)?,
)
|| !valid_digest(&self.parent_state_binding)
{
return Err(AgentGraphError::ParentBinding);
}
self.coordinator_authority
.validate()
.map_err(AgentGraphError::Delegation)?;
if self.coordinator_authority.role != AgentRole::Coordinator {
return Err(AgentGraphError::CoordinatorRole);
}
self.budget.validate()?;
if self.nodes.is_empty() || self.nodes.len() > self.budget.max_nodes {
return Err(AgentGraphError::NodeLimit);
}
if self
.nodes
.windows(2)
.any(|nodes| nodes[0].node_id >= nodes[1].node_id)
{
return Err(AgentGraphError::NonCanonicalNodeOrder);
}
let nodes: BTreeMap<&str, &AgentGraphNode> = self
.nodes
.iter()
.map(|node| (node.node_id.as_str(), node))
.collect();
if nodes.len() != self.nodes.len() {
return Err(AgentGraphError::DuplicateNodeId);
}
let mut authority_ids = BTreeSet::from([self.coordinator_authority.agent_id]);
let mut authority_hashes =
BTreeSet::from([self.coordinator_authority.authority_hash.as_str()]);
let mut provider_turns = 0usize;
let mut tool_calls = 0u64;
for node in &self.nodes {
node.validate()?;
if !authority_ids.insert(node.authority.agent_id)
|| !authority_hashes.insert(node.authority.authority_hash.as_str())
{
return Err(AgentGraphError::AuthorityReused(node.node_id.clone()));
}
provider_turns = provider_turns
.checked_add(node.budget.max_provider_turns)
.ok_or(AgentGraphError::GraphBudgetExceeded)?;
tool_calls = tool_calls
.checked_add(node.budget.max_tool_calls)
.ok_or(AgentGraphError::GraphBudgetExceeded)?;
if node.budget.max_duration_ms > self.budget.max_duration_ms {
return Err(AgentGraphError::GraphBudgetExceeded);
}
for dependency in &node.dependencies {
if dependency == &node.node_id || !nodes.contains_key(dependency.as_str()) {
return Err(AgentGraphError::UnknownDependency {
node_id: node.node_id.clone(),
dependency: dependency.clone(),
});
}
}
}
if provider_turns > self.budget.max_provider_turns
|| tool_calls > self.budget.max_tool_calls
{
return Err(AgentGraphError::GraphBudgetExceeded);
}
self.topological_order()?;
self.validate_review_barriers(&nodes)?;
self.validate_decomposition(parent)?;
self.validate_sibling_write_scopes(&nodes)?;
Ok(())
}
pub fn digest(&self, parent: &TaskContract) -> Result<String, AgentGraphError> {
self.validate(parent)?;
Ok(hash_json(
&serde_json::to_value(self).map_err(|_| AgentGraphError::Serialization)?,
))
}
pub fn topological_order(&self) -> Result<Vec<String>, AgentGraphError> {
let mut remaining: BTreeMap<&str, BTreeSet<&str>> = self
.nodes
.iter()
.map(|node| {
(
node.node_id.as_str(),
node.dependencies.iter().map(String::as_str).collect(),
)
})
.collect();
let mut order = Vec::with_capacity(remaining.len());
while !remaining.is_empty() {
let ready: Vec<&str> = remaining
.iter()
.filter_map(|(node_id, dependencies)| dependencies.is_empty().then_some(*node_id))
.collect();
if ready.is_empty() {
return Err(AgentGraphError::Cycle);
}
for node_id in ready {
remaining.remove(node_id);
for dependencies in remaining.values_mut() {
dependencies.remove(node_id);
}
order.push(node_id.to_owned());
}
}
Ok(order)
}
pub fn node(&self, node_id: &str) -> Option<&AgentGraphNode> {
self.nodes
.binary_search_by_key(&node_id, |node| node.node_id.as_str())
.ok()
.map(|index| &self.nodes[index])
}
fn validate_review_barriers(
&self,
nodes: &BTreeMap<&str, &AgentGraphNode>,
) -> Result<(), AgentGraphError> {
let mut reviewers: BTreeMap<&str, Vec<&str>> = BTreeMap::new();
let mut adversaries: BTreeMap<&str, Vec<&str>> = BTreeMap::new();
for node in &self.nodes {
match node.role {
AgentRole::Worker => {
if node.dependencies.iter().any(|dependency| {
nodes
.get(dependency.as_str())
.is_some_and(|dependency| dependency.role == AgentRole::Worker)
}) {
return Err(AgentGraphError::IncompleteProofBarrier(
node.node_id.clone(),
));
}
}
AgentRole::Reviewer | AgentRole::Adversary => {
let target = node
.review_target_node_id
.as_deref()
.ok_or_else(|| AgentGraphError::ReviewTarget(node.node_id.clone()))?;
if node.dependencies != BTreeSet::from([target.to_owned()])
|| nodes
.get(target)
.is_none_or(|target| target.role != AgentRole::Worker)
{
return Err(AgentGraphError::ReviewTarget(node.node_id.clone()));
}
let collection = if node.role == AgentRole::Reviewer {
&mut reviewers
} else {
&mut adversaries
};
collection.entry(target).or_default().push(&node.node_id);
}
AgentRole::Coordinator => {
return Err(AgentGraphError::NodeRole(node.node_id.clone()));
}
}
}
for worker in self
.nodes
.iter()
.filter(|node| node.role == AgentRole::Worker)
{
if reviewers.get(worker.node_id.as_str()).map(Vec::len) != Some(1)
|| adversaries.get(worker.node_id.as_str()).map(Vec::len) != Some(1)
{
return Err(AgentGraphError::IncompleteProofBarrier(
worker.node_id.clone(),
));
}
}
for worker in self
.nodes
.iter()
.filter(|node| node.role == AgentRole::Worker && !node.dependencies.is_empty())
{
let mut target_workers = BTreeSet::new();
for dependency in &worker.dependencies {
let gate = nodes.get(dependency.as_str()).ok_or_else(|| {
AgentGraphError::UnknownDependency {
node_id: worker.node_id.clone(),
dependency: dependency.clone(),
}
})?;
if !matches!(gate.role, AgentRole::Reviewer | AgentRole::Adversary) {
return Err(AgentGraphError::IncompleteProofBarrier(
worker.node_id.clone(),
));
}
target_workers.insert(
gate.review_target_node_id
.as_deref()
.ok_or_else(|| AgentGraphError::ReviewTarget(gate.node_id.clone()))?,
);
}
for target in target_workers {
let reviewer = reviewers
.get(target)
.and_then(|values| values.first())
.copied()
.ok_or_else(|| AgentGraphError::IncompleteProofBarrier(target.to_owned()))?;
let adversary = adversaries
.get(target)
.and_then(|values| values.first())
.copied()
.ok_or_else(|| AgentGraphError::IncompleteProofBarrier(target.to_owned()))?;
if !worker.dependencies.contains(reviewer)
|| !worker.dependencies.contains(adversary)
{
return Err(AgentGraphError::IncompleteProofBarrier(
worker.node_id.clone(),
));
}
}
}
Ok(())
}
fn validate_decomposition(&self, parent: &TaskContract) -> Result<(), AgentGraphError> {
let expected: BTreeSet<&str> = parent
.criteria
.iter()
.map(|criterion| criterion.id.as_str())
.collect();
let mut observed = BTreeSet::new();
for node in self
.nodes
.iter()
.filter(|node| node.role == AgentRole::Worker)
{
for criterion_id in &node.delegated_criterion_ids {
if !expected.contains(criterion_id.as_str()) {
return Err(AgentGraphError::UnknownCriterion(criterion_id.clone()));
}
if !observed.insert(criterion_id.as_str()) {
return Err(AgentGraphError::CriterionAssignedTwice(
criterion_id.clone(),
));
}
}
}
if observed != expected {
return Err(AgentGraphError::IncompleteDecomposition);
}
Ok(())
}
fn validate_sibling_write_scopes(
&self,
nodes: &BTreeMap<&str, &AgentGraphNode>,
) -> Result<(), AgentGraphError> {
let workers: Vec<&AgentGraphNode> = self
.nodes
.iter()
.filter(|node| node.role == AgentRole::Worker)
.collect();
for (index, left) in workers.iter().enumerate() {
for right in workers.iter().skip(index + 1) {
if depends_on(nodes, left.node_id.as_str(), right.node_id.as_str())
|| depends_on(nodes, right.node_id.as_str(), left.node_id.as_str())
{
continue;
}
for left_scope in &left.write_paths {
for right_scope in &right.write_paths {
if scope_contains(left_scope, right_scope)
|| scope_contains(right_scope, left_scope)
{
return Err(AgentGraphError::SiblingWriteConflict {
left_node_id: left.node_id.clone(),
right_node_id: right.node_id.clone(),
left_scope: left_scope.clone(),
right_scope: right_scope.clone(),
});
}
}
}
}
}
Ok(())
}
}
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord)]
#[serde(rename_all = "snake_case")]
pub enum AgentNodeState {
Pending,
Running,
HandoffReady,
AwaitingMerge,
Succeeded,
Blocked,
Failed,
Cancelled,
Invalidated,
}
impl AgentNodeState {
pub const fn is_terminal(self) -> bool {
matches!(
self,
Self::Succeeded | Self::Blocked | Self::Failed | Self::Cancelled | Self::Invalidated
)
}
}
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum SchedulerDecisionKind {
Start,
HandoffRecorded,
ReviewRecorded,
MergeCommitted,
Block,
Fail,
Cancel,
Invalidate,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct SchedulerDecision {
pub schema_version: String,
pub protocol_version: String,
pub graph_id: Uuid,
pub sequence: u64,
pub node_id: String,
pub kind: SchedulerDecisionKind,
pub from_state: AgentNodeState,
pub to_state: AgentNodeState,
#[serde(skip_serializing_if = "Option::is_none")]
pub session_id: Option<Uuid>,
#[serde(skip_serializing_if = "Option::is_none")]
pub delegation_id: Option<Uuid>,
#[serde(skip_serializing_if = "Option::is_none")]
pub receipt_hash: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub workspace_generation: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub reason_code: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub previous_hash: Option<String>,
pub hash: String,
}
impl SchedulerDecision {
#[allow(clippy::too_many_arguments)]
pub fn new(
graph_id: Uuid,
sequence: u64,
node_id: impl Into<String>,
kind: SchedulerDecisionKind,
from_state: AgentNodeState,
to_state: AgentNodeState,
session_id: Option<Uuid>,
delegation_id: Option<Uuid>,
receipt_hash: Option<String>,
workspace_generation: Option<String>,
reason_code: Option<String>,
previous_hash: Option<String>,
) -> Result<Self, AgentGraphError> {
let mut decision = Self {
schema_version: SCHEMA_VERSION.to_owned(),
protocol_version: AGENT_GRAPH_PROTOCOL_VERSION.to_owned(),
graph_id,
sequence,
node_id: node_id.into(),
kind,
from_state,
to_state,
session_id,
delegation_id,
receipt_hash,
workspace_generation,
reason_code,
previous_hash,
hash: String::new(),
};
decision.validate_material()?;
decision.hash = decision.material_digest();
Ok(decision)
}
pub fn validate(&self) -> Result<(), AgentGraphError> {
self.validate_material()?;
if !valid_digest(&self.hash) || self.hash != self.material_digest() {
return Err(AgentGraphError::DecisionDigest);
}
Ok(())
}
fn validate_material(&self) -> Result<(), AgentGraphError> {
if self.schema_version != SCHEMA_VERSION {
return Err(AgentGraphError::UnsupportedSchema(
self.schema_version.clone(),
));
}
if self.protocol_version != AGENT_GRAPH_PROTOCOL_VERSION {
return Err(AgentGraphError::UnsupportedProtocol(
self.protocol_version.clone(),
));
}
if !valid_node_id(&self.node_id)
|| self
.previous_hash
.as_deref()
.is_some_and(|value| !valid_digest(value))
|| self
.receipt_hash
.as_deref()
.is_some_and(|value| !valid_digest(value))
|| self
.workspace_generation
.as_deref()
.is_some_and(|value| !valid_digest(value))
|| self
.reason_code
.as_deref()
.is_some_and(|value| value.trim().is_empty())
{
return Err(AgentGraphError::InvalidDecision);
}
let valid = match self.kind {
SchedulerDecisionKind::Start => {
self.from_state == AgentNodeState::Pending
&& self.to_state == AgentNodeState::Running
&& self.session_id.is_none()
&& self.receipt_hash.is_none()
&& self.workspace_generation.is_some()
&& self.reason_code.is_none()
}
SchedulerDecisionKind::HandoffRecorded => {
self.from_state == AgentNodeState::Running
&& self.to_state == AgentNodeState::HandoffReady
&& self.session_id.is_some()
&& self.delegation_id.is_some()
&& self.receipt_hash.is_some()
&& self.workspace_generation.is_some()
&& self.reason_code.is_none()
}
SchedulerDecisionKind::ReviewRecorded => {
self.from_state == AgentNodeState::Running
&& self.to_state == AgentNodeState::AwaitingMerge
&& self.session_id.is_some()
&& self.delegation_id.is_some()
&& self.receipt_hash.is_some()
&& self.workspace_generation.is_some()
&& self.reason_code.is_none()
}
SchedulerDecisionKind::MergeCommitted => {
matches!(
self.from_state,
AgentNodeState::HandoffReady | AgentNodeState::AwaitingMerge
) && self.to_state == AgentNodeState::Succeeded
&& self.session_id.is_none()
&& self.delegation_id.is_some()
&& self.receipt_hash.is_some()
&& self.workspace_generation.is_some()
&& self
.reason_code
.as_deref()
.is_none_or(|reason| reason == "recovered_persisted_merge")
}
SchedulerDecisionKind::Block => {
!self.from_state.is_terminal()
&& self.to_state == AgentNodeState::Blocked
&& self.reason_code.is_some()
}
SchedulerDecisionKind::Fail => {
!self.from_state.is_terminal()
&& self.to_state == AgentNodeState::Failed
&& self.reason_code.is_some()
}
SchedulerDecisionKind::Cancel => {
!self.from_state.is_terminal()
&& self.to_state == AgentNodeState::Cancelled
&& self.reason_code.is_some()
}
SchedulerDecisionKind::Invalidate => {
!self.from_state.is_terminal()
&& self.to_state == AgentNodeState::Invalidated
&& self.reason_code.is_some()
}
};
if !valid {
return Err(AgentGraphError::InvalidTransition);
}
Ok(())
}
fn material_digest(&self) -> String {
hash_json(&serde_json::json!({
"schemaVersion": self.schema_version,
"protocolVersion": self.protocol_version,
"graphId": self.graph_id,
"sequence": self.sequence,
"nodeId": self.node_id,
"kind": self.kind,
"fromState": self.from_state,
"toState": self.to_state,
"sessionId": self.session_id,
"delegationId": self.delegation_id,
"receiptHash": self.receipt_hash,
"workspaceGeneration": self.workspace_generation,
"reasonCode": self.reason_code,
"previousHash": self.previous_hash,
}))
}
}
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum AgentGraphOutcome {
Verified,
Blocked,
Failed,
Cancelled,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct SchedulerReceipt {
pub schema_version: String,
pub protocol_version: String,
pub graph_hash: String,
pub decisions: Vec<SchedulerDecision>,
pub terminal_node_states: BTreeMap<String, AgentNodeState>,
pub outcome: AgentGraphOutcome,
pub final_workspace_generation: String,
}
impl SchedulerReceipt {
pub fn validate(
&self,
graph: &AgentGraphPlan,
parent: &TaskContract,
) -> Result<(), AgentGraphError> {
graph.validate(parent)?;
if self.schema_version != SCHEMA_VERSION {
return Err(AgentGraphError::UnsupportedSchema(
self.schema_version.clone(),
));
}
if self.protocol_version != AGENT_GRAPH_PROTOCOL_VERSION {
return Err(AgentGraphError::UnsupportedProtocol(
self.protocol_version.clone(),
));
}
if self.graph_hash != graph.digest(parent)?
|| !valid_digest(&self.final_workspace_generation)
{
return Err(AgentGraphError::ReceiptBinding);
}
let mut states: BTreeMap<String, AgentNodeState> = graph
.nodes
.iter()
.map(|node| (node.node_id.clone(), AgentNodeState::Pending))
.collect();
let mut previous_hash = None;
for (sequence, decision) in self.decisions.iter().enumerate() {
decision.validate()?;
if decision.graph_id != graph.graph_id
|| decision.sequence != sequence as u64
|| decision.previous_hash != previous_hash
{
return Err(AgentGraphError::DecisionChain);
}
let node = graph
.node(&decision.node_id)
.ok_or_else(|| AgentGraphError::InvalidNodeId(decision.node_id.clone()))?;
let kind_matches_role = match decision.kind {
SchedulerDecisionKind::HandoffRecorded => node.role == AgentRole::Worker,
SchedulerDecisionKind::ReviewRecorded => {
matches!(node.role, AgentRole::Reviewer | AgentRole::Adversary)
}
_ => true,
};
if !kind_matches_role {
return Err(AgentGraphError::InvalidTransition);
}
if decision.kind == SchedulerDecisionKind::Start {
let ready = match node.role {
AgentRole::Worker => node.dependencies.iter().all(|dependency| {
states.get(dependency) == Some(&AgentNodeState::Succeeded)
}),
AgentRole::Reviewer | AgentRole::Adversary => {
node.review_target_node_id.as_ref().is_some_and(|target| {
states.get(target) == Some(&AgentNodeState::HandoffReady)
})
}
AgentRole::Coordinator => false,
};
if !ready {
return Err(AgentGraphError::DecisionChain);
}
}
let state = states
.get_mut(&decision.node_id)
.ok_or_else(|| AgentGraphError::InvalidNodeId(decision.node_id.clone()))?;
if *state != decision.from_state {
return Err(AgentGraphError::DecisionChain);
}
*state = decision.to_state;
previous_hash = Some(decision.hash.clone());
}
let expected_final_generation = self
.decisions
.iter()
.rev()
.find(|decision| decision.kind == SchedulerDecisionKind::MergeCommitted)
.and_then(|decision| decision.workspace_generation.as_deref())
.unwrap_or(graph.parent_state_binding.as_str());
if self.final_workspace_generation != expected_final_generation {
return Err(AgentGraphError::ReceiptBinding);
}
if states != self.terminal_node_states || states.values().any(|state| !state.is_terminal())
{
return Err(AgentGraphError::TerminalStates);
}
let expected = if states
.values()
.all(|state| *state == AgentNodeState::Succeeded)
{
AgentGraphOutcome::Verified
} else if states
.values()
.any(|state| *state == AgentNodeState::Cancelled)
{
AgentGraphOutcome::Cancelled
} else if states
.values()
.any(|state| matches!(state, AgentNodeState::Failed | AgentNodeState::Invalidated))
{
AgentGraphOutcome::Failed
} else {
AgentGraphOutcome::Blocked
};
if self.outcome != expected {
return Err(AgentGraphError::TerminalStates);
}
Ok(())
}
pub fn digest(
&self,
graph: &AgentGraphPlan,
parent: &TaskContract,
) -> Result<String, AgentGraphError> {
self.validate(graph, parent)?;
Ok(hash_json(
&serde_json::to_value(self).map_err(|_| AgentGraphError::Serialization)?,
))
}
}
fn depends_on(
nodes: &BTreeMap<&str, &AgentGraphNode>,
node_id: &str,
possible_ancestor: &str,
) -> bool {
let mut pending = vec![node_id];
let mut seen = BTreeSet::new();
while let Some(current) = pending.pop() {
if !seen.insert(current) {
continue;
}
let Some(node) = nodes.get(current) else {
continue;
};
for dependency in &node.dependencies {
if dependency == possible_ancestor {
return true;
}
pending.push(dependency);
}
}
false
}
fn valid_node_id(value: &str) -> bool {
!value.is_empty()
&& value.len() <= 96
&& value
.bytes()
.all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'_' | b'-'))
}
#[derive(Debug, Error, PartialEq, Eq)]
pub enum AgentGraphError {
#[error("unsupported agent graph schema: {0}")]
UnsupportedSchema(String),
#[error("unsupported agent graph protocol: {0}")]
UnsupportedProtocol(String),
#[error("agent graph parent contract is invalid")]
InvalidParentContract,
#[error("agent graph does not match the parent contract or state")]
ParentBinding,
#[error("agent graph coordinator authority has the wrong role")]
CoordinatorRole,
#[error("agent graph budget is invalid")]
InvalidGraphBudget,
#[error("agent graph node limit is invalid or exceeded")]
NodeLimit,
#[error("agent graph nodes are not in canonical node-ID order")]
NonCanonicalNodeOrder,
#[error("agent graph contains a duplicate node ID")]
DuplicateNodeId,
#[error("invalid agent graph node ID: {0}")]
InvalidNodeId(String),
#[error("agent graph node has an invalid role: {0}")]
NodeRole(String),
#[error("agent graph node has invalid action or workspace authority: {0}")]
NodeAuthority(String),
#[error("agent graph reuses an authority: {0}")]
AuthorityReused(String),
#[error("node {node_id} references unknown dependency {dependency}")]
UnknownDependency {
node_id: String,
dependency: String,
},
#[error("agent graph contains a dependency cycle")]
Cycle,
#[error("review/adversary node has an invalid target: {0}")]
ReviewTarget(String),
#[error("worker lacks a complete reviewer/adversary proof barrier: {0}")]
IncompleteProofBarrier(String),
#[error("agent graph criterion is unknown: {0}")]
UnknownCriterion(String),
#[error("agent graph criterion is assigned more than once: {0}")]
CriterionAssignedTwice(String),
#[error("agent graph does not assign every parent criterion exactly once")]
IncompleteDecomposition,
#[error("node {node_id} writes outside its read scope: {path}")]
WriteOutsideReadScope {
node_id: String,
path: String,
},
#[error(
"parallel siblings {left_node_id} ({left_scope}) and {right_node_id} ({right_scope}) have overlapping write scopes"
)]
SiblingWriteConflict {
left_node_id: String,
right_node_id: String,
left_scope: String,
right_scope: String,
},
#[error("agent graph aggregate budget is exceeded")]
GraphBudgetExceeded,
#[error("scheduler decision is invalid")]
InvalidDecision,
#[error("scheduler state transition is invalid")]
InvalidTransition,
#[error("scheduler decision digest is invalid")]
DecisionDigest,
#[error("scheduler decision chain is invalid")]
DecisionChain,
#[error("scheduler terminal states are invalid")]
TerminalStates,
#[error("scheduler receipt is not bound to its graph")]
ReceiptBinding,
#[error("agent graph contract serialization failed")]
Serialization,
#[error(transparent)]
Delegation(#[from] AgentError),
}
#[cfg(test)]
mod tests {
use super::*;
use crate::{ClaimScope, Criterion};
fn parent_contract() -> TaskContract {
let mut criterion = Criterion::required("result", "produce result.txt");
criterion.evidence_requirement.allowed_producers = BTreeSet::from(["fs.patch".to_owned()]);
criterion.evidence_requirement.freshness =
crate::EvidenceFreshness::FinalWorkspaceGeneration;
let mut contract = TaskContract::new("produce result", vec![criterion]);
contract.confirmed = true;
contract.claim_scope = ClaimScope::Task;
contract
}
fn authority(role: AgentRole, name: &str) -> AgentAuthority {
AgentAuthority::new(role, name, name, "model", None).unwrap()
}
fn node(
node_id: &str,
role: AgentRole,
target: Option<&str>,
criteria: &[&str],
) -> AgentGraphNode {
AgentGraphNode {
node_id: node_id.to_owned(),
role,
dependencies: target
.map(|target| BTreeSet::from([target.to_owned()]))
.unwrap_or_default(),
review_target_node_id: target.map(str::to_owned),
delegated_criterion_ids: criteria.iter().map(|value| (*value).to_owned()).collect(),
added_constraints: Vec::new(),
authority: authority(role, node_id),
read_paths: BTreeSet::from(["result.txt".to_owned()]),
write_paths: if role == AgentRole::Worker {
BTreeSet::from(["result.txt".to_owned()])
} else {
BTreeSet::new()
},
budget: AgentBudget {
max_provider_turns: 2,
max_tool_calls: 2,
max_duration_ms: 1_000,
},
allowed_action_classes: if role == AgentRole::Worker {
BTreeSet::from([ActionClass::WorkspaceWrite])
} else {
BTreeSet::from([ActionClass::Read])
},
}
}
fn plan(parent: &TaskContract) -> AgentGraphPlan {
AgentGraphPlan::new(
Uuid::new_v4(),
parent,
"a".repeat(64),
authority(AgentRole::Coordinator, "coordinator"),
AgentGraphBudget {
max_nodes: 3,
max_concurrency: 2,
max_provider_turns: 6,
max_tool_calls: 6,
max_duration_ms: 3_000,
},
vec![
node("worker", AgentRole::Worker, None, &["result"]),
node("reviewer", AgentRole::Reviewer, Some("worker"), &[]),
node("adversary", AgentRole::Adversary, Some("worker"), &[]),
],
)
.unwrap()
}
fn template_node(
node_id: &str,
role: AgentRole,
target: Option<&str>,
criteria: &[&str],
) -> AgentGraphNodeTemplate {
AgentGraphNodeTemplate {
node_id: node_id.to_owned(),
role,
dependencies: target
.map(|target| BTreeSet::from([target.to_owned()]))
.unwrap_or_default(),
review_target_node_id: target.map(str::to_owned),
delegated_criterion_ids: criteria.iter().map(|value| (*value).to_owned()).collect(),
added_constraints: Vec::new(),
authority: AgentAuthorityTemplate {
profile: node_id.to_owned(),
provider: "openai".to_owned(),
model: "test-model".to_owned(),
credential_id: Some("usability".to_owned()),
},
read_paths: BTreeSet::from(["result.txt".to_owned()]),
write_paths: if role == AgentRole::Worker {
BTreeSet::from(["result.txt".to_owned()])
} else {
BTreeSet::new()
},
budget: AgentBudget {
max_provider_turns: 2,
max_tool_calls: 2,
max_duration_ms: 1_000,
},
allowed_action_classes: if role == AgentRole::Worker {
BTreeSet::from([ActionClass::WorkspaceWrite])
} else {
BTreeSet::from([ActionClass::Read])
},
}
}
fn template() -> AgentGraphTemplate {
AgentGraphTemplate {
schema_version: SCHEMA_VERSION.to_owned(),
protocol_version: AGENT_GRAPH_PROTOCOL_VERSION.to_owned(),
coordinator_authority: AgentAuthorityTemplate {
profile: "coordinator".to_owned(),
provider: "openai".to_owned(),
model: "test-model".to_owned(),
credential_id: Some("usability".to_owned()),
},
budget: AgentGraphBudget {
max_nodes: 3,
max_concurrency: 2,
max_provider_turns: 6,
max_tool_calls: 6,
max_duration_ms: 3_000,
},
nodes: vec![
template_node("worker", AgentRole::Worker, None, &["result"]),
template_node("reviewer", AgentRole::Reviewer, Some("worker"), &[]),
template_node("adversary", AgentRole::Adversary, Some("worker"), &[]),
],
}
}
fn cancelled_receipt(graph: &AgentGraphPlan, parent: &TaskContract) -> SchedulerReceipt {
let mut previous = None;
let mut decisions = Vec::new();
for (sequence, node_id) in ["adversary", "reviewer", "worker"].into_iter().enumerate() {
let decision = SchedulerDecision::new(
graph.graph_id,
sequence as u64,
node_id,
SchedulerDecisionKind::Cancel,
AgentNodeState::Pending,
AgentNodeState::Cancelled,
None,
None,
None,
None,
Some("parent_cancelled".to_owned()),
previous,
)
.unwrap();
previous = Some(decision.hash.clone());
decisions.push(decision);
}
SchedulerReceipt {
schema_version: SCHEMA_VERSION.to_owned(),
protocol_version: AGENT_GRAPH_PROTOCOL_VERSION.to_owned(),
graph_hash: graph.digest(parent).unwrap(),
decisions,
terminal_node_states: BTreeMap::from([
("adversary".to_owned(), AgentNodeState::Cancelled),
("reviewer".to_owned(), AgentNodeState::Cancelled),
("worker".to_owned(), AgentNodeState::Cancelled),
]),
outcome: AgentGraphOutcome::Cancelled,
final_workspace_generation: graph.parent_state_binding.clone(),
}
}
#[test]
fn graph_plan_requires_decomposition_review_barriers_and_budget() {
let parent = parent_contract();
let graph = plan(&parent);
graph.validate(&parent).unwrap();
assert_eq!(graph.digest(&parent).unwrap().len(), 64);
let mut missing_adversary = graph.clone();
missing_adversary
.nodes
.retain(|node| node.role != AgentRole::Adversary);
assert_eq!(
missing_adversary.validate(&parent),
Err(AgentGraphError::IncompleteProofBarrier("worker".to_owned()))
);
let mut excessive_budget = graph.clone();
excessive_budget.budget.max_tool_calls = 5;
assert_eq!(
excessive_budget.validate(&parent),
Err(AgentGraphError::GraphBudgetExceeded)
);
let mut cycle = graph.clone();
cycle
.nodes
.iter_mut()
.find(|node| node.node_id == "worker")
.unwrap()
.dependencies
.insert("reviewer".to_owned());
assert_eq!(cycle.validate(&parent), Err(AgentGraphError::Cycle));
}
#[test]
fn graph_plan_rejects_sibling_write_conflicts() {
let mut parent = parent_contract();
let mut second_criterion = Criterion::required("second", "produce nested result");
second_criterion.evidence_requirement.allowed_producers =
BTreeSet::from(["fs.create".to_owned()]);
second_criterion.evidence_requirement.freshness =
crate::EvidenceFreshness::FinalWorkspaceGeneration;
parent.criteria.push(second_criterion);
parent.validate().unwrap();
let mut second = node("worker-two", AgentRole::Worker, None, &["second"]);
second.write_paths = BTreeSet::from(["result.txt/child".to_owned()]);
let result = AgentGraphPlan::new(
Uuid::new_v4(),
&parent,
"a".repeat(64),
authority(AgentRole::Coordinator, "coordinator"),
AgentGraphBudget {
max_nodes: 6,
max_concurrency: 2,
max_provider_turns: 12,
max_tool_calls: 12,
max_duration_ms: 6_000,
},
vec![
node("worker", AgentRole::Worker, None, &["result"]),
node("reviewer", AgentRole::Reviewer, Some("worker"), &[]),
node("adversary", AgentRole::Adversary, Some("worker"), &[]),
second,
node("reviewer-two", AgentRole::Reviewer, Some("worker-two"), &[]),
node(
"adversary-two",
AgentRole::Adversary,
Some("worker-two"),
&[],
),
],
);
assert!(matches!(
result,
Err(AgentGraphError::SiblingWriteConflict { .. })
));
}
#[test]
fn scheduler_receipt_rejects_duplicate_reordered_and_tampered_decisions() {
let parent = parent_contract();
let graph = plan(&parent);
let receipt = cancelled_receipt(&graph, &parent);
receipt.validate(&graph, &parent).unwrap();
let mut duplicate = receipt.clone();
duplicate
.decisions
.insert(1, duplicate.decisions[0].clone());
assert_eq!(
duplicate.validate(&graph, &parent),
Err(AgentGraphError::DecisionChain)
);
let mut reordered = receipt.clone();
reordered.decisions.swap(0, 1);
assert_eq!(
reordered.validate(&graph, &parent),
Err(AgentGraphError::DecisionChain)
);
let mut tampered = receipt;
tampered.decisions[1].reason_code = Some("forged".to_owned());
assert_eq!(
tampered.validate(&graph, &parent),
Err(AgentGraphError::DecisionDigest)
);
}
#[test]
fn graph_template_mints_fresh_authorities_and_binds_exact_parent() {
let parent = parent_contract();
let parent_session_id = Uuid::new_v4();
let graph_template = template();
let first = graph_template
.bind(parent_session_id, &parent, "a".repeat(64))
.unwrap();
let second = graph_template
.bind(parent_session_id, &parent, "a".repeat(64))
.unwrap();
first.validate(&parent).unwrap();
second.validate(&parent).unwrap();
assert_ne!(first.graph_id, second.graph_id);
assert_ne!(
first.coordinator_authority.agent_id,
second.coordinator_authority.agent_id
);
assert!(
first
.nodes
.iter()
.zip(&second.nodes)
.all(|(left, right)| left.authority.agent_id != right.authority.agent_id)
);
assert_eq!(first.parent_session_id, parent_session_id);
assert_eq!(first.parent_contract_id, parent.id);
assert_eq!(first.parent_state_binding, "a".repeat(64));
let mut value = serde_json::to_value(graph_template).unwrap();
value["unexpected"] = serde_json::json!(true);
assert!(serde_json::from_value::<AgentGraphTemplate>(value).is_err());
}
#[test]
fn graph_json_rejects_unknown_fields() {
let parent = parent_contract();
let graph = plan(&parent);
let mut value = serde_json::to_value(graph).unwrap();
value["unexpected"] = serde_json::json!(true);
assert!(serde_json::from_value::<AgentGraphPlan>(value).is_err());
}
}