weavegraph 0.7.0

Graph-driven, concurrent agent workflow framework with versioned state, deterministic barrier merges, and rich diagnostics.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108

//! Feature-gated [`RuntimeObserver`] backed by the [`metrics`] crate.
//!
//! Enable with `features = ["metrics"]`. Metrics are forwarded to whatever
//! [`metrics`]-compatible recorder is installed (e.g. `metrics-exporter-prometheus`).
//!
//! ## Metric inventory
//!
//! | Metric | Kind | Labels | Description |
//! |--------|------|--------|-------------|
//! | `weavegraph.node.invocations` | counter | `node`, `outcome` | Completed node executions |
//! | `weavegraph.node.step_duration_ms` | histogram | `node` | Superstep wall-clock duration |
//! | `weavegraph.invocation.count` | counter | `outcome` | Completed workflow invocations |
//! | `weavegraph.invocation.duration_ms` | histogram | — | Invocation wall-clock duration |
//! | `weavegraph.checkpoint.saves` | counter | `backend` | Successful checkpoint saves |
//! | `weavegraph.checkpoint.save_duration_ms` | histogram | `backend` | Checkpoint save duration |
//! | `weavegraph.checkpoint.loads` | counter | `backend` | Sessions resumed from checkpoint |
//! | `weavegraph.event_bus.emits` | counter | `scope` | Events emitted through the bus |
//!
//! Per-session and per-invocation identifiers are deliberately excluded from labels
//! to keep cardinality bounded in long-running services.
//!
//! ## Usage
//!
//! ```rust,no_run
//! use std::sync::Arc;
//! use weavegraph::runtimes::{AppRunner, metrics_observer::MetricsObserver};
//! # use weavegraph::app::App;
//!
//! # async fn example(app: App) {
//! let runner = AppRunner::builder()
//!     .app(app)
//!     .observer(Arc::new(MetricsObserver))
//!     .build()
//!     .await;
//! # }
//! ```

use std::panic::RefUnwindSafe;

use crate::runtimes::observer::{
    CheckpointLoadMeta, CheckpointSaveMeta, EventBusEmitMeta, InvocationFinishMeta,
    InvocationOutcome, NodeFinishMeta, NodeOutcome, RuntimeObserver,
};

/// A [`RuntimeObserver`] that forwards lifecycle events to the [`metrics`] crate.
///
/// Install a compatible recorder before starting the runner; see the
/// [module documentation](self) for the full metric inventory.
#[derive(Debug, Clone, Copy)]
pub struct MetricsObserver;

impl RefUnwindSafe for MetricsObserver {}

impl RuntimeObserver for MetricsObserver {
    fn on_invocation_finish(&self, meta: &InvocationFinishMeta<'_>) {
        let outcome = match meta.outcome {
            InvocationOutcome::Completed => "completed",
            InvocationOutcome::Error => "error",
        };
        metrics::counter!("weavegraph.invocation.count", "outcome" => outcome).increment(1);
        metrics::histogram!("weavegraph.invocation.duration_ms").record(meta.duration_ms as f64);
    }

    fn on_node_finish(&self, meta: &NodeFinishMeta<'_>) {
        let node = meta.node_kind.encode().to_string();
        let outcome = match meta.outcome {
            NodeOutcome::Completed => "completed",
            NodeOutcome::Skipped => "skipped",
            NodeOutcome::Error => "error",
        };
        if meta.outcome != NodeOutcome::Skipped {
            metrics::histogram!("weavegraph.node.step_duration_ms", "node" => node.clone())
                .record(meta.step_duration_ms as f64);
        }
        metrics::counter!(
            "weavegraph.node.invocations",
            "node" => node,
            "outcome" => outcome
        )
        .increment(1);
    }

    fn on_checkpoint_load(&self, meta: &CheckpointLoadMeta<'_>) {
        metrics::counter!(
            "weavegraph.checkpoint.loads",
            "backend" => meta.backend.to_string()
        )
        .increment(1);
    }

    fn on_checkpoint_save(&self, meta: &CheckpointSaveMeta<'_>) {
        let backend = meta.backend.to_string();
        metrics::histogram!(
            "weavegraph.checkpoint.save_duration_ms",
            "backend" => backend.clone()
        )
        .record(meta.duration_ms as f64);
        metrics::counter!("weavegraph.checkpoint.saves", "backend" => backend).increment(1);
    }

    fn on_event_bus_emit(&self, meta: &EventBusEmitMeta<'_>) {
        metrics::counter!(
            "weavegraph.event_bus.emits",
            "scope" => meta.scope.to_string()
        )
        .increment(1);
    }
}