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::RunCancellationRequested { request } => {
354            cancel_open_subjects(graph, &run_id, envelope.sequence, &mut operations)?;
355            update_object(
356                graph,
357                &run_id,
358                envelope.sequence,
359                [
360                    ("status", json!("cancelling")),
361                    (
362                        "cancellation",
363                        json!({
364                            "request": request,
365                            "requested_at": envelope.timestamp,
366                            "sequence": envelope.sequence,
367                        }),
368                    ),
369                ],
370                &mut operations,
371            )?;
372        }
373        FlowEvent::RunCancelled { reason } => update_object(
374            graph,
375            &run_id,
376            envelope.sequence,
377            [("status", json!("cancelled")), ("reason", json!(reason))],
378            &mut operations,
379        )?,
380        FlowEvent::RunTimedOut { deadline, reason } => {
381            let error = reason
382                .clone()
383                .unwrap_or_else(|| format!("workflow timed out at {deadline}"));
384            update_object(
385                graph,
386                &run_id,
387                envelope.sequence,
388                [
389                    ("status", json!("failed")),
390                    ("error", json!(error)),
391                    ("deadline", json!(deadline)),
392                    ("reason", json!(reason)),
393                    (
394                        "terminal_outcome",
395                        json!({
396                            "type": "timed_out",
397                            "deadline": deadline,
398                            "reason": reason,
399                        }),
400                    ),
401                ],
402                &mut operations,
403            )?;
404        }
405        FlowEvent::RunRetryExhausted {
406            step_id,
407            attempt,
408            error,
409        } => update_object(
410            graph,
411            &run_id,
412            envelope.sequence,
413            [
414                ("status", json!("failed")),
415                ("error", json!(error)),
416                ("step_id", json!(step_id)),
417                ("attempt", json!(attempt)),
418                (
419                    "terminal_outcome",
420                    json!({
421                        "type": "retry_exhausted",
422                        "step_id": step_id,
423                        "attempt": attempt,
424                        "error": error,
425                    }),
426                ),
427            ],
428            &mut operations,
429        )?,
430        FlowEvent::RunHostShutdown { reason } => {
431            let error = reason
432                .clone()
433                .unwrap_or_else(|| "workflow terminated by host shutdown".to_string());
434            update_object(
435                graph,
436                &run_id,
437                envelope.sequence,
438                [
439                    ("status", json!("failed")),
440                    ("error", json!(error)),
441                    ("reason", json!(reason)),
442                    (
443                        "terminal_outcome",
444                        json!({"type": "host_shutdown", "reason": reason}),
445                    ),
446                ],
447                &mut operations,
448            )?;
449        }
450        FlowEvent::RunProgressRecorded { progress } => add_subject(
451            graph,
452            SubjectProjection {
453                run_object_id: &run_id,
454                raw_run_id: &envelope.run_id,
455                kind: "progress",
456                id_field: "progress_id",
457                id: &progress.progress_id,
458                object_type: "workflow_progress",
459                sequence: envelope.sequence,
460                extra: serde_json::to_value(progress).map_err(|error| {
461                    RuntimeError::InvalidExternalProjection(format!(
462                        "failed to serialize workflow progress: {error}"
463                    ))
464                })?,
465            },
466            &mut operations,
467        )?,
468        FlowEvent::ChildOperationLinked { child } => add_subject(
469            graph,
470            SubjectProjection {
471                run_object_id: &run_id,
472                raw_run_id: &envelope.run_id,
473                kind: "child_operation",
474                id_field: "reference_id",
475                id: &child.reference_id,
476                object_type: "workflow_child_operation",
477                sequence: envelope.sequence,
478                extra: serde_json::to_value(child).map_err(|error| {
479                    RuntimeError::InvalidExternalProjection(format!(
480                        "failed to serialize child operation reference: {error}"
481                    ))
482                })?,
483            },
484            &mut operations,
485        )?,
486        FlowEvent::StepCreated {
487            step_id,
488            step_name,
489            input,
490            retry,
491        } => {
492            let object_id = step_object_id(&envelope.run_id, step_id);
493            operations.push(PatchOperation::AddObject {
494                id: object_id.clone(),
495                object_type: "workflow_step".to_string(),
496                data: json!({"run_id": envelope.run_id, "step_id": step_id, "name": step_name,
497                    "input": input, "retry": retry, "status": "created", "last_sequence": envelope.sequence}),
498            });
499            operations.push(PatchOperation::AddRelation {
500                id: contains_relation_id(&envelope.run_id, "step", step_id),
501                relation_type: "contains".to_string(),
502                source: run_id.clone(),
503                target: object_id,
504                data: json!({"kind": "step"}),
505            });
506            touch_run(graph, &run_id, envelope.sequence, &mut operations)?;
507        }
508        FlowEvent::StepStarted { step_id, attempt } => update_subject(
509            graph,
510            &run_id,
511            &step_object_id(&envelope.run_id, step_id),
512            envelope.sequence,
513            [("status", json!("running")), ("attempt", json!(attempt))],
514            &mut operations,
515        )?,
516        FlowEvent::StepCompleted { step_id, output } => update_subject(
517            graph,
518            &run_id,
519            &step_object_id(&envelope.run_id, step_id),
520            envelope.sequence,
521            [("status", json!("completed")), ("output", output.clone())],
522            &mut operations,
523        )?,
524        FlowEvent::StepRetrying {
525            step_id,
526            attempt,
527            error,
528            retry_after,
529        } => update_subject(
530            graph,
531            &run_id,
532            &step_object_id(&envelope.run_id, step_id),
533            envelope.sequence,
534            [
535                ("status", json!("retrying")),
536                ("attempt", json!(attempt)),
537                ("error", json!(error)),
538                ("retry_after", json!(retry_after)),
539            ],
540            &mut operations,
541        )?,
542        FlowEvent::StepFailed {
543            step_id,
544            attempt,
545            error,
546        } => update_subject(
547            graph,
548            &run_id,
549            &step_object_id(&envelope.run_id, step_id),
550            envelope.sequence,
551            [
552                ("status", json!("failed")),
553                ("attempt", json!(attempt)),
554                ("error", json!(error)),
555            ],
556            &mut operations,
557        )?,
558        FlowEvent::WaitCreated { wait_id, resume_at } => add_subject(
559            graph,
560            SubjectProjection {
561                run_object_id: &run_id,
562                raw_run_id: &envelope.run_id,
563                kind: "wait",
564                id_field: "wait_id",
565                id: wait_id,
566                object_type: "workflow_wait",
567                sequence: envelope.sequence,
568                extra: json!({"resume_at": resume_at, "status": "waiting"}),
569            },
570            &mut operations,
571        )?,
572        FlowEvent::WaitCompleted { wait_id } => update_subject(
573            graph,
574            &run_id,
575            &subject_object_id(&envelope.run_id, "wait", wait_id),
576            envelope.sequence,
577            [("status", json!("completed"))],
578            &mut operations,
579        )?,
580        FlowEvent::HookCreated {
581            hook_id,
582            token: _,
583            metadata,
584        } => add_subject(
585            graph,
586            SubjectProjection {
587                run_object_id: &run_id,
588                raw_run_id: &envelope.run_id,
589                kind: "hook",
590                id_field: "hook_id",
591                id: hook_id,
592                object_type: "workflow_hook",
593                sequence: envelope.sequence,
594                extra: json!({"metadata": metadata, "status": "waiting"}),
595            },
596            &mut operations,
597        )?,
598        FlowEvent::HookReceived { hook_id, payload } => update_subject(
599            graph,
600            &run_id,
601            &subject_object_id(&envelope.run_id, "hook", hook_id),
602            envelope.sequence,
603            [("status", json!("received")), ("payload", payload.clone())],
604            &mut operations,
605        )?,
606        FlowEvent::HookDisposed { hook_id } => update_subject(
607            graph,
608            &run_id,
609            &subject_object_id(&envelope.run_id, "hook", hook_id),
610            envelope.sequence,
611            [("status", json!("disposed"))],
612            &mut operations,
613        )?,
614    }
615    Ok(GraphPatch::new(graph.version(), operations))
616}
617
618fn update_subject<const N: usize>(
619    graph: &crate::StateGraph,
620    run_id: &str,
621    subject_id: &str,
622    sequence: u64,
623    fields: [(&str, Value); N],
624    operations: &mut Vec<PatchOperation>,
625) -> Result<(), RuntimeError> {
626    update_object(graph, subject_id, sequence, fields, operations)?;
627    touch_run(graph, run_id, sequence, operations)
628}
629
630struct SubjectProjection<'a> {
631    run_object_id: &'a str,
632    raw_run_id: &'a str,
633    kind: &'a str,
634    id_field: &'a str,
635    id: &'a str,
636    object_type: &'a str,
637    sequence: u64,
638    extra: Value,
639}
640
641fn add_subject(
642    graph: &crate::StateGraph,
643    subject: SubjectProjection<'_>,
644    operations: &mut Vec<PatchOperation>,
645) -> Result<(), RuntimeError> {
646    let object_id = subject_object_id(subject.raw_run_id, subject.kind, subject.id);
647    let mut data = subject.extra.as_object().cloned().unwrap_or_default();
648    data.insert("run_id".to_string(), json!(subject.raw_run_id));
649    data.insert(subject.id_field.to_string(), json!(subject.id));
650    data.insert("last_sequence".to_string(), json!(subject.sequence));
651    operations.push(PatchOperation::AddObject {
652        id: object_id.clone(),
653        object_type: subject.object_type.to_string(),
654        data: Value::Object(data),
655    });
656    operations.push(PatchOperation::AddRelation {
657        id: contains_relation_id(subject.raw_run_id, subject.kind, subject.id),
658        relation_type: "contains".to_string(),
659        source: subject.run_object_id.to_string(),
660        target: object_id,
661        data: json!({"kind": subject.kind}),
662    });
663    touch_run(graph, subject.run_object_id, subject.sequence, operations)
664}
665
666fn cancel_open_subjects(
667    graph: &crate::StateGraph,
668    run_id: &str,
669    sequence: u64,
670    operations: &mut Vec<PatchOperation>,
671) -> Result<(), RuntimeError> {
672    for relation in graph.relations_from(run_id) {
673        if relation.relation_type != "contains" {
674            continue;
675        }
676        let subject = graph.object(&relation.target).ok_or_else(|| {
677            RuntimeError::InvalidExternalProjection(format!(
678                "projected relation `{}` references missing object `{}`",
679                relation.id, relation.target
680            ))
681        })?;
682        let status = subject.data.get("status").and_then(Value::as_str);
683        match (subject.object_type.as_str(), status) {
684            ("workflow_step", Some("created" | "running" | "retrying")) => update_object(
685                graph,
686                &subject.id,
687                sequence,
688                [("status", json!("cancelled")), ("retry_after", Value::Null)],
689                operations,
690            )?,
691            ("workflow_wait", Some("waiting")) | ("workflow_hook", Some("waiting" | "active")) => {
692                update_object(
693                    graph,
694                    &subject.id,
695                    sequence,
696                    [("status", json!("cancelled"))],
697                    operations,
698                )?
699            }
700            _ => {}
701        }
702    }
703    Ok(())
704}
705
706fn touch_run(
707    graph: &crate::StateGraph,
708    run_id: &str,
709    sequence: u64,
710    operations: &mut Vec<PatchOperation>,
711) -> Result<(), RuntimeError> {
712    update_object(graph, run_id, sequence, [], operations)
713}
714
715fn update_object<const N: usize>(
716    graph: &crate::StateGraph,
717    id: &str,
718    sequence: u64,
719    fields: [(&str, Value); N],
720    operations: &mut Vec<PatchOperation>,
721) -> Result<(), RuntimeError> {
722    let object = graph.object(id).ok_or_else(|| {
723        RuntimeError::InvalidExternalProjection(format!("projected object `{id}` does not exist"))
724    })?;
725    let mut data: Map<String, Value> = object.data.as_object().cloned().unwrap_or_default();
726    for (key, value) in fields {
727        data.insert(key.to_string(), value);
728    }
729    data.insert("last_sequence".to_string(), json!(sequence));
730    operations.push(PatchOperation::UpdateObject {
731        id: id.to_string(),
732        expected_version: object.version,
733        data: Value::Object(data),
734    });
735    Ok(())
736}
737
738pub fn run_object_id(run_id: &str) -> String {
739    format!("flow:run:{run_id}")
740}
741pub fn step_object_id(run_id: &str, step_id: &str) -> String {
742    subject_object_id(run_id, "step", step_id)
743}
744fn subject_object_id(run_id: &str, kind: &str, id: &str) -> String {
745    format!("flow:{kind}:{run_id}:{id}")
746}
747fn contains_relation_id(run_id: &str, kind: &str, id: &str) -> String {
748    format!("flow:contains:{run_id}:{kind}:{id}")
749}
750
751#[cfg(test)]
752mod tests;
a3s_code_core/flow_graph/mod.rs

a3s_code_core/flow_graph/
mod.rs
