1use crate::{
2 ActorRef, ApprovalRequest, ApprovalStatus, Behavior, BehaviorContext, Event, EventId,
3 EventPattern, EventStore, GraphSnapshot, Node, NodeId, Pack, Policy, PolicyAction,
4 PolicyDecision, PolicyId, StateError, StateOp, YoAgentState,
5};
6use serde_json::json;
7use std::collections::BTreeMap;
8use std::sync::Arc;
9
10pub struct YoAgentRuntime<S: EventStore> {
11 state: YoAgentState<S>,
12 packs: BTreeMap<String, Pack>,
13 behaviors: Vec<Arc<dyn Behavior>>,
14 policies: Vec<Policy>,
15}
16
17impl<S: EventStore> YoAgentRuntime<S> {
18 pub fn new(state: YoAgentState<S>) -> Self {
19 Self {
20 state,
21 packs: BTreeMap::new(),
22 behaviors: Vec::new(),
23 policies: Vec::new(),
24 }
25 }
26
27 pub fn state(&self) -> YoAgentState<S> {
28 self.state.clone()
29 }
30
31 pub fn register_pack(&mut self, pack: Pack) {
32 self.packs.insert(pack.id.0.clone(), pack);
33 }
34
35 pub fn register_behavior<B>(&mut self, behavior: B)
36 where
37 B: Behavior + 'static,
38 {
39 self.behaviors.push(Arc::new(behavior));
40 }
41
42 pub fn register_policy(&mut self, policy: Policy) {
43 self.policies.push(policy);
44 }
45
46 pub async fn emit_event(&self, event: Event) -> Result<EventId, StateError> {
47 let event_id = self.state.record_event(event.clone()).await?;
48 self.run_behaviors(&event).await?;
49 Ok(event_id)
50 }
51
52 pub async fn create_typed_node(
53 &self,
54 actor: ActorRef,
55 id: NodeId,
56 kind: impl Into<String>,
57 props: serde_json::Value,
58 ) -> Result<EventId, StateError> {
59 self.ensure_allowed(PolicyAction::CreateNode, Some(id.clone()), actor.clone())
60 .await?;
61 let node = Node::new(id.clone(), kind.into(), props.clone());
62 for pack in self.packs.values() {
63 pack.validate_node(&node)?;
64 }
65 self.state
66 .apply_ops(
67 actor,
68 vec![StateOp::CreateNode {
69 id,
70 kind: node.kind,
71 props,
72 }],
73 )
74 .await
75 }
76
77 pub async fn create_typed_relation(
78 &self,
79 actor: ActorRef,
80 from: NodeId,
81 rel: impl Into<String>,
82 to: NodeId,
83 props: serde_json::Value,
84 ) -> Result<EventId, StateError> {
85 self.ensure_allowed(
86 PolicyAction::CreateRelation,
87 Some(from.clone()),
88 actor.clone(),
89 )
90 .await?;
91 let rel = rel.into();
92 let graph = self.state.graph().await;
93 let from_node = graph
94 .get_node(&from)
95 .ok_or_else(|| StateError::NodeNotFound(from.clone()))?;
96 let to_node = graph
97 .get_node(&to)
98 .ok_or_else(|| StateError::NodeNotFound(to.clone()))?;
99 let relation = crate::Relation::new(from.clone(), rel.clone(), to.clone(), props.clone());
100 for pack in self.packs.values() {
101 pack.validate_relation(&relation, from_node, to_node)?;
102 }
103 self.state
104 .apply_ops(
105 actor,
106 vec![StateOp::CreateRelation {
107 from,
108 rel,
109 to,
110 props,
111 }],
112 )
113 .await
114 }
115
116 pub async fn request_approval(
117 &self,
118 policy_id: PolicyId,
119 action: PolicyAction,
120 target: Option<NodeId>,
121 requested_by: ActorRef,
122 reason: Option<String>,
123 ) -> Result<EventId, StateError> {
124 let request = ApprovalRequest {
125 id: NodeId::generate(),
126 policy_id,
127 action,
128 target,
129 requested_by: requested_by.clone(),
130 status: ApprovalStatus::Pending,
131 reason,
132 metadata: json!({}),
133 };
134 self.state
135 .apply_ops(
136 requested_by,
137 vec![StateOp::CreateNode {
138 id: request.id.clone(),
139 kind: "approval_request".to_string(),
140 props: serde_json::to_value(request)?,
141 }],
142 )
143 .await
144 }
145
146 pub async fn approve_request(
147 &self,
148 actor: ActorRef,
149 request_id: NodeId,
150 reason: Option<String>,
151 ) -> Result<EventId, StateError> {
152 self.state
153 .apply_ops(
154 actor,
155 vec![StateOp::UpdateNode {
156 id: request_id,
157 props: json!({
158 "status": ApprovalStatus::Approved,
159 "approval_reason": reason,
160 }),
161 }],
162 )
163 .await
164 }
165
166 pub async fn reject_request(
167 &self,
168 actor: ActorRef,
169 request_id: NodeId,
170 reason: Option<String>,
171 ) -> Result<EventId, StateError> {
172 self.state
173 .apply_ops(
174 actor,
175 vec![StateOp::UpdateNode {
176 id: request_id,
177 props: json!({
178 "status": ApprovalStatus::Rejected,
179 "rejection_reason": reason,
180 }),
181 }],
182 )
183 .await
184 }
185
186 pub async fn graph(&self) -> GraphSnapshot {
187 self.state.graph().await
188 }
189
190 async fn ensure_allowed(
191 &self,
192 action: PolicyAction,
193 target: Option<NodeId>,
194 actor: ActorRef,
195 ) -> Result<(), StateError> {
196 for policy in &self.policies {
197 if policy.action != action {
198 continue;
199 }
200 match policy.decision {
201 PolicyDecision::Allow => {}
202 PolicyDecision::Deny => {
203 return Err(StateError::PolicyDenied(
204 policy
205 .reason
206 .clone()
207 .unwrap_or_else(|| policy.title.clone()),
208 ));
209 }
210 PolicyDecision::RequireApproval => {
211 self.request_approval(
212 policy.id.clone(),
213 action,
214 target,
215 actor,
216 policy.reason.clone(),
217 )
218 .await?;
219 return Err(StateError::PolicyDenied(format!(
220 "approval required by policy {}",
221 policy.title
222 )));
223 }
224 }
225 }
226 Ok(())
227 }
228
229 async fn run_behaviors(&self, event: &Event) -> Result<(), StateError> {
230 for behavior in &self.behaviors {
231 let pattern: EventPattern = behavior.pattern();
232 if !pattern.matches(event) {
233 continue;
234 }
235 let ops = behavior
236 .handle(
237 BehaviorContext {
238 graph: self.state.graph().await,
239 replaying: false,
240 },
241 event,
242 )
243 .await?;
244 if !ops.is_empty() {
245 self.state.apply_ops(event.actor.clone(), ops).await?;
246 }
247 }
248 Ok(())
249 }
250}