monoloop-loop 0.1.2

Minimal extensible Loop: lossless canonical subscription, empty-capable tools
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
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116

//! D-026 / D-053: claim-gate fail-closed typing on StartedRuntime.
//!
//! Missing provider session id on ExternalAgent create → InvariantFailed
//! (must not remapped to Cancelled).

use monoloop_connector::{FakeConnectorConfig, FakeConnectorFactory, FakeSessionAdapterConfig};
use monoloop_contracts::{
    transaction_delivery, user_text_input, ChannelCapabilities, ChannelDefaults, ChannelId,
    ChannelKind, ChannelLimits, ContinuationPolicy, DeliveryLimits, DialectDescriptor,
    ExchangeMode, InvocationConfig, McpConfigurationCapability, McpReachability, OptionPolicy,
    SessionMode, ShutdownWaitOutcome, ToolExecutionMode, TransactionEndKind,
    TransactionSubmitRequest,
};
use monoloop_interpreter::DefaultInterpreterFactory;
use monoloop_loop::{
    ChannelBinding, ChannelRegistry, HostToolRegistry, RuntimeBootstrap, RuntimeConfig,
    StartedRuntime, TestTextEncoder,
};
use std::collections::BTreeSet;
use std::sync::Arc;
use std::time::Duration;

fn external_agent_omit_session(id: &str) -> ChannelBinding {
    let d = DialectDescriptor::test_raw();
    let connector_cfg = FakeConnectorConfig {
        omit_created_session_id: true,
        ..Default::default()
    };
    ChannelBinding {
        id: ChannelId::try_new(id).unwrap(),
        kind: ChannelKind::ExternalAgent,
        tool_mode: ToolExecutionMode::None,
        connector_factory: Arc::new(FakeConnectorFactory::external_agent_with_connector_config(
            FakeSessionAdapterConfig::default(),
            connector_cfg,
        )),
        encoder: Arc::new(TestTextEncoder),
        interpreter: Arc::new(DefaultInterpreterFactory::new()),
        endpoint_ref: "default".into(),
        credential_ref: None,
        defaults: ChannelDefaults::default(),
        capabilities: ChannelCapabilities {
            session_mode: SessionMode::External,
            mcp_configuration: McpConfigurationCapability::None,
            mcp_reachability: McpReachability::None,
            exchange_mode: ExchangeMode::Bidirectional,
            continuation_policies: BTreeSet::from([ContinuationPolicy::CallerControlled]),
            supports_distinct_session_concurrency: true,
            input_dialect: d.clone(),
            output_dialect: d,
            option_policy: OptionPolicy::external_agent(),
        },
        limits: ChannelLimits::default(),
    }
}

#[test]
fn create_without_provider_session_id_ends_invariant_failed() {
    let started = StartedRuntime::start(RuntimeBootstrap {
        config: RuntimeConfig {
            enable_mcp_listener: false,
            ..Default::default()
        },
        channels: ChannelRegistry::build(vec![external_agent_omit_session("agent")]).unwrap(),
        tools: HostToolRegistry::empty(),
    })
    .expect("start");
    let handle = started.handle.clone();

    let (delivery, receiver) =
        transaction_delivery(DeliveryLimits::try_new(32, 64 * 1024).unwrap()).unwrap();
    // `receiver` is mut for completion.recv(); events drained only if needed.
    handle
        .submit(TransactionSubmitRequest {
            channel_id: ChannelId::try_new("agent").unwrap(),
            session_id: None,
            input: user_text_input("hello").unwrap(),
            session_config: None,
            invocation_config: InvocationConfig {
                deadline: Some(Duration::from_secs(5)),
                continuation_policy: ContinuationPolicy::CallerControlled,
                ..Default::default()
            },
            tools: vec![],
            delivery,
        })
        .expect("admit");

    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap();
    let completion = rt
        .block_on(async {
            tokio::time::timeout(Duration::from_secs(5), receiver.completion.recv()).await
        })
        .expect("completion timeout")
        .expect("completion");

    assert_eq!(
        completion.end.kind,
        TransactionEndKind::InvariantFailed,
        "missing provider sessionId must not remapped to Cancelled; got {:?}",
        completion.end.kind
    );

    let mut owner = started.owner;
    let outcome = rt.block_on(async {
        owner.begin_shutdown();
        owner.wait_stopped(Duration::from_secs(2)).await
    });
    assert!(
        matches!(outcome, ShutdownWaitOutcome::Stopped(_)),
        "expected Stopped, got {outcome:?}"
    );
}