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

//! Event stream lifecycle helpers for workflow execution.
use crate::event_bus::{Event, EventBus, INVOCATION_END_SCOPE, STREAM_END_SCOPE};

/// Internal reason a workflow's event stream is being closed.
pub(crate) enum StreamEndReason {
    /// Workflow ran to completion.
    Completed {
        /// Final step number.
        step: u64,
    },
    /// Workflow halted due to an error.
    Error {
        /// Step at which the error occurred, if known.
        step: Option<u64>,
        /// Human-readable error description.
        error: String,
    },
}

impl StreamEndReason {
    fn format_message(&self, session_id: &str) -> String {
        match self {
            Self::Completed { step } => {
                format!("session={session_id} status=completed step={step}")
            }
            Self::Error { step, error } => match step {
                Some(s) => format!("session={session_id} status=error step={s} error={error}"),
                None => format!("session={session_id} status=error error={error}"),
            },
        }
    }
}

/// Emit a stream termination event and optionally close the event channel.
///
/// Called when a workflow completes or errors. Closes the channel only when
/// `event_stream_taken` is true, then resets the flag.
pub(crate) fn finalize_event_stream(
    event_bus: &EventBus,
    session_id: &str,
    reason: StreamEndReason,
    event_stream_taken: &mut bool,
) {
    let message = reason.format_message(session_id);

    if let Err(err) = event_bus
        .get_emitter()
        .emit(Event::diagnostic(STREAM_END_SCOPE, message.clone()))
    {
        tracing::debug!(
            session = %session_id,
            scope = STREAM_END_SCOPE,
            completion_message = %message,
            error = ?err,
            "failed to emit stream termination event"
        );
    }

    if *event_stream_taken {
        event_bus.close_channel();
        *event_stream_taken = false;
    }
}

/// Emit an invocation completion marker without closing the event channel.
pub(crate) fn emit_invocation_end(event_bus: &EventBus, session_id: &str, reason: StreamEndReason) {
    let message = reason.format_message(session_id);

    if let Err(err) = event_bus
        .get_emitter()
        .emit(Event::diagnostic(INVOCATION_END_SCOPE, message.clone()))
    {
        tracing::debug!(
            session = %session_id,
            scope = INVOCATION_END_SCOPE,
            completion_message = %message,
            error = ?err,
            "failed to emit invocation completion event"
        );
    }
}