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, FlowDecisionClaimState, FlowDecisionLedger,
27    MemoryFlowDecisionLedger,
28};
29
30pub const FLOW_GRAPH_SOURCE: &str = "a3s-flow";
31
32#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
33#[serde(rename_all = "snake_case")]
34pub enum FlowGraphHealthStatus {
35    Healthy,
36    Degraded,
37}
38
39#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
40pub struct FlowGraphHealthSnapshot {
41    pub status: FlowGraphHealthStatus,
42    pub attempted: u64,
43    pub applied: u64,
44    pub duplicates: u64,
45    pub failures: u64,
46    pub sequence_gaps: u64,
47    pub event_conflicts: u64,
48    pub cancellations: u64,
49    pub in_flight: u64,
50    pub average_projection_micros: u64,
51    pub max_projection_micros: u64,
52    pub last_success_at_ms: Option<u64>,
53    pub last_failure_at_ms: Option<u64>,
54    pub last_error: Option<String>,
55}
56
57#[derive(Default)]
58struct FlowGraphMetrics {
59    attempted: AtomicU64,
60    applied: AtomicU64,
61    duplicates: AtomicU64,
62    failures: AtomicU64,
63    sequence_gaps: AtomicU64,
64    event_conflicts: AtomicU64,
65    cancellations: AtomicU64,
66    in_flight: AtomicU64,
67    total_projection_micros: AtomicU64,
68    max_projection_micros: AtomicU64,
69    last_success_at_ms: AtomicU64,
70    last_failure_at_ms: AtomicU64,
71}
72
73#[derive(Clone)]
74pub struct FlowGraphObserver {
75    runtime: Arc<Mutex<GraphRuntime>>,
76    last_error: Arc<RwLock<Option<String>>>,
77    metrics: Arc<FlowGraphMetrics>,
78}
79
80impl FlowGraphObserver {
81    pub fn new(runtime: Arc<Mutex<GraphRuntime>>) -> Self {
82        Self {
83            runtime,
84            last_error: Arc::new(RwLock::new(None)),
85            metrics: Arc::new(FlowGraphMetrics::default()),
86        }
87    }
88
89    pub fn runtime(&self) -> Arc<Mutex<GraphRuntime>> {
90        Arc::clone(&self.runtime)
91    }
92
93    pub async fn last_error(&self) -> Option<String> {
94        self.last_error.read().await.clone()
95    }
96
97    pub async fn health(&self) -> FlowGraphHealthSnapshot {
98        let attempted = self.metrics.attempted.load(Ordering::Relaxed);
99        let failures = self.metrics.failures.load(Ordering::Relaxed);
100        let total_micros = self.metrics.total_projection_micros.load(Ordering::Relaxed);
101        let last_error = self.last_error().await;
102        let last_success_at_ms = nonzero(self.metrics.last_success_at_ms.load(Ordering::Relaxed));
103        let last_failure_at_ms = nonzero(self.metrics.last_failure_at_ms.load(Ordering::Relaxed));
104        let failure_is_latest = last_failure_at_ms.unwrap_or(0) >= last_success_at_ms.unwrap_or(0)
105            && last_failure_at_ms.is_some();
106        FlowGraphHealthSnapshot {
107            status: if last_error.is_some() || failure_is_latest {
108                FlowGraphHealthStatus::Degraded
109            } else {
110                FlowGraphHealthStatus::Healthy
111            },
112            attempted,
113            applied: self.metrics.applied.load(Ordering::Relaxed),
114            duplicates: self.metrics.duplicates.load(Ordering::Relaxed),
115            failures,
116            sequence_gaps: self.metrics.sequence_gaps.load(Ordering::Relaxed),
117            event_conflicts: self.metrics.event_conflicts.load(Ordering::Relaxed),
118            cancellations: self.metrics.cancellations.load(Ordering::Relaxed),
119            in_flight: self.metrics.in_flight.load(Ordering::Relaxed),
120            average_projection_micros: total_micros.checked_div(attempted).unwrap_or(0),
121            max_projection_micros: self.metrics.max_projection_micros.load(Ordering::Relaxed),
122            last_success_at_ms,
123            last_failure_at_ms,
124            last_error,
125        }
126    }
127
128    pub async fn project(
129        &self,
130        envelope: FlowEventEnvelope,
131    ) -> Result<ExternalProjectionOutcome, RuntimeError> {
132        let mut in_flight = ProjectionInFlight::new(Arc::clone(&self.metrics));
133        let mut runtime = self.runtime.lock().await;
134        let external = external_event(&envelope);
135        let result = match runtime.check_external(&external) {
136            Ok(Some(outcome)) => Ok(outcome),
137            Ok(None) => match projection_patch(&runtime, &envelope) {
138                Ok(patch) => runtime.project_external(external, patch),
139                Err(error) => Err(error),
140            },
141            Err(error) => Err(error),
142        };
143        drop(runtime);
144        let elapsed = in_flight.finish();
145        self.record_result(&envelope.event, elapsed, &result).await;
146        result
147    }
148
149    /// Replay committed Flow history in sequence order, applying only events
150    /// not already represented by the restored graph cursor.
151    pub async fn catch_up(
152        &self,
153        mut history: Vec<FlowEventEnvelope>,
154    ) -> Result<usize, RuntimeError> {
155        history.sort_by(|left, right| {
156            left.run_id
157                .cmp(&right.run_id)
158                .then(left.sequence.cmp(&right.sequence))
159        });
160        let mut applied = 0;
161        for envelope in history {
162            if self.project(envelope).await? == ExternalProjectionOutcome::Applied {
163                applied += 1;
164            }
165        }
166        Ok(applied)
167    }
168
169    async fn record_result(
170        &self,
171        event: &FlowEvent,
172        elapsed: u64,
173        result: &Result<ExternalProjectionOutcome, RuntimeError>,
174    ) {
175        match result {
176            Ok(ExternalProjectionOutcome::Applied) => {
177                self.metrics.applied.fetch_add(1, Ordering::Relaxed);
178                self.metrics
179                    .last_success_at_ms
180                    .store(now_ms(), Ordering::Relaxed);
181                *self.last_error.write().await = None;
182                tracing::debug!(
183                    event_key = event.event_key(),
184                    outcome = "applied",
185                    duration_micros = elapsed,
186                    "flow graph projection"
187                );
188            }
189            Ok(ExternalProjectionOutcome::Duplicate) => {
190                self.metrics.duplicates.fetch_add(1, Ordering::Relaxed);
191                self.metrics
192                    .last_success_at_ms
193                    .store(now_ms(), Ordering::Relaxed);
194                *self.last_error.write().await = None;
195                tracing::debug!(
196                    event_key = event.event_key(),
197                    outcome = "duplicate",
198                    duration_micros = elapsed,
199                    "flow graph projection"
200                );
201            }
202            Err(error) => {
203                self.metrics.failures.fetch_add(1, Ordering::Relaxed);
204                if matches!(error, RuntimeError::ExternalSequenceDiverged { .. }) {
205                    self.metrics.sequence_gaps.fetch_add(1, Ordering::Relaxed);
206                }
207                if matches!(error, RuntimeError::ExternalEventConflict { .. }) {
208                    self.metrics.event_conflicts.fetch_add(1, Ordering::Relaxed);
209                }
210                self.metrics
211                    .last_failure_at_ms
212                    .store(now_ms(), Ordering::Relaxed);
213                *self.last_error.write().await = Some(error.to_string());
214                tracing::warn!(event_key = event.event_key(), error = %error, duration_micros = elapsed, "flow graph projection failed");
215            }
216        }
217    }
218}
219
220struct ProjectionInFlight {
221    metrics: Arc<FlowGraphMetrics>,
222    started: Instant,
223    completed: bool,
224}
225
226impl ProjectionInFlight {
227    fn new(metrics: Arc<FlowGraphMetrics>) -> Self {
228        metrics.attempted.fetch_add(1, Ordering::Relaxed);
229        metrics.in_flight.fetch_add(1, Ordering::Relaxed);
230        Self {
231            metrics,
232            started: Instant::now(),
233            completed: false,
234        }
235    }
236
237    fn finish(&mut self) -> u64 {
238        let elapsed = self.started.elapsed().as_micros().min(u128::from(u64::MAX)) as u64;
239        self.metrics.in_flight.fetch_sub(1, Ordering::Relaxed);
240        self.metrics
241            .total_projection_micros
242            .fetch_add(elapsed, Ordering::Relaxed);
243        self.metrics
244            .max_projection_micros
245            .fetch_max(elapsed, Ordering::Relaxed);
246        self.completed = true;
247        elapsed
248    }
249}
250
251impl Drop for ProjectionInFlight {
252    fn drop(&mut self) {
253        if self.completed {
254            return;
255        }
256        let elapsed = self.started.elapsed().as_micros().min(u128::from(u64::MAX)) as u64;
257        self.metrics.in_flight.fetch_sub(1, Ordering::Relaxed);
258        self.metrics.failures.fetch_add(1, Ordering::Relaxed);
259        self.metrics.cancellations.fetch_add(1, Ordering::Relaxed);
260        self.metrics
261            .total_projection_micros
262            .fetch_add(elapsed, Ordering::Relaxed);
263        self.metrics
264            .max_projection_micros
265            .fetch_max(elapsed, Ordering::Relaxed);
266        self.metrics
267            .last_failure_at_ms
268            .store(now_ms(), Ordering::Relaxed);
269    }
270}
271
272#[async_trait]
273impl FlowEventObserver for FlowGraphObserver {
274    async fn observe(&self, envelope: FlowEventEnvelope) {
275        let _ = self.project(envelope).await;
276    }
277}
278
279fn nonzero(value: u64) -> Option<u64> {
280    (value != 0).then_some(value)
281}
282
283fn now_ms() -> u64 {
284    use std::time::{SystemTime, UNIX_EPOCH};
285    SystemTime::now()
286        .duration_since(UNIX_EPOCH)
287        .unwrap_or_default()
288        .as_millis()
289        .min(u128::from(u64::MAX)) as u64
290}
291
292fn external_event(envelope: &FlowEventEnvelope) -> ExternalEvent {
293    ExternalEvent {
294        source: FLOW_GRAPH_SOURCE.to_string(),
295        stream_id: envelope.run_id.clone(),
296        sequence: envelope.sequence,
297        event_id: envelope.event_id.to_string(),
298        name: envelope.event.event_key().to_string(),
299        payload: flow_event_payload(&envelope.event),
300    }
301}
302
303fn flow_event_payload(event: &FlowEvent) -> Value {
304    match event {
305        // Hook tokens are execution capabilities and must not enter the graph
306        // audit log. Flow retains the authoritative secret-bearing event.
307        FlowEvent::HookCreated {
308            hook_id, metadata, ..
309        } => json!({"type": "hook_created", "hook_id": hook_id, "metadata": metadata}),
310        _ => serde_json::to_value(event).unwrap_or(Value::Null),
311    }
312}
313
314fn projection_patch(
315    runtime: &GraphRuntime,
316    envelope: &FlowEventEnvelope,
317) -> Result<GraphPatch, RuntimeError> {
318    let graph = runtime.graph();
319    let run_id = run_object_id(&envelope.run_id);
320    let mut operations = Vec::new();
321    match &envelope.event {
322        FlowEvent::RunCreated { spec, input } => operations.push(PatchOperation::AddObject {
323            id: run_id,
324            object_type: "workflow_run".to_string(),
325            data: json!({
326                "run_id": envelope.run_id,
327                "status": "created",
328                "spec": spec,
329                "input": input,
330                "last_sequence": envelope.sequence,
331            }),
332        }),
333        FlowEvent::RunStarted => update_object(
334            graph,
335            &run_id,
336            envelope.sequence,
337            [("status", json!("running"))],
338            &mut operations,
339        )?,
340        FlowEvent::RunCompleted { output } => update_object(
341            graph,
342            &run_id,
343            envelope.sequence,
344            [("status", json!("completed")), ("output", output.clone())],
345            &mut operations,
346        )?,
347        FlowEvent::RunFailed { error } => update_object(
348            graph,
349            &run_id,
350            envelope.sequence,
351            [("status", json!("failed")), ("error", json!(error))],
352            &mut operations,
353        )?,
354        FlowEvent::RunCancellationRequested { request } => {
355            cancel_open_subjects(graph, &run_id, envelope.sequence, &mut operations)?;
356            update_object(
357                graph,
358                &run_id,
359                envelope.sequence,
360                [
361                    ("status", json!("cancelling")),
362                    (
363                        "cancellation",
364                        json!({
365                            "request": request,
366                            "requested_at": envelope.timestamp,
367                            "sequence": envelope.sequence,
368                        }),
369                    ),
370                ],
371                &mut operations,
372            )?;
373        }
374        FlowEvent::RunCancelled { reason } => update_object(
375            graph,
376            &run_id,
377            envelope.sequence,
378            [("status", json!("cancelled")), ("reason", json!(reason))],
379            &mut operations,
380        )?,
381        FlowEvent::RunTimedOut { deadline, reason } => {
382            let error = reason
383                .clone()
384                .unwrap_or_else(|| format!("workflow timed out at {deadline}"));
385            update_object(
386                graph,
387                &run_id,
388                envelope.sequence,
389                [
390                    ("status", json!("failed")),
391                    ("error", json!(error)),
392                    ("deadline", json!(deadline)),
393                    ("reason", json!(reason)),
394                    (
395                        "terminal_outcome",
396                        json!({
397                            "type": "timed_out",
398                            "deadline": deadline,
399                            "reason": reason,
400                        }),
401                    ),
402                ],
403                &mut operations,
404            )?;
405        }
406        FlowEvent::RunRetryExhausted {
407            step_id,
408            attempt,
409            error,
410        } => update_object(
411            graph,
412            &run_id,
413            envelope.sequence,
414            [
415                ("status", json!("failed")),
416                ("error", json!(error)),
417                ("step_id", json!(step_id)),
418                ("attempt", json!(attempt)),
419                (
420                    "terminal_outcome",
421                    json!({
422                        "type": "retry_exhausted",
423                        "step_id": step_id,
424                        "attempt": attempt,
425                        "error": error,
426                    }),
427                ),
428            ],
429            &mut operations,
430        )?,
431        FlowEvent::RunHostShutdown { reason } => {
432            let error = reason
433                .clone()
434                .unwrap_or_else(|| "workflow terminated by host shutdown".to_string());
435            update_object(
436                graph,
437                &run_id,
438                envelope.sequence,
439                [
440                    ("status", json!("failed")),
441                    ("error", json!(error)),
442                    ("reason", json!(reason)),
443                    (
444                        "terminal_outcome",
445                        json!({"type": "host_shutdown", "reason": reason}),
446                    ),
447                ],
448                &mut operations,
449            )?;
450        }
451        FlowEvent::RunProgressRecorded { progress } => add_subject(
452            graph,
453            SubjectProjection {
454                run_object_id: &run_id,
455                raw_run_id: &envelope.run_id,
456                kind: "progress",
457                id_field: "progress_id",
458                id: &progress.progress_id,
459                object_type: "workflow_progress",
460                sequence: envelope.sequence,
461                extra: serde_json::to_value(progress).map_err(|error| {
462                    RuntimeError::InvalidExternalProjection(format!(
463                        "failed to serialize workflow progress: {error}"
464                    ))
465                })?,
466            },
467            &mut operations,
468        )?,
469        FlowEvent::ChildOperationLinked { child } => add_subject(
470            graph,
471            SubjectProjection {
472                run_object_id: &run_id,
473                raw_run_id: &envelope.run_id,
474                kind: "child_operation",
475                id_field: "reference_id",
476                id: &child.reference_id,
477                object_type: "workflow_child_operation",
478                sequence: envelope.sequence,
479                extra: serde_json::to_value(child).map_err(|error| {
480                    RuntimeError::InvalidExternalProjection(format!(
481                        "failed to serialize child operation reference: {error}"
482                    ))
483                })?,
484            },
485            &mut operations,
486        )?,
487        FlowEvent::StepCreated {
488            step_id,
489            step_name,
490            input,
491            retry,
492        } => {
493            let object_id = step_object_id(&envelope.run_id, step_id);
494            operations.push(PatchOperation::AddObject {
495                id: object_id.clone(),
496                object_type: "workflow_step".to_string(),
497                data: json!({"run_id": envelope.run_id, "step_id": step_id, "name": step_name,
498                    "input": input, "retry": retry, "status": "created", "last_sequence": envelope.sequence}),
499            });
500            operations.push(PatchOperation::AddRelation {
501                id: contains_relation_id(&envelope.run_id, "step", step_id),
502                relation_type: "contains".to_string(),
503                source: run_id.clone(),
504                target: object_id,
505                data: json!({"kind": "step"}),
506            });
507            touch_run(graph, &run_id, envelope.sequence, &mut operations)?;
508        }
509        FlowEvent::StepStarted { step_id, attempt } => update_subject(
510            graph,
511            &run_id,
512            &step_object_id(&envelope.run_id, step_id),
513            envelope.sequence,
514            [("status", json!("running")), ("attempt", json!(attempt))],
515            &mut operations,
516        )?,
517        FlowEvent::StepCompleted { step_id, output } => update_subject(
518            graph,
519            &run_id,
520            &step_object_id(&envelope.run_id, step_id),
521            envelope.sequence,
522            [("status", json!("completed")), ("output", output.clone())],
523            &mut operations,
524        )?,
525        FlowEvent::StepRetrying {
526            step_id,
527            attempt,
528            error,
529            retry_after,
530        } => update_subject(
531            graph,
532            &run_id,
533            &step_object_id(&envelope.run_id, step_id),
534            envelope.sequence,
535            [
536                ("status", json!("retrying")),
537                ("attempt", json!(attempt)),
538                ("error", json!(error)),
539                ("retry_after", json!(retry_after)),
540            ],
541            &mut operations,
542        )?,
543        FlowEvent::StepFailed {
544            step_id,
545            attempt,
546            error,
547        } => update_subject(
548            graph,
549            &run_id,
550            &step_object_id(&envelope.run_id, step_id),
551            envelope.sequence,
552            [
553                ("status", json!("failed")),
554                ("attempt", json!(attempt)),
555                ("error", json!(error)),
556            ],
557            &mut operations,
558        )?,
559        FlowEvent::WaitCreated { wait_id, resume_at } => add_subject(
560            graph,
561            SubjectProjection {
562                run_object_id: &run_id,
563                raw_run_id: &envelope.run_id,
564                kind: "wait",
565                id_field: "wait_id",
566                id: wait_id,
567                object_type: "workflow_wait",
568                sequence: envelope.sequence,
569                extra: json!({"resume_at": resume_at, "status": "waiting"}),
570            },
571            &mut operations,
572        )?,
573        FlowEvent::WaitCompleted { wait_id } => update_subject(
574            graph,
575            &run_id,
576            &subject_object_id(&envelope.run_id, "wait", wait_id),
577            envelope.sequence,
578            [("status", json!("completed"))],
579            &mut operations,
580        )?,
581        FlowEvent::HookCreated {
582            hook_id,
583            token: _,
584            metadata,
585        } => add_subject(
586            graph,
587            SubjectProjection {
588                run_object_id: &run_id,
589                raw_run_id: &envelope.run_id,
590                kind: "hook",
591                id_field: "hook_id",
592                id: hook_id,
593                object_type: "workflow_hook",
594                sequence: envelope.sequence,
595                extra: json!({"metadata": metadata, "status": "waiting"}),
596            },
597            &mut operations,
598        )?,
599        FlowEvent::HookReceived { hook_id, payload } => update_subject(
600            graph,
601            &run_id,
602            &subject_object_id(&envelope.run_id, "hook", hook_id),
603            envelope.sequence,
604            [("status", json!("received")), ("payload", payload.clone())],
605            &mut operations,
606        )?,
607        FlowEvent::HookDisposed { hook_id } => update_subject(
608            graph,
609            &run_id,
610            &subject_object_id(&envelope.run_id, "hook", hook_id),
611            envelope.sequence,
612            [("status", json!("disposed"))],
613            &mut operations,
614        )?,
615        // Flow events added compatibly during the 1.x line remain durably
616        // observed by the external cursor even before Code projects their
617        // domain-specific fields.
618        _ => {}
619    }
620    Ok(GraphPatch::new(graph.version(), operations))
621}
622
623fn update_subject<const N: usize>(
624    graph: &crate::StateGraph,
625    run_id: &str,
626    subject_id: &str,
627    sequence: u64,
628    fields: [(&str, Value); N],
629    operations: &mut Vec<PatchOperation>,
630) -> Result<(), RuntimeError> {
631    update_object(graph, subject_id, sequence, fields, operations)?;
632    touch_run(graph, run_id, sequence, operations)
633}
634
635struct SubjectProjection<'a> {
636    run_object_id: &'a str,
637    raw_run_id: &'a str,
638    kind: &'a str,
639    id_field: &'a str,
640    id: &'a str,
641    object_type: &'a str,
642    sequence: u64,
643    extra: Value,
644}
645
646fn add_subject(
647    graph: &crate::StateGraph,
648    subject: SubjectProjection<'_>,
649    operations: &mut Vec<PatchOperation>,
650) -> Result<(), RuntimeError> {
651    let object_id = subject_object_id(subject.raw_run_id, subject.kind, subject.id);
652    let mut data = subject.extra.as_object().cloned().unwrap_or_default();
653    data.insert("run_id".to_string(), json!(subject.raw_run_id));
654    data.insert(subject.id_field.to_string(), json!(subject.id));
655    data.insert("last_sequence".to_string(), json!(subject.sequence));
656    operations.push(PatchOperation::AddObject {
657        id: object_id.clone(),
658        object_type: subject.object_type.to_string(),
659        data: Value::Object(data),
660    });
661    operations.push(PatchOperation::AddRelation {
662        id: contains_relation_id(subject.raw_run_id, subject.kind, subject.id),
663        relation_type: "contains".to_string(),
664        source: subject.run_object_id.to_string(),
665        target: object_id,
666        data: json!({"kind": subject.kind}),
667    });
668    touch_run(graph, subject.run_object_id, subject.sequence, operations)
669}
670
671fn cancel_open_subjects(
672    graph: &crate::StateGraph,
673    run_id: &str,
674    sequence: u64,
675    operations: &mut Vec<PatchOperation>,
676) -> Result<(), RuntimeError> {
677    for relation in graph.relations_from(run_id) {
678        if relation.relation_type != "contains" {
679            continue;
680        }
681        let subject = graph.object(&relation.target).ok_or_else(|| {
682            RuntimeError::InvalidExternalProjection(format!(
683                "projected relation `{}` references missing object `{}`",
684                relation.id, relation.target
685            ))
686        })?;
687        let status = subject.data.get("status").and_then(Value::as_str);
688        match (subject.object_type.as_str(), status) {
689            ("workflow_step", Some("created" | "running" | "retrying")) => update_object(
690                graph,
691                &subject.id,
692                sequence,
693                [("status", json!("cancelled")), ("retry_after", Value::Null)],
694                operations,
695            )?,
696            ("workflow_wait", Some("waiting")) | ("workflow_hook", Some("waiting" | "active")) => {
697                update_object(
698                    graph,
699                    &subject.id,
700                    sequence,
701                    [("status", json!("cancelled"))],
702                    operations,
703                )?
704            }
705            _ => {}
706        }
707    }
708    Ok(())
709}
710
711fn touch_run(
712    graph: &crate::StateGraph,
713    run_id: &str,
714    sequence: u64,
715    operations: &mut Vec<PatchOperation>,
716) -> Result<(), RuntimeError> {
717    update_object(graph, run_id, sequence, [], operations)
718}
719
720fn update_object<const N: usize>(
721    graph: &crate::StateGraph,
722    id: &str,
723    sequence: u64,
724    fields: [(&str, Value); N],
725    operations: &mut Vec<PatchOperation>,
726) -> Result<(), RuntimeError> {
727    let object = graph.object(id).ok_or_else(|| {
728        RuntimeError::InvalidExternalProjection(format!("projected object `{id}` does not exist"))
729    })?;
730    let mut data: Map<String, Value> = object.data.as_object().cloned().unwrap_or_default();
731    for (key, value) in fields {
732        data.insert(key.to_string(), value);
733    }
734    data.insert("last_sequence".to_string(), json!(sequence));
735    operations.push(PatchOperation::UpdateObject {
736        id: id.to_string(),
737        expected_version: object.version,
738        data: Value::Object(data),
739    });
740    Ok(())
741}
742
743pub fn run_object_id(run_id: &str) -> String {
744    format!("flow:run:{run_id}")
745}
746pub fn step_object_id(run_id: &str, step_id: &str) -> String {
747    subject_object_id(run_id, "step", step_id)
748}
749fn subject_object_id(run_id: &str, kind: &str, id: &str) -> String {
750    format!("flow:{kind}:{run_id}:{id}")
751}
752fn contains_relation_id(run_id: &str, kind: &str, id: &str) -> String {
753    format!("flow:contains:{run_id}:{kind}:{id}")
754}
755
756#[cfg(test)]
757mod tests;
a3s_code_core/flow_graph/mod.rs

a3s_code_core/flow_graph/
mod.rs
