1use super::behavior::{Behavior, BehaviorContext};
2use super::graph::{record_hash, replay_strict};
3use super::{
4 GraphDiff, GraphEvent, GraphEventRecord, GraphPatch, PatchOperation, ReplayError, StateGraph,
5 GRAPH_EVENT_SCHEMA_VERSION,
6};
7use std::collections::VecDeque;
8use std::sync::Arc;
9use thiserror::Error;
10
11#[derive(Debug, Clone, Copy)]
12pub struct RuntimeLimits {
13 pub max_events: usize,
14 pub max_behavior_depth: usize,
15}
16
17impl Default for RuntimeLimits {
18 fn default() -> Self {
19 Self {
20 max_events: 10_000,
21 max_behavior_depth: 64,
22 }
23 }
24}
25
26pub struct GraphRuntime {
27 branch_id: String,
28 correlation_id: Option<String>,
29 graph: StateGraph,
30 events: Vec<GraphEventRecord>,
31 behaviors: Vec<Arc<dyn Behavior>>,
32 pending: VecDeque<(usize, GraphEventRecord)>,
33 limits: RuntimeLimits,
34}
35
36impl GraphRuntime {
37 pub fn new() -> Self {
38 Self::with_limits(RuntimeLimits::default())
39 }
40
41 pub fn with_limits(limits: RuntimeLimits) -> Self {
42 Self {
43 branch_id: new_id("branch"),
44 correlation_id: None,
45 graph: StateGraph::default(),
46 events: Vec::new(),
47 behaviors: Vec::new(),
48 pending: VecDeque::new(),
49 limits,
50 }
51 }
52
53 pub fn with_correlation_id(mut self, correlation_id: impl Into<String>) -> Self {
54 self.correlation_id = Some(correlation_id.into());
55 self
56 }
57
58 pub fn restore(events: Vec<GraphEventRecord>) -> Result<Self, ReplayError> {
59 let graph = replay_strict(&events)?;
60 let branch_id = events
61 .last()
62 .map(|record| record.branch_id.clone())
63 .unwrap_or_else(|| new_id("branch"));
64 let correlation_id = events
65 .iter()
66 .rev()
67 .find_map(|record| record.correlation_id.clone());
68 Ok(Self {
69 branch_id,
70 correlation_id,
71 graph,
72 events,
73 behaviors: Vec::new(),
74 pending: VecDeque::new(),
75 limits: RuntimeLimits::default(),
76 })
77 }
78
79 pub fn branch_id(&self) -> &str {
80 &self.branch_id
81 }
82
83 pub fn graph(&self) -> &StateGraph {
84 &self.graph
85 }
86
87 pub fn events(&self) -> &[GraphEventRecord] {
88 &self.events
89 }
90
91 pub fn register(&mut self, behavior: Arc<dyn Behavior>) -> Result<(), RuntimeError> {
92 if self
93 .behaviors
94 .iter()
95 .any(|registered| registered.name() == behavior.name())
96 {
97 return Err(RuntimeError::DuplicateBehavior(behavior.name().to_string()));
98 }
99 self.behaviors.push(behavior);
100 Ok(())
101 }
102
103 pub fn emit(&mut self, event: GraphEvent) -> Result<GraphEventRecord, RuntimeError> {
104 let record = self.append(event, None)?;
105 self.pending.push_back((0, record.clone()));
106 self.drain_behaviors()?;
107 Ok(record)
108 }
109
110 pub fn run_goal(&mut self, goal: impl Into<String>) -> Result<GraphEventRecord, RuntimeError> {
111 self.emit(GraphEvent::GoalCreated { goal: goal.into() })
112 }
113
114 pub fn propose_patch(
115 &mut self,
116 patch: GraphPatch,
117 causation_id: Option<String>,
118 ) -> Result<bool, RuntimeError> {
119 let records = self.apply_patch(patch, causation_id, 0)?;
120 self.drain_behaviors()?;
121 Ok(records)
122 }
123
124 pub fn fork_at(&self, sequence_exclusive: u64) -> Result<Self, RuntimeError> {
125 let end = usize::try_from(sequence_exclusive)
126 .map_err(|_| RuntimeError::InvalidFork(sequence_exclusive))?;
127 if end > self.events.len() {
128 return Err(RuntimeError::InvalidFork(sequence_exclusive));
129 }
130 let events = self.events[..end].to_vec();
131 let graph = replay_strict(&events)?;
132 let parent_branch_id = self.branch_id.clone();
133 let mut fork = Self {
134 branch_id: new_id("branch"),
135 correlation_id: self.correlation_id.clone(),
136 graph,
137 events,
138 behaviors: self.behaviors.clone(),
139 pending: VecDeque::new(),
140 limits: self.limits,
141 };
142 let cause = fork.events.last().map(|record| record.id.clone());
143 fork.append(
144 GraphEvent::BranchForked {
145 parent_branch_id,
146 fork_sequence: sequence_exclusive,
147 },
148 cause,
149 )?;
150 Ok(fork)
151 }
152
153 pub fn diff(&self, other: &Self) -> GraphDiff {
154 self.graph.diff(&other.graph)
155 }
156
157 pub fn strict_replay(records: &[GraphEventRecord]) -> Result<StateGraph, ReplayError> {
158 replay_strict(records)
159 }
160
161 fn drain_behaviors(&mut self) -> Result<(), RuntimeError> {
162 while let Some((depth, triggering_event)) = self.pending.pop_front() {
163 if depth >= self.limits.max_behavior_depth {
164 return Err(RuntimeError::BehaviorDepthExceeded(
165 self.limits.max_behavior_depth,
166 ));
167 }
168 let matching = self
169 .behaviors
170 .iter()
171 .filter(|behavior| behavior.filter().matches(&triggering_event, &self.graph))
172 .cloned()
173 .collect::<Vec<_>>();
174 for behavior in matching {
175 let cause = Some(triggering_event.id.clone());
176 self.append(
177 GraphEvent::BehaviorStarted {
178 name: behavior.name().to_string(),
179 },
180 cause.clone(),
181 )?;
182 let result = behavior.evaluate(BehaviorContext {
183 graph: &self.graph,
184 event: &triggering_event,
185 });
186 match result {
187 Ok(patches) => {
188 for patch in patches {
189 self.apply_patch(patch, cause.clone(), depth + 1)?;
190 }
191 self.append(
192 GraphEvent::BehaviorCompleted {
193 name: behavior.name().to_string(),
194 },
195 cause,
196 )?;
197 }
198 Err(error) => {
199 self.append(
200 GraphEvent::BehaviorFailed {
201 name: behavior.name().to_string(),
202 error: error.to_string(),
203 },
204 cause,
205 )?;
206 }
207 }
208 }
209 }
210 Ok(())
211 }
212
213 fn apply_patch(
214 &mut self,
215 patch: GraphPatch,
216 causation_id: Option<String>,
217 depth: usize,
218 ) -> Result<bool, RuntimeError> {
219 let patch_id = new_id("patch");
220 let validation = self.validate_patch(&patch);
221 let required_events = validation
222 .as_ref()
223 .map_or(2, |mutation_events| mutation_events.len().saturating_add(2));
224 if self.events.len().saturating_add(required_events) > self.limits.max_events {
225 return Err(RuntimeError::EventLimitExceeded(self.limits.max_events));
226 }
227 let proposed = self.append(
228 GraphEvent::PatchProposed {
229 patch_id: patch_id.clone(),
230 patch: patch.clone(),
231 },
232 causation_id,
233 )?;
234 self.pending.push_back((depth, proposed.clone()));
235
236 let mutation_events = match validation {
237 Ok(events) => events,
238 Err(reason) => {
239 let rejected = self.append(
240 GraphEvent::PatchRejected { patch_id, reason },
241 Some(proposed.id),
242 )?;
243 self.pending.push_back((depth, rejected));
244 return Ok(false);
245 }
246 };
247
248 let mut last_cause = proposed.id;
249 for event in mutation_events {
250 let record = self.append(event, Some(last_cause))?;
251 last_cause = record.id.clone();
252 self.pending.push_back((depth, record));
253 }
254 let applied = self.append(GraphEvent::PatchApplied { patch_id }, Some(last_cause))?;
255 self.pending.push_back((depth, applied));
256 Ok(true)
257 }
258
259 fn validate_patch(&self, patch: &GraphPatch) -> Result<Vec<GraphEvent>, String> {
260 if patch.expected_graph_version != self.graph.version() {
261 return Err(format!(
262 "graph version conflict: expected {}, current {}",
263 patch.expected_graph_version,
264 self.graph.version()
265 ));
266 }
267 let mut candidate = self.graph.clone();
268 let mut events = Vec::with_capacity(patch.operations.len());
269 for operation in &patch.operations {
270 let event = operation_event(operation, &candidate)?;
271 candidate.apply(&event).map_err(|error| error.to_string())?;
272 events.push(event);
273 }
274 Ok(events)
275 }
276
277 fn append(
278 &mut self,
279 event: GraphEvent,
280 causation_id: Option<String>,
281 ) -> Result<GraphEventRecord, RuntimeError> {
282 if self.events.len() >= self.limits.max_events {
283 return Err(RuntimeError::EventLimitExceeded(self.limits.max_events));
284 }
285 let state_version_before = self.graph.version();
286 self.graph.apply(&event)?;
287 let mut record = GraphEventRecord {
288 schema_version: GRAPH_EVENT_SCHEMA_VERSION,
289 id: new_id("event"),
290 sequence: self.events.len() as u64,
291 timestamp_ms: now_ms(),
292 branch_id: self.branch_id.clone(),
293 causation_id,
294 correlation_id: self.correlation_id.clone(),
295 state_version_before,
296 state_version_after: self.graph.version(),
297 state_hash_after: self.graph.state_hash()?,
298 previous_record_hash: self.events.last().map(|record| record.record_hash.clone()),
299 record_hash: String::new(),
300 event,
301 };
302 record.record_hash = record_hash(&record)?;
303 self.events.push(record.clone());
304 Ok(record)
305 }
306}
307
308impl Default for GraphRuntime {
309 fn default() -> Self {
310 Self::new()
311 }
312}
313
314fn operation_event(operation: &PatchOperation, graph: &StateGraph) -> Result<GraphEvent, String> {
315 Ok(match operation {
316 PatchOperation::AddObject {
317 id,
318 object_type,
319 data,
320 } => GraphEvent::ObjectCreated {
321 id: id.clone(),
322 object_type: object_type.clone(),
323 data: data.clone(),
324 },
325 PatchOperation::UpdateObject {
326 id,
327 expected_version,
328 data,
329 } => {
330 let current = graph
331 .object(id)
332 .ok_or_else(|| format!("object `{id}` does not exist"))?;
333 if current.version != *expected_version {
334 return Err(format!(
335 "object `{id}` version conflict: expected {expected_version}, current {}",
336 current.version
337 ));
338 }
339 GraphEvent::ObjectUpdated {
340 id: id.clone(),
341 version: current.version + 1,
342 data: data.clone(),
343 }
344 }
345 PatchOperation::RemoveObject {
346 id,
347 expected_version,
348 } => {
349 let current = graph
350 .object(id)
351 .ok_or_else(|| format!("object `{id}` does not exist"))?;
352 if current.version != *expected_version {
353 return Err(format!("object `{id}` version conflict"));
354 }
355 GraphEvent::ObjectRemoved {
356 id: id.clone(),
357 version: current.version + 1,
358 }
359 }
360 PatchOperation::AddRelation {
361 id,
362 relation_type,
363 source,
364 target,
365 data,
366 } => GraphEvent::RelationCreated {
367 id: id.clone(),
368 relation_type: relation_type.clone(),
369 source: source.clone(),
370 target: target.clone(),
371 data: data.clone(),
372 },
373 PatchOperation::UpdateRelation {
374 id,
375 expected_version,
376 data,
377 } => {
378 let current = graph
379 .relation(id)
380 .ok_or_else(|| format!("relation `{id}` does not exist"))?;
381 if current.version != *expected_version {
382 return Err(format!("relation `{id}` version conflict"));
383 }
384 GraphEvent::RelationUpdated {
385 id: id.clone(),
386 version: current.version + 1,
387 data: data.clone(),
388 }
389 }
390 PatchOperation::RemoveRelation {
391 id,
392 expected_version,
393 } => {
394 let current = graph
395 .relation(id)
396 .ok_or_else(|| format!("relation `{id}` does not exist"))?;
397 if current.version != *expected_version {
398 return Err(format!("relation `{id}` version conflict"));
399 }
400 GraphEvent::RelationRemoved {
401 id: id.clone(),
402 version: current.version + 1,
403 }
404 }
405 })
406}
407
408fn new_id(prefix: &str) -> String {
409 format!("{prefix}-{}", uuid::Uuid::new_v4())
410}
411
412fn now_ms() -> u64 {
413 use std::time::{SystemTime, UNIX_EPOCH};
414 SystemTime::now()
415 .duration_since(UNIX_EPOCH)
416 .unwrap_or_default()
417 .as_millis()
418 .min(u128::from(u64::MAX)) as u64
419}
420
421#[derive(Debug, Error)]
422pub enum RuntimeError {
423 #[error(transparent)]
424 Replay(#[from] ReplayError),
425 #[error("behavior `{0}` is already registered")]
426 DuplicateBehavior(String),
427 #[error("event limit of {0} exceeded")]
428 EventLimitExceeded(usize),
429 #[error("behavior recursion depth of {0} exceeded")]
430 BehaviorDepthExceeded(usize),
431 #[error("cannot fork at exclusive sequence {0}")]
432 InvalidFork(u64),
433}