met 1.0.0-beta-preview

Metrale Engine: pure Rust LLM inference server (HTTP API, scheduler, TUI, CLI).
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
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140

// SPDX-License-Identifier: MIT OR Apache-2.0

//! 2026-09-25: Tests that a prefill failing after the response sink was
//! taken reaches the request: an `Error` frame on a stream, an `Err` on a
//! blocking call. `handle_prefill_start_error` has no access to the failing
//! request's sink, so `start_chunked_prefill` must deliver it; these tests
//! call it with a model whose `prefill_chunk` fails.
//!
//! Owner: scheduler.
//! Invariants: none beyond the types.

use std::sync::Arc;
use std::sync::atomic::AtomicBool;

use futures::StreamExt;

use super::test_support::PreemptStubModel;
use crate::api::InferenceRequest;
use crate::api::inference_types::StreamEvent;

/// 2026-09-25: One request body; only `$variant` and its sink fields vary.
macro_rules! request {
    ($variant:ident { $($sink:tt)* }) => {
        InferenceRequest::$variant {
            prompt_tokens: Arc::new(vec![1, 2, 3]),
            session_hash: 0,
            lookup_tenant: None,
            adapter_slot: -1,
            src_lang_id: 0,
            tgt_lang_id: 0,
            num_beams: 1,
            length_penalty: 1.0,
            early_stopping: false,
            image_pixels: vec![],
            max_tokens: 4,
            min_tokens: 0,
            temperature: 0.0,
            top_k: 0,
            top_p: 1.0,
            top_n_sigma: 0.0,
            min_p: 0.0,
            repetition_penalty: 1.0,
            presence_penalty: 0.0,
            frequency_penalty: 0.0,
            dry_multiplier: 0.0,
            dry_base: 0.0,
            dry_allowed_length: 0,
            lz_penalty: 0.0,
            logit_bias: vec![],
            stop_tokens: vec![],
            enable_thinking: false,
            thinking_budget: None,
            repetition_detection: None,
            require_tool_call: false,
            tools_present: false,
            suppress_tool_call: false,
            disable_mtp: true,
            grammar_spec: None,
            seed: Some(42),
            top_logprobs: None,
            prompt_logprobs: None,
            echo: false,
            timeout_at: None,
            $($sink)*
        }
    };
}

/// 2026-09-25: Call `start_chunked_prefill`, not deferred, with a model
/// whose `prefill_chunk` fails.
fn start(req: InferenceRequest) -> anyhow::Result<()> {
    let mut engine = None;
    super::prefill_a_step::start_chunked_prefill(
        &super::sched_ctx::SchedCtx::for_test(),
        None,
        None,
        None,
        None,
        &PreemptStubModel::default(),
        req,
        &[],
        64,
        0,
        0,
        &mut engine,
        Some(0),
        false,
        None,
        None,
    )
    .map(|_| ())
}

#[tokio::test]
async fn a_streaming_prefill_start_error_reaches_the_client_as_an_error_frame() {
    let (token_tx, token_rx) = tokio::sync::mpsc::channel::<StreamEvent>(16);
    let err = start(request!(Streaming {
        token_tx,
        cancel_flag: Arc::new(AtomicBool::new(false)),
    }))
    .expect_err("the stub's prefill_chunk refuses");
    assert!(
        format!("{err:#}").contains("unused in preempt tests"),
        "{err:#}"
    );

    // 2026-09-25: Read the channel through `stream_terminal::terminated`,
    // as the SSE layer does, so the test sees the frames the client sees.
    let events: Vec<StreamEvent> = crate::api::stream_terminal::terminated(
        tokio_stream::wrappers::ReceiverStream::new(token_rx),
    )
    .collect()
    .await;
    let [StreamEvent::Error(msg)] = events.as_slice() else {
        panic!(
            "expected exactly one Error frame, got {} events",
            events.len()
        );
    };
    assert!(
        msg.starts_with("prefill_chunk failed:") && msg.contains("unused in preempt tests"),
        "the frame must carry the real reason, not the no-result backstop: {msg}"
    );
}

#[tokio::test]
async fn a_blocking_prefill_start_error_reaches_the_caller_as_an_err() {
    let (response_tx, response_rx) = tokio::sync::oneshot::channel();
    start(request!(Blocking { response_tx })).expect_err("the stub's prefill_chunk refuses");
    let got = response_rx
        .await
        .expect("the sink must be answered, not dropped (a drop reads as 'Inference cancelled')");
    let Err(e) = got else {
        panic!("a failed prefill must not produce a response");
    };
    assert!(
        format!("{e:#}").starts_with("prefill_chunk failed:"),
        "{e:#}"
    );
}