awaken-runtime 0.6.0

Phase-based execution engine, plugin system, and agent loop 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

use super::{BackgroundTaskManager, TaskHandle};
use crate::extensions::background::{BackgroundTaskStateKey, TaskId, TaskParentContext};

impl BackgroundTaskManager {
    fn parent_run_id(parent_context: &TaskParentContext) -> Option<&str> {
        parent_context
            .run_id
            .as_deref()
            .map(str::trim)
            .filter(|run_id| !run_id.is_empty())
    }

    pub(super) fn is_parent_run_cancelled(
        &self,
        parent_context: &TaskParentContext,
    ) -> Option<String> {
        let run_id = Self::parent_run_id(parent_context)?;
        self.cancelled_run_ids
            .read()
            .contains(run_id)
            .then(|| run_id.to_string())
    }

    pub(super) async fn insert_handle_and_cancel_if_parent_run_cancelled(
        &self,
        task_id: TaskId,
        handle: TaskHandle,
        parent_context: &TaskParentContext,
    ) {
        let mut handles = self.handles.lock().await;
        handles.insert(task_id.clone(), handle);
        if self.is_parent_run_cancelled(parent_context).is_some()
            && let Some(handle) = handles.get(&task_id)
        {
            handle.cancel_handle.cancel();
        }
    }

    pub async fn cancel_descendants_for_run(&self, parent_run_id: &str) -> usize {
        // Retain cancelled run ids for this manager's lifetime. This is what
        // rejects late descendant spawns after the one-shot live handle sweep.
        self.cancelled_run_ids
            .write()
            .insert(parent_run_id.to_string());
        let root_task_ids = self
            .store()
            .and_then(|store| store.read::<BackgroundTaskStateKey>())
            .map(|snapshot| {
                snapshot
                    .tasks
                    .values()
                    .filter(|meta| {
                        !meta.status.is_terminal()
                            && meta.parent_context.run_id.as_deref() == Some(parent_run_id)
                            && meta.parent_context.task_id.is_none()
                    })
                    .map(|meta| meta.task_id.clone())
                    .collect::<Vec<_>>()
            })
            .unwrap_or_default();

        let mut cancelled = 0usize;
        for task_id in root_task_ids {
            cancelled += self.cancel_tree(&task_id).await;
        }
        cancelled
    }
}