awaken-server 0.6.0

Multi-protocol HTTP server with SSE, mailbox, and protocol adapters for Awaken
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
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138

use std::sync::Arc;

use awaken_runtime::EventBuffer;
use awaken_server_contract::contract::commit_coordinator::CanonicalEventStager;
use awaken_server_contract::contract::event_sink::EventSink;
use awaken_server_contract::contract::mailbox::RunDispatch;
use awaken_server_contract::{
    AgentEventNormalizationContext, DurableEventSink, OutboxServerEventPublisher,
    RuntimeEventDurability, ScopedAgentEventNormalizer,
};

use crate::transport::channel_sink::ReconnectableEventSink;

use super::{Mailbox, MailboxError};

#[derive(Clone)]
pub(super) struct RuntimeEventCaptureConfig {
    pub(super) mode: RuntimeEventDurability,
    pub(super) origin: String,
}
impl Mailbox {
    /// Enable optional canonical runtime event capture for this mailbox.
    ///
    /// The wrapper preserves the existing public constructor shape: callers opt in
    /// by applying this builder before putting the mailbox behind an `Arc`.
    pub fn with_runtime_event_capture(
        mut self,
        mode: RuntimeEventDurability,
        origin: impl Into<String>,
    ) -> Result<Self, MailboxError> {
        let origin = origin.into();
        if matches!(mode, RuntimeEventDurability::Disabled) {
            self.runtime_event_capture = None;
            return Ok(self);
        }
        if !self.executor.has_commit_coordinator() {
            return Err(MailboxError::Validation(
                "runtime event capture requires an executor with CommitCoordinator".to_string(),
            ));
        }
        if self.staged_coordinator.is_none() {
            return Err(MailboxError::Validation(
                "runtime event capture requires an executor with StagedCommitCoordinator"
                    .to_string(),
            ));
        }
        AgentEventNormalizationContext::new("validation-thread", "validation-run", &origin)
            .map_err(|error| MailboxError::Validation(error.to_string()))?;
        self.runtime_event_capture = Some(RuntimeEventCaptureConfig { mode, origin });
        Ok(self)
    }

    /// Attach a publisher for advisory server-authored canonical events.
    pub fn with_server_event_publisher(
        mut self,
        publisher: Arc<dyn OutboxServerEventPublisher>,
        origin: impl Into<String>,
    ) -> Result<Self, MailboxError> {
        let origin = origin.into();
        AgentEventNormalizationContext::new("validation-thread", "validation-run", &origin)
            .map_err(|error| MailboxError::Validation(error.to_string()))?;
        self.server_event_publisher = Some(publisher);
        self.server_event_origin = origin;
        Ok(self)
    }

    fn wrap_runtime_event_sink(
        &self,
        inner: Arc<dyn EventSink>,
        thread_id: &str,
        run_id: &str,
        correlation_id: Option<String>,
        resumed: bool,
        event_buffer: Option<Arc<EventBuffer>>,
    ) -> Arc<dyn EventSink> {
        let Some(capture) = &self.runtime_event_capture else {
            return inner;
        };
        let context =
            AgentEventNormalizationContext::new(thread_id, run_id, capture.origin.clone()).expect(
                "mailbox runtime event capture context must use non-empty thread/run/origin",
            );
        let context = if let Some(c) = correlation_id {
            context.with_correlation_id(c)
        } else {
            context
        };
        let normalizer: Arc<ScopedAgentEventNormalizer> = Arc::new(if resumed {
            ScopedAgentEventNormalizer::new_resumed(context)
        } else {
            ScopedAgentEventNormalizer::new(context)
        });
        let buffer = event_buffer.expect("runtime event capture requires a per-run EventBuffer");
        Arc::new(DurableEventSink::new(
            inner,
            buffer as Arc<dyn CanonicalEventStager>,
            normalizer,
            capture.mode,
        ))
    }

    fn wrap_reconnectable_runtime_event_sink(
        &self,
        inner: Arc<ReconnectableEventSink>,
        thread_id: &str,
        run_id: &str,
        correlation_id: String,
        resumed: bool,
        event_buffer: Option<Arc<EventBuffer>>,
    ) -> Arc<dyn EventSink> {
        self.wrap_runtime_event_sink(
            inner as Arc<dyn EventSink>,
            thread_id,
            run_id,
            Some(correlation_id),
            resumed,
            event_buffer,
        )
    }

    pub(super) fn wrap_dispatch_runtime_event_sink(
        &self,
        inner: Arc<ReconnectableEventSink>,
        dispatch: &RunDispatch,
        correlation_id: String,
        resumed: bool,
        event_buffer: Option<Arc<EventBuffer>>,
    ) -> Arc<dyn EventSink> {
        self.wrap_reconnectable_runtime_event_sink(
            inner,
            &dispatch.thread_id(),
            &dispatch.run_id(),
            correlation_id,
            resumed,
            event_buffer,
        )
    }
}