laser-sdk 0.3.2

LaserData SDK over Apache Iggy: typed streaming, operational and lakehouse query, materialization destinations, key-value state, copy-on-write forks, and an optional agent runtime with the Agent Data Exchange Protocol (AGDX).
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

use crate::harness;
use bytes::Bytes;
use laser_sdk::prelude::full::*;
use std::time::Duration;

struct ToolRunner;

impl AgentHandler for ToolRunner {
    async fn handle(&self, message: &AgentMessage, ctx: &AgentCtx<'_>) -> Result<(), LaserError> {
        let output = Bytes::from(format!(
            "result: {}",
            String::from_utf8_lossy(&message.payload)
        ));
        ctx.reply_on(AgentTopic::ToolResults, output).await
    }
}

#[tokio::test]
#[serial_test::serial(integration)]
async fn given_a_tool_runner_when_requesting_then_should_await_the_correlated_reply() {
    let laser = harness::laser().await;
    Agent::builder()
        .id("tool".parse().expect("tool is a valid agent id"))
        .listen_on(AgentTopic::ToolCalls)
        .handler(ToolRunner)
        .build()
        .spawn(laser.clone());

    let correlation = Provenance::builder()
        .conversation_id(ConversationId::new())
        .build();
    let reply = laser
        .request(
            AgentTopic::ToolCalls,
            AgentTopic::ToolResults,
            Bytes::from_static(b"search"),
            &correlation,
            Duration::from_secs(10),
        )
        .await
        .expect("the tool result should arrive before the timeout");

    assert_eq!(reply.payload.as_slice(), b"result: search");
    assert_eq!(
        reply
            .provenance
            .agent
            .as_ref()
            .expect("agent should be set")
            .as_str(),
        "tool"
    );
}

#[tokio::test]
#[serial_test::serial(integration)]
async fn given_no_responder_when_requesting_then_should_time_out() {
    let laser = harness::laser().await;
    let correlation = Provenance::builder()
        .conversation_id(ConversationId::new())
        .build();
    let result = laser
        .request(
            AgentTopic::ToolCalls,
            AgentTopic::ToolResults,
            Bytes::from_static(b"unanswered"),
            &correlation,
            Duration::from_millis(300),
        )
        .await;
    assert!(matches!(result, Err(LaserError::Timeout("reply"))));
}