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

//! Live tool-call decision delivery for `Mailbox`.
//!
//! `send_decision` itself is a tiny sync helper that stays in `mailbox.rs`;
//! the async live-delivery path lives here so cancel.rs is not burdened with
//! decision-specific routing.

use awaken_server_contract::contract::lifecycle::RunStatus;
use awaken_server_contract::contract::mailbox::{
    LiveDeliveryOutcome, LiveRunCommand, LiveRunTarget, RunDispatchStatus,
};
use awaken_server_contract::contract::suspension::ToolCallResume;

use super::{Mailbox, MailboxError, live_target_for_dispatch, live_target_for_run};

impl Mailbox {
    /// Forward a tool-call decision locally or through targeted live delivery.
    ///
    /// Live delivery is at-least-once when the remote run accepts the decision
    /// but the ack is lost before the durable fallback is enqueued. Consumers
    /// must treat `(tool_call_id, decision_id)` as idempotent.
    pub async fn send_decision_live(
        &self,
        id: &str,
        tool_call_id: String,
        resume: ToolCallResume,
    ) -> Result<bool, MailboxError> {
        if self
            .executor
            .send_decision(id, tool_call_id.clone(), resume.clone())
        {
            self.record_mailbox_decision_received_for_id(id, &tool_call_id, &resume, "local_live")
                .await;
            return Ok(true);
        }

        if let Some(dispatch) = self.store.load_dispatch(id).await?
            && dispatch.status() == RunDispatchStatus::Claimed
        {
            let delivered = self
                .deliver_live_decision(
                    &live_target_for_dispatch(&dispatch),
                    vec![(tool_call_id.clone(), resume.clone())],
                )
                .await?;
            if delivered {
                self.record_mailbox_decision_received_for_dispatch(
                    &dispatch,
                    &tool_call_id,
                    &resume,
                    "remote_live",
                )
                .await;
            }
            return Ok(delivered);
        }

        let run = if let Some(run) = self.run_store.load_run(id).await? {
            Some(run)
        } else {
            self.run_store.latest_run(id).await?
        };
        if let Some(run) = run
            && matches!(run.status, RunStatus::Running | RunStatus::Waiting)
        {
            let delivered = self
                .deliver_live_decision(
                    &live_target_for_run(&run),
                    vec![(tool_call_id.clone(), resume.clone())],
                )
                .await?;
            if delivered {
                self.record_mailbox_decision_received_for_run(
                    &run,
                    &tool_call_id,
                    &resume,
                    "remote_live",
                )
                .await;
            }
            return Ok(delivered);
        }

        Ok(false)
    }

    async fn deliver_live_decision(
        &self,
        target: &LiveRunTarget,
        decisions: Vec<(String, ToolCallResume)>,
    ) -> Result<bool, MailboxError> {
        match self
            .store
            .deliver_live_to(target, LiveRunCommand::Decision(decisions))
            .await?
        {
            LiveDeliveryOutcome::Delivered => Ok(true),
            LiveDeliveryOutcome::NoSubscriber => Ok(false),
        }
    }
}