harn-serve 0.10.122

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
110
111

use super::*;

struct FailingFlushSink;

impl harn_vm::agent_events::AgentEventSink for FailingFlushSink {
    fn handle_event(&self, _event: &harn_vm::agent_events::AgentEvent) {}

    fn flush(&self) -> harn_vm::agent_events::AgentEventSinkFlush<'_> {
        Box::pin(async {
            Err(harn_vm::agent_events::AgentEventSinkError::new(
                "a2a_test",
                "injected append failure",
            ))
        })
    }
}

fn test_task_params(text: &str) -> JsonValue {
    json!({
        "message": {
            "metadata": {"target_agent": "triage"},
            "parts": [{"type": "text", "text": text}]
        }
    })
}

#[tokio::test]
async fn waits_for_sink_failure_before_completing_and_clears_registration() {
    let (_dir, server) = test_server(
        r#"
import { agent_progress } from "std/agent/progress"

pub fn triage(harness: Harness, task: string) -> string {
  agent_progress(harness.agent, {message: "Persist this progress."})
  return task
}
"#,
    );
    let task = server
        .prepare_task(&test_task_params("hello"), AuthRequest::default())
        .await
        .unwrap_or_else(|_| panic!("prepare task"));
    let session_id = super::a2a_worker_session_id(&task.id);
    harn_vm::agent_events::register_sink(session_id.clone(), Arc::new(FailingFlushSink));

    server.run_task_to_completion(&task).await;

    let task_json = server.task_json(&task.id);
    assert_eq!(task_json["status"]["state"], "failed");
    assert!(task_json["history"]
        .as_array()
        .expect("task history")
        .iter()
        .any(|message| message.to_string().contains("injected append failure")));
    assert_eq!(
        harn_vm::agent_events::session_external_sink_count(&session_id),
        0
    );
}

#[tokio::test]
async fn cancellation_keeps_sink_failure_without_losing_cancelled_status() {
    let (_dir, server) = test_server(
        r#"
import { agent_progress } from "std/agent/progress"

pub fn triage(harness: Harness, task: string) -> string {
  agent_progress(harness.agent, {message: "Ready for cancellation."})
  while true {
    if is_cancelled() {
      return task
    }
  }
}
"#,
    );
    let task = server
        .prepare_task(&test_task_params("cancel me"), AuthRequest::default())
        .await
        .unwrap_or_else(|_| panic!("prepare task"));
    let session_id = super::a2a_worker_session_id(&task.id);
    harn_vm::agent_events::register_sink(session_id.clone(), Arc::new(FailingFlushSink));
    let mut events = server.subscribe(&task.id).expect("subscribe to task");
    let task_id = task.id.clone();
    let runner = {
        let server = server.clone();
        tokio::spawn(async move {
            server.run_task_to_completion(&task).await;
        })
    };

    while let Some(event) = events.next().await {
        if event.pointer("/result/kind").and_then(JsonValue::as_str) == Some("status-update") {
            break;
        }
    }
    let cancelled = server.cancel_task(&task_id).expect("cancel running task");
    assert_eq!(cancelled["status"]["state"], "cancelled");
    runner.await.expect("task runner");

    let task_json = server.task_json(&task_id);
    assert_eq!(task_json["status"]["state"], "cancelled");
    assert!(task_json["metadata"]["harn"]["persistenceError"]
        .as_str()
        .expect("cancelled task persistence error")
        .contains("injected append failure"));
    assert_eq!(
        harn_vm::agent_events::session_external_sink_count(&session_id),
        0
    );
}