harn-serve 0.10.121

Shared outbound workflow server core for Harn adapters
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

use super::*;

pub(super) struct ResetActiveEventLog;

impl Drop for ResetActiveEventLog {
    fn drop(&mut self) {
        harn_vm::event_log::reset_active_event_log();
    }
}

#[test]
fn cancelled_prompt_retains_persistence_failure_metadata() {
    let error =
        harn_vm::agent_events::AgentEventSinkError::new("event_log", "injected append failure");

    let result = super::super::prompt::cancelled_prompt_result(Some(&error));

    assert_eq!(result["stopReason"], "cancelled");
    assert!(result["_meta"]["harn"]["persistenceError"]
        .as_str()
        .expect("persistence error metadata")
        .contains("injected append failure"));
}

#[tokio::test(flavor = "current_thread")]
async fn completed_prompt_is_durable_before_immediate_session_load() {
    let local = tokio::task::LocalSet::new();
    local
        .run_until(async {
            let _reset = ResetActiveEventLog;
            let _log = harn_vm::event_log::install_memory_for_current_thread(32);
            let (request_tx, mut response_rx, server, session_id) =
                start_acp_channel_session().await;
            let prompt = format!(
                "__host_agent_emit_event(\"{session_id}\", \"progress_reported\", \
                 {{message: \"durable before prompt response\"}})"
            );

            request_tx
                .send(serde_json::json!({
                    "jsonrpc": "2.0",
                    "id": 2,
                    "method": "session/prompt",
                    "params": {
                        "sessionId": session_id,
                        "prompt": [{"type": "text", "text": prompt}],
                    },
                }))
                .expect("send session/prompt");

            loop {
                let message = recv_json(&mut response_rx).await;
                if message["method"] == "host/capabilities" {
                    request_tx
                        .send(serde_json::json!({
                            "jsonrpc": "2.0",
                            "id": message["id"].clone(),
                            "result": {},
                        }))
                        .expect("send host capabilities response");
                } else if message["id"] == 2 {
                    assert_eq!(message["result"]["stopReason"], "end_turn");
                    break;
                }
            }

            request_tx
                .send(serde_json::json!({
                    "jsonrpc": "2.0",
                    "id": 3,
                    "method": "session/load",
                    "params": {"sessionId": session_id},
                }))
                .expect("send immediate session/load");

            let mut replay_notification = None;
            let loaded = loop {
                let message = recv_json(&mut response_rx).await;
                if message["id"] == 3 {
                    break message;
                }
                if message["method"] == "session/update" && message["_harn"]["replayed"] == true {
                    replay_notification = Some(message);
                }
            };

            let replay_notification =
                replay_notification.expect("session/load must replay the prompt event");
            assert_eq!(
                replay_notification["params"]["update"]["_meta"]["harn"]["message"],
                "durable before prompt response"
            );
            let replayed = loaded["result"]["replayed"]
                .as_array()
                .expect("session/load replay list");
            assert_eq!(
                replayed
                    .iter()
                    .filter(|event| event["type"] == "progress_reported")
                    .count(),
                1,
                "the completed prompt event must be persisted exactly once"
            );

            drop(request_tx);
            server.await.expect("ACP channel server task");
        })
        .await;
}