a3s_code_core/flow_graph/
mod.rs

1//! One-way projection from committed A3S Flow history into the state graph.
2//!
3//! Flow remains authoritative for scheduling and execution. This module only
4//! maintains an auditable domain projection; it never edits a Flow snapshot.
5
6use crate::state_graph::{
7    ExternalEvent, ExternalProjectionOutcome, GraphPatch, GraphRuntime, PatchOperation,
8    RuntimeError,
9};
10use a3s_flow::{FlowEvent, FlowEventEnvelope, FlowEventObserver};
11use async_trait::async_trait;
12use serde::{Deserialize, Serialize};
13use serde_json::{json, Map, Value};
14use std::sync::atomic::{AtomicU64, Ordering};
15use std::sync::Arc;
16use std::time::Instant;
17use tokio::sync::{Mutex, RwLock};
18
19mod decision;
20mod decision_ledger;
21pub use decision::{
22    FlowDecision, FlowDecisionDispatchError, FlowDecisionDispatcher, FlowDecisionHealthSnapshot,
23    FlowDecisionHealthStatus, FlowDecisionRequest, FlowDecisionSink, FlowDecisionStep,
24};
25pub use decision_ledger::{
26    FileFlowDecisionLedger, FlowDecisionClaimOutcome, FlowDecisionLedger, MemoryFlowDecisionLedger,
27};
28
29pub const FLOW_GRAPH_SOURCE: &str = "a3s-flow";
30
31#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
32#[serde(rename_all = "snake_case")]
33pub enum FlowGraphHealthStatus {
34    Healthy,
35    Degraded,
36}
37
38#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
39pub struct FlowGraphHealthSnapshot {
40    pub status: FlowGraphHealthStatus,
41    pub attempted: u64,
42    pub applied: u64,
43    pub duplicates: u64,
44    pub failures: u64,
45    pub sequence_gaps: u64,
46    pub event_conflicts: u64,
47    pub cancellations: u64,
48    pub in_flight: u64,
49    pub average_projection_micros: u64,
50    pub max_projection_micros: u64,
51    pub last_success_at_ms: Option<u64>,
52    pub last_failure_at_ms: Option<u64>,
53    pub last_error: Option<String>,
54}
55
56#[derive(Default)]
57struct FlowGraphMetrics {
58    attempted: AtomicU64,
59    applied: AtomicU64,
60    duplicates: AtomicU64,
61    failures: AtomicU64,
62    sequence_gaps: AtomicU64,
63    event_conflicts: AtomicU64,
64    cancellations: AtomicU64,
65    in_flight: AtomicU64,
66    total_projection_micros: AtomicU64,
67    max_projection_micros: AtomicU64,
68    last_success_at_ms: AtomicU64,
69    last_failure_at_ms: AtomicU64,
70}
71
72#[derive(Clone)]
73pub struct FlowGraphObserver {
74    runtime: Arc<Mutex<GraphRuntime>>,
75    last_error: Arc<RwLock<Option<String>>>,
76    metrics: Arc<FlowGraphMetrics>,
77}
78
79impl FlowGraphObserver {
80    pub fn new(runtime: Arc<Mutex<GraphRuntime>>) -> Self {
81        Self {
82            runtime,
83            last_error: Arc::new(RwLock::new(None)),
84            metrics: Arc::new(FlowGraphMetrics::default()),
85        }
86    }
87
88    pub fn runtime(&self) -> Arc<Mutex<GraphRuntime>> {
89        Arc::clone(&self.runtime)
90    }
91
92    pub async fn last_error(&self) -> Option<String> {
93        self.last_error.read().await.clone()
94    }
95
96    pub async fn health(&self) -> FlowGraphHealthSnapshot {
97        let attempted = self.metrics.attempted.load(Ordering::Relaxed);
98        let failures = self.metrics.failures.load(Ordering::Relaxed);
99        let total_micros = self.metrics.total_projection_micros.load(Ordering::Relaxed);
100        let last_error = self.last_error().await;
101        let last_success_at_ms = nonzero(self.metrics.last_success_at_ms.load(Ordering::Relaxed));
102        let last_failure_at_ms = nonzero(self.metrics.last_failure_at_ms.load(Ordering::Relaxed));
103        let failure_is_latest = last_failure_at_ms.unwrap_or(0) >= last_success_at_ms.unwrap_or(0)
104            && last_failure_at_ms.is_some();
105        FlowGraphHealthSnapshot {
106            status: if last_error.is_some() || failure_is_latest {
107                FlowGraphHealthStatus::Degraded
108            } else {
109                FlowGraphHealthStatus::Healthy
110            },
111            attempted,
112            applied: self.metrics.applied.load(Ordering::Relaxed),
113            duplicates: self.metrics.duplicates.load(Ordering::Relaxed),
114            failures,
115            sequence_gaps: self.metrics.sequence_gaps.load(Ordering::Relaxed),
116            event_conflicts: self.metrics.event_conflicts.load(Ordering::Relaxed),
117            cancellations: self.metrics.cancellations.load(Ordering::Relaxed),
118            in_flight: self.metrics.in_flight.load(Ordering::Relaxed),
119            average_projection_micros: total_micros.checked_div(attempted).unwrap_or(0),
120            max_projection_micros: self.metrics.max_projection_micros.load(Ordering::Relaxed),
121            last_success_at_ms,
122            last_failure_at_ms,
123            last_error,
124        }
125    }
126
127    pub async fn project(
128        &self,
129        envelope: FlowEventEnvelope,
130    ) -> Result<ExternalProjectionOutcome, RuntimeError> {
131        let mut in_flight = ProjectionInFlight::new(Arc::clone(&self.metrics));
132        let mut runtime = self.runtime.lock().await;
133        let external = external_event(&envelope);
134        let result = match runtime.check_external(&external) {
135            Ok(Some(outcome)) => Ok(outcome),
136            Ok(None) => match projection_patch(&runtime, &envelope) {
137                Ok(patch) => runtime.project_external(external, patch),
138                Err(error) => Err(error),
139            },
140            Err(error) => Err(error),
141        };
142        drop(runtime);
143        let elapsed = in_flight.finish();
144        self.record_result(&envelope.event, elapsed, &result).await;
145        result
146    }
147
148    /// Replay committed Flow history in sequence order, applying only events
149    /// not already represented by the restored graph cursor.
150    pub async fn catch_up(
151        &self,
152        mut history: Vec<FlowEventEnvelope>,
153    ) -> Result<usize, RuntimeError> {
154        history.sort_by(|left, right| {
155            left.run_id
156                .cmp(&right.run_id)
157                .then(left.sequence.cmp(&right.sequence))
158        });
159        let mut applied = 0;
160        for envelope in history {
161            if self.project(envelope).await? == ExternalProjectionOutcome::Applied {
162                applied += 1;
163            }
164        }
165        Ok(applied)
166    }
167
168    async fn record_result(
169        &self,
170        event: &FlowEvent,
171        elapsed: u64,
172        result: &Result<ExternalProjectionOutcome, RuntimeError>,
173    ) {
174        match result {
175            Ok(ExternalProjectionOutcome::Applied) => {
176                self.metrics.applied.fetch_add(1, Ordering::Relaxed);
177                self.metrics
178                    .last_success_at_ms
179                    .store(now_ms(), Ordering::Relaxed);
180                *self.last_error.write().await = None;
181                tracing::debug!(
182                    event_key = event.event_key(),
183                    outcome = "applied",
184                    duration_micros = elapsed,
185                    "flow graph projection"
186                );
187            }
188            Ok(ExternalProjectionOutcome::Duplicate) => {
189                self.metrics.duplicates.fetch_add(1, Ordering::Relaxed);
190                self.metrics
191                    .last_success_at_ms
192                    .store(now_ms(), Ordering::Relaxed);
193                *self.last_error.write().await = None;
194                tracing::debug!(
195                    event_key = event.event_key(),
196                    outcome = "duplicate",
197                    duration_micros = elapsed,
198                    "flow graph projection"
199                );
200            }
201            Err(error) => {
202                self.metrics.failures.fetch_add(1, Ordering::Relaxed);
203                if matches!(error, RuntimeError::ExternalSequenceDiverged { .. }) {
204                    self.metrics.sequence_gaps.fetch_add(1, Ordering::Relaxed);
205                }
206                if matches!(error, RuntimeError::ExternalEventConflict { .. }) {
207                    self.metrics.event_conflicts.fetch_add(1, Ordering::Relaxed);
208                }
209                self.metrics
210                    .last_failure_at_ms
211                    .store(now_ms(), Ordering::Relaxed);
212                *self.last_error.write().await = Some(error.to_string());
213                tracing::warn!(event_key = event.event_key(), error = %error, duration_micros = elapsed, "flow graph projection failed");
214            }
215        }
216    }
217}
218
219struct ProjectionInFlight {
220    metrics: Arc<FlowGraphMetrics>,
221    started: Instant,
222    completed: bool,
223}
224
225impl ProjectionInFlight {
226    fn new(metrics: Arc<FlowGraphMetrics>) -> Self {
227        metrics.attempted.fetch_add(1, Ordering::Relaxed);
228        metrics.in_flight.fetch_add(1, Ordering::Relaxed);
229        Self {
230            metrics,
231            started: Instant::now(),
232            completed: false,
233        }
234    }
235
236    fn finish(&mut self) -> u64 {
237        let elapsed = self.started.elapsed().as_micros().min(u128::from(u64::MAX)) as u64;
238        self.metrics.in_flight.fetch_sub(1, Ordering::Relaxed);
239        self.metrics
240            .total_projection_micros
241            .fetch_add(elapsed, Ordering::Relaxed);
242        self.metrics
243            .max_projection_micros
244            .fetch_max(elapsed, Ordering::Relaxed);
245        self.completed = true;
246        elapsed
247    }
248}
249
250impl Drop for ProjectionInFlight {
251    fn drop(&mut self) {
252        if self.completed {
253            return;
254        }
255        let elapsed = self.started.elapsed().as_micros().min(u128::from(u64::MAX)) as u64;
256        self.metrics.in_flight.fetch_sub(1, Ordering::Relaxed);
257        self.metrics.failures.fetch_add(1, Ordering::Relaxed);
258        self.metrics.cancellations.fetch_add(1, Ordering::Relaxed);
259        self.metrics
260            .total_projection_micros
261            .fetch_add(elapsed, Ordering::Relaxed);
262        self.metrics
263            .max_projection_micros
264            .fetch_max(elapsed, Ordering::Relaxed);
265        self.metrics
266            .last_failure_at_ms
267            .store(now_ms(), Ordering::Relaxed);
268    }
269}
270
271#[async_trait]
272impl FlowEventObserver for FlowGraphObserver {
273    async fn observe(&self, envelope: FlowEventEnvelope) {
274        let _ = self.project(envelope).await;
275    }
276}
277
278fn nonzero(value: u64) -> Option<u64> {
279    (value != 0).then_some(value)
280}
281
282fn now_ms() -> u64 {
283    use std::time::{SystemTime, UNIX_EPOCH};
284    SystemTime::now()
285        .duration_since(UNIX_EPOCH)
286        .unwrap_or_default()
287        .as_millis()
288        .min(u128::from(u64::MAX)) as u64
289}
290
291fn external_event(envelope: &FlowEventEnvelope) -> ExternalEvent {
292    ExternalEvent {
293        source: FLOW_GRAPH_SOURCE.to_string(),
294        stream_id: envelope.run_id.clone(),
295        sequence: envelope.sequence,
296        event_id: envelope.event_id.to_string(),
297        name: envelope.event.event_key().to_string(),
298        payload: flow_event_payload(&envelope.event),
299    }
300}
301
302fn flow_event_payload(event: &FlowEvent) -> Value {
303    match event {
304        // Hook tokens are execution capabilities and must not enter the graph
305        // audit log. Flow retains the authoritative secret-bearing event.
306        FlowEvent::HookCreated {
307            hook_id, metadata, ..
308        } => json!({"type": "hook_created", "hook_id": hook_id, "metadata": metadata}),
309        _ => serde_json::to_value(event).unwrap_or(Value::Null),
310    }
311}
312
313fn projection_patch(
314    runtime: &GraphRuntime,
315    envelope: &FlowEventEnvelope,
316) -> Result<GraphPatch, RuntimeError> {
317    let graph = runtime.graph();
318    let run_id = run_object_id(&envelope.run_id);
319    let mut operations = Vec::new();
320    match &envelope.event {
321        FlowEvent::RunCreated { spec, input } => operations.push(PatchOperation::AddObject {
322            id: run_id,
323            object_type: "workflow_run".to_string(),
324            data: json!({
325                "run_id": envelope.run_id,
326                "status": "created",
327                "spec": spec,
328                "input": input,
329                "last_sequence": envelope.sequence,
330            }),
331        }),
332        FlowEvent::RunStarted => update_object(
333            graph,
334            &run_id,
335            envelope.sequence,
336            [("status", json!("running"))],
337            &mut operations,
338        )?,
339        FlowEvent::RunCompleted { output } => update_object(
340            graph,
341            &run_id,
342            envelope.sequence,
343            [("status", json!("completed")), ("output", output.clone())],
344            &mut operations,
345        )?,
346        FlowEvent::RunFailed { error } => update_object(
347            graph,
348            &run_id,
349            envelope.sequence,
350            [("status", json!("failed")), ("error", json!(error))],
351            &mut operations,
352        )?,
353        FlowEvent::RunCancelled { reason } => update_object(
354            graph,
355            &run_id,
356            envelope.sequence,
357            [("status", json!("cancelled")), ("reason", json!(reason))],
358            &mut operations,
359        )?,
360        FlowEvent::StepCreated {
361            step_id,
362            step_name,
363            input,
364            retry,
365        } => {
366            let object_id = step_object_id(&envelope.run_id, step_id);
367            operations.push(PatchOperation::AddObject {
368                id: object_id.clone(),
369                object_type: "workflow_step".to_string(),
370                data: json!({"run_id": envelope.run_id, "step_id": step_id, "name": step_name,
371                    "input": input, "retry": retry, "status": "created", "last_sequence": envelope.sequence}),
372            });
373            operations.push(PatchOperation::AddRelation {
374                id: contains_relation_id(&envelope.run_id, "step", step_id),
375                relation_type: "contains".to_string(),
376                source: run_id.clone(),
377                target: object_id,
378                data: json!({"kind": "step"}),
379            });
380            touch_run(graph, &run_id, envelope.sequence, &mut operations)?;
381        }
382        FlowEvent::StepStarted { step_id, attempt } => update_subject(
383            graph,
384            &run_id,
385            &step_object_id(&envelope.run_id, step_id),
386            envelope.sequence,
387            [("status", json!("running")), ("attempt", json!(attempt))],
388            &mut operations,
389        )?,
390        FlowEvent::StepCompleted { step_id, output } => update_subject(
391            graph,
392            &run_id,
393            &step_object_id(&envelope.run_id, step_id),
394            envelope.sequence,
395            [("status", json!("completed")), ("output", output.clone())],
396            &mut operations,
397        )?,
398        FlowEvent::StepRetrying {
399            step_id,
400            attempt,
401            error,
402            retry_after,
403        } => update_subject(
404            graph,
405            &run_id,
406            &step_object_id(&envelope.run_id, step_id),
407            envelope.sequence,
408            [
409                ("status", json!("retrying")),
410                ("attempt", json!(attempt)),
411                ("error", json!(error)),
412                ("retry_after", json!(retry_after)),
413            ],
414            &mut operations,
415        )?,
416        FlowEvent::StepFailed {
417            step_id,
418            attempt,
419            error,
420        } => update_subject(
421            graph,
422            &run_id,
423            &step_object_id(&envelope.run_id, step_id),
424            envelope.sequence,
425            [
426                ("status", json!("failed")),
427                ("attempt", json!(attempt)),
428                ("error", json!(error)),
429            ],
430            &mut operations,
431        )?,
432        FlowEvent::WaitCreated { wait_id, resume_at } => add_subject(
433            graph,
434            SubjectProjection {
435                run_object_id: &run_id,
436                raw_run_id: &envelope.run_id,
437                kind: "wait",
438                id: wait_id,
439                object_type: "workflow_wait",
440                sequence: envelope.sequence,
441                extra: json!({"resume_at": resume_at, "status": "waiting"}),
442            },
443            &mut operations,
444        )?,
445        FlowEvent::WaitCompleted { wait_id } => update_subject(
446            graph,
447            &run_id,
448            &subject_object_id(&envelope.run_id, "wait", wait_id),
449            envelope.sequence,
450            [("status", json!("completed"))],
451            &mut operations,
452        )?,
453        FlowEvent::HookCreated {
454            hook_id,
455            token: _,
456            metadata,
457        } => add_subject(
458            graph,
459            SubjectProjection {
460                run_object_id: &run_id,
461                raw_run_id: &envelope.run_id,
462                kind: "hook",
463                id: hook_id,
464                object_type: "workflow_hook",
465                sequence: envelope.sequence,
466                extra: json!({"metadata": metadata, "status": "waiting"}),
467            },
468            &mut operations,
469        )?,
470        FlowEvent::HookReceived { hook_id, payload } => update_subject(
471            graph,
472            &run_id,
473            &subject_object_id(&envelope.run_id, "hook", hook_id),
474            envelope.sequence,
475            [("status", json!("received")), ("payload", payload.clone())],
476            &mut operations,
477        )?,
478        FlowEvent::HookDisposed { hook_id } => update_subject(
479            graph,
480            &run_id,
481            &subject_object_id(&envelope.run_id, "hook", hook_id),
482            envelope.sequence,
483            [("status", json!("disposed"))],
484            &mut operations,
485        )?,
486    }
487    Ok(GraphPatch::new(graph.version(), operations))
488}
489
490fn update_subject<const N: usize>(
491    graph: &crate::StateGraph,
492    run_id: &str,
493    subject_id: &str,
494    sequence: u64,
495    fields: [(&str, Value); N],
496    operations: &mut Vec<PatchOperation>,
497) -> Result<(), RuntimeError> {
498    update_object(graph, subject_id, sequence, fields, operations)?;
499    touch_run(graph, run_id, sequence, operations)
500}
501
502struct SubjectProjection<'a> {
503    run_object_id: &'a str,
504    raw_run_id: &'a str,
505    kind: &'a str,
506    id: &'a str,
507    object_type: &'a str,
508    sequence: u64,
509    extra: Value,
510}
511
512fn add_subject(
513    graph: &crate::StateGraph,
514    subject: SubjectProjection<'_>,
515    operations: &mut Vec<PatchOperation>,
516) -> Result<(), RuntimeError> {
517    let object_id = subject_object_id(subject.raw_run_id, subject.kind, subject.id);
518    let mut data = subject.extra.as_object().cloned().unwrap_or_default();
519    data.insert("run_id".to_string(), json!(subject.raw_run_id));
520    data.insert(format!("{}_id", subject.kind), json!(subject.id));
521    data.insert("last_sequence".to_string(), json!(subject.sequence));
522    operations.push(PatchOperation::AddObject {
523        id: object_id.clone(),
524        object_type: subject.object_type.to_string(),
525        data: Value::Object(data),
526    });
527    operations.push(PatchOperation::AddRelation {
528        id: contains_relation_id(subject.raw_run_id, subject.kind, subject.id),
529        relation_type: "contains".to_string(),
530        source: subject.run_object_id.to_string(),
531        target: object_id,
532        data: json!({"kind": subject.kind}),
533    });
534    touch_run(graph, subject.run_object_id, subject.sequence, operations)
535}
536
537fn touch_run(
538    graph: &crate::StateGraph,
539    run_id: &str,
540    sequence: u64,
541    operations: &mut Vec<PatchOperation>,
542) -> Result<(), RuntimeError> {
543    update_object(graph, run_id, sequence, [], operations)
544}
545
546fn update_object<const N: usize>(
547    graph: &crate::StateGraph,
548    id: &str,
549    sequence: u64,
550    fields: [(&str, Value); N],
551    operations: &mut Vec<PatchOperation>,
552) -> Result<(), RuntimeError> {
553    let object = graph.object(id).ok_or_else(|| {
554        RuntimeError::InvalidExternalProjection(format!("projected object `{id}` does not exist"))
555    })?;
556    let mut data: Map<String, Value> = object.data.as_object().cloned().unwrap_or_default();
557    for (key, value) in fields {
558        data.insert(key.to_string(), value);
559    }
560    data.insert("last_sequence".to_string(), json!(sequence));
561    operations.push(PatchOperation::UpdateObject {
562        id: id.to_string(),
563        expected_version: object.version,
564        data: Value::Object(data),
565    });
566    Ok(())
567}
568
569pub fn run_object_id(run_id: &str) -> String {
570    format!("flow:run:{run_id}")
571}
572pub fn step_object_id(run_id: &str, step_id: &str) -> String {
573    subject_object_id(run_id, "step", step_id)
574}
575fn subject_object_id(run_id: &str, kind: &str, id: &str) -> String {
576    format!("flow:{kind}:{run_id}:{id}")
577}
578fn contains_relation_id(run_id: &str, kind: &str, id: &str) -> String {
579    format!("flow:contains:{run_id}:{kind}:{id}")
580}
581
582#[cfg(test)]
583mod tests;
a3s_code_core/flow_graph/mod.rs

a3s_code_core/flow_graph/
mod.rs
