unified-agent-api 0.3.5

Agent-agnostic facade and registry for wrapper backends
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

use super::*;

use super::support::{CountingStream, GatedCountingStream};

#[tokio::test]
async fn pump_with_cancel_closes_universal_stream_but_still_drains_typed_stream() {
    let total = 3usize;
    let consumed = std::sync::Arc::new(AtomicUsize::new(0));
    let (gate_tx, gate_rx) = oneshot::channel::<()>();

    let stream = GatedCountingStream {
        first: Some(Ok::<ToyEvent, ToyBackendError>(ToyEvent::Text(
            "ev-0".to_string(),
        ))),
        rest: (1..total)
            .map(|i| Ok::<ToyEvent, ToyBackendError>(ToyEvent::Text(format!("ev-{i}"))))
            .collect(),
        gate_rx: Some(gate_rx),
        consumed: consumed.clone(),
    };
    let events: DynBackendEventStream<_, _> = Box::pin(stream);

    let adapter = std::sync::Arc::new(ToyAdapter { fail_spawn: false });
    let (tx, mut rx) = mpsc::channel::<crate::AgentWrapperEvent>(8);
    let cancel = HarnessCancelSignal::new();
    let handle = tokio::spawn(pump_backend_events_with_cancel(
        adapter,
        events,
        tx,
        cancel.clone(),
        None,
    ));

    let first = rx.recv().await.expect("at least one forwarded event");
    assert_eq!(first.kind, AgentWrapperEventKind::TextOutput);

    cancel.cancel();

    assert!(
        tokio::time::timeout(Duration::from_secs(1), rx.recv())
            .await
            .expect("recv does not hang")
            .is_none(),
        "universal stream must close after cancellation"
    );

    let _ = gate_tx.send(());
    handle.await.expect("pump task completes");
    assert_eq!(
        consumed.load(Ordering::SeqCst),
        total,
        "typed backend stream must be drained to end after cancellation"
    );
}

#[tokio::test]
async fn pump_with_cancel_preserves_drain_on_drop_posture() {
    let total = 20usize;
    let consumed = std::sync::Arc::new(AtomicUsize::new(0));
    let items: VecDeque<Result<ToyEvent, ToyBackendError>> = (0..total)
        .map(|i| Ok::<ToyEvent, ToyBackendError>(ToyEvent::Text(format!("ev-{i}"))))
        .collect();

    let events = CountingStream {
        items,
        consumed: consumed.clone(),
    };
    let events: DynBackendEventStream<_, _> = Box::pin(events);

    let adapter = std::sync::Arc::new(ToyAdapter { fail_spawn: false });
    let (tx, mut rx) = mpsc::channel::<crate::AgentWrapperEvent>(1);
    let cancel = HarnessCancelSignal::new();
    let handle = tokio::spawn(pump_backend_events_with_cancel(
        adapter, events, tx, cancel, None,
    ));

    let first = rx.recv().await.expect("at least one forwarded event");
    assert_eq!(first.kind, AgentWrapperEventKind::TextOutput);
    drop(rx);

    handle.await.expect("pump task completes");
    assert_eq!(
        consumed.load(Ordering::SeqCst),
        total,
        "backend stream must be fully drained after receiver drop"
    );
}