harn-vm 0.10.129

Async bytecode virtual machine for the Harn programming language
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

use crate::value::{VmJoinHandle, VmTaskHandle, VmValue};

pub(super) type VmTaskJoinResult =
    Result<Result<(VmValue, String), crate::value::VmError>, tokio::task::JoinError>;

/// Preserve the durable lifecycle owner whenever a joined task did not
/// complete successfully. Every join path delegates this decision here so a
/// newly added await/cancel surface cannot silently strand the task's journal.
pub(super) fn finish_task_join(
    joined: VmTaskJoinResult,
    task_id: String,
    runtimes: crate::agent_lifecycle_cleanup::CleanupRuntimes,
) -> VmTaskJoinResult {
    if !matches!(&joined, Ok(Ok(_))) {
        schedule_task_cleanup(task_id, runtimes);
    }
    joined
}

/// Stop a task from a synchronous unwind boundary, then finish its durable
/// agent lifecycle on an inherited runtime child. The journal and writer lease
/// remain visible if cleanup fails, so absence never masquerades as success.
pub(crate) fn abort_task_detached(
    task: VmTaskHandle,
    runtimes: crate::agent_lifecycle_cleanup::CleanupRuntimes,
) {
    let task_id = task.wait_task_id.clone();
    // A Tokio JoinHandle remains a value after yielding Ready, but polling it
    // again panics. Cleanup needs the task identity, not its discarded output,
    // so never hand an already-finished handle to the waiter.
    if task.handle.is_finished() {
        schedule_task_cleanup(task_id, runtimes);
        return;
    }
    task.cancel_token
        .store(true, std::sync::atomic::Ordering::SeqCst);
    task.handle.abort();
    schedule_task_cleanup_after(task_id, runtimes, async move {
        let _ = task.handle.await;
    });
}

/// Activate the process-owned cleanup reservation established before the agent
/// session started. Recovery retries until the terminal write commits; it has
/// no late admission or retry-exhaustion path that can discard the owner.
pub(crate) fn schedule_task_cleanup(
    task_id: String,
    runtimes: crate::agent_lifecycle_cleanup::CleanupRuntimes,
) {
    schedule_task_cleanup_after(task_id, runtimes, std::future::ready(()));
}

fn schedule_task_cleanup_after<F>(
    task_id: String,
    runtimes: crate::agent_lifecycle_cleanup::CleanupRuntimes,
    before_cleanup: F,
) where
    F: std::future::Future<Output = ()> + Send + 'static,
{
    crate::vm::subtask::spawn_lifecycle_cleanup(async move {
        before_cleanup.await;
        crate::agent_lifecycle_cleanup::schedule(task_id, runtimes);
    });
}

pub(super) async fn abort_task_and_wait(
    mut task: VmTaskHandle,
    execution_id: &str,
) -> Result<(), crate::value::VmError> {
    let task_id = task.wait_task_id.clone();
    task.cancel_token
        .store(true, std::sync::atomic::Ordering::SeqCst);
    abort_join_and_wait(&mut task.handle).await;
    crate::llm::agent_session_host::cancellation::abandon_task_sessions(execution_id, &task_id)
        .await
}

pub(super) async fn abort_join_and_wait(handle: &mut VmJoinHandle) {
    handle.abort();
    let _ = handle.await;
}

pub(super) struct AwaitingTask {
    task: Option<VmTaskHandle>,
    runtimes: crate::agent_lifecycle_cleanup::CleanupRuntimes,
}

impl AwaitingTask {
    pub(super) fn new(
        task: VmTaskHandle,
        runtimes: crate::agent_lifecycle_cleanup::CleanupRuntimes,
    ) -> Self {
        Self {
            task: Some(task),
            runtimes,
        }
    }

    /// Await the child while retaining cancellation ownership until its join
    /// handle resolves. A failed join still needs lifecycle cleanup, but the
    /// completed handle must never be polled a second time by the detached
    /// cleanup path.
    pub(super) async fn join(mut self) -> VmTaskJoinResult {
        let joined = (&mut self.task.as_mut().expect("awaiting task present").handle).await;
        let task = self.task.take().expect("awaiting task present after join");
        finish_task_join(joined, task.wait_task_id, self.runtimes.clone())
    }
}

impl Drop for AwaitingTask {
    fn drop(&mut self) {
        if let Some(task) = self.task.take() {
            abort_task_detached(task, self.runtimes.clone());
        }
    }
}

#[cfg(test)]
#[path = "task_cleanup_tests.rs"]
mod tests;

pub(super) enum StepPreHookAction {
    Allow(Vec<VmValue>),
    Deny(String),
}