unified-agent-api 0.2.2

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

use super::*;

use super::support::ControlledEndStream;

#[tokio::test]
async fn cancellation_closes_events_but_does_not_accelerate_completion() {
    let adapter = std::sync::Arc::new(ToyAdapter { fail_spawn: false });
    let cancel = HarnessCancelSignal::new();

    let (backend_done_tx, backend_done_rx) = oneshot::channel::<()>();
    let completion: DynBackendCompletionFuture<_, _> = Box::pin(async move {
        backend_done_rx.await.expect("backend completion released");
        Ok::<ToyCompletion, ToyBackendError>(ToyCompletion)
    });

    let (events_finish_tx, events_finish_rx) = oneshot::channel::<()>();
    let events = ControlledEndStream::<ToyEvent, ToyBackendError> {
        first: Some(Ok::<ToyEvent, ToyBackendError>(ToyEvent::Text(
            "one".to_string(),
        ))),
        finish_rx: events_finish_rx,
    };
    let events: DynBackendEventStream<_, _> = Box::pin(events);

    let (tx, rx) = mpsc::channel::<crate::AgentWrapperEvent>(8);
    let (completion_tx, completion_rx) =
        oneshot::channel::<Result<AgentWrapperCompletion, AgentWrapperError>>();

    let pump_task = tokio::spawn(pump_backend_events_with_cancel(
        adapter.clone(),
        events,
        tx,
        cancel.clone(),
        None,
    ));
    let completion_task = tokio::spawn(send_completion_with_cancel(
        adapter,
        completion,
        cancel.clone(),
        None,
        completion_tx,
    ));

    let mut handle = crate::run_handle_gate::build_gated_run_handle(rx, completion_rx);

    let first = handle.events.next().await.expect("first event forwarded");
    assert_eq!(first.kind, AgentWrapperEventKind::TextOutput);

    cancel.cancel();

    assert!(
        tokio::time::timeout(Duration::from_secs(1), handle.events.next())
            .await
            .expect("events do not hang")
            .is_none(),
        "events stream must be closed after cancellation"
    );

    {
        let waker = noop_waker();
        let mut cx = Context::from_waker(&waker);
        assert!(matches!(
            handle.completion.as_mut().poll(&mut cx),
            Poll::Pending
        ));
    }

    let _ = backend_done_tx.send(());
    let outcome = handle.completion.await;
    assert!(
        matches!(
            outcome,
            Err(AgentWrapperError::Backend { ref message }) if message == "cancelled"
        ),
        "completion must resolve to the pinned cancelled error after backend exit"
    );

    let _ = events_finish_tx.send(());
    pump_task.await.expect("pump task completes");
    completion_task.await.expect("completion task completes");
}