Compare commits
19
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
03fe47e91b
|
||
|
|
ae994e3ca5
|
||
|
|
f31b599f30
|
||
|
|
d94b780e5b
|
||
|
|
af34741aab
|
||
|
|
e048743592
|
||
|
|
df5766396c
|
||
|
|
f5fd678247
|
||
|
|
acef01bc4c
|
||
|
|
441cddea37
|
||
|
|
74df8fd07f
|
||
|
|
6945cf9871
|
||
|
|
7f9e240a78
|
||
|
|
e9b43b572e
|
||
|
|
9d9c9be78a
|
||
|
|
b9355e0248
|
||
|
|
e739507803
|
||
|
|
ffafe3f4dc
|
||
|
|
91255619b6
|
@@ -262,7 +262,7 @@ jobs:
|
||||
# fast instead of triggering a slower uncached rebuild; only a real
|
||||
# sccache fault drops the cache. See script/ci-cargo-escalate.sh.
|
||||
- name: Clippy (sccache escalation)
|
||||
run: script/ci-cargo-escalate.sh cargo clippy --workspace -- -D warnings
|
||||
run: script/ci-cargo-escalate.sh cargo clippy --workspace --all-targets -- -D warnings
|
||||
|
||||
test:
|
||||
name: Test
|
||||
|
||||
@@ -179,8 +179,13 @@ jobs:
|
||||
# in sync with build-prerelease.yml): a signal death (rustc SIGSEGV
|
||||
# / OOM-kill) keeps the cache and fails fast instead of an uncached
|
||||
# rebuild; only a real sccache fault drops the cache.
|
||||
# --all-targets so `#[cfg(test)]` modules, tests/, benches/ and
|
||||
# examples/ are linted too (#286). Without it clippy sees only lib
|
||||
# and bin targets, so `-D warnings` never gated a single line of
|
||||
# test code — which is how two tests lost their `#[test]` attribute
|
||||
# and stopped running unnoticed.
|
||||
- name: Clippy (sccache escalation)
|
||||
run: script/ci-cargo-escalate.sh cargo clippy --workspace -- -D warnings
|
||||
run: script/ci-cargo-escalate.sh cargo clippy --workspace --all-targets -- -D warnings
|
||||
|
||||
test:
|
||||
name: Test
|
||||
|
||||
@@ -167,7 +167,7 @@ doc-comment; touch points (the `Job` enum, the dispatch handlers, the
|
||||
cargo build --release # build all crates
|
||||
cargo run -p cortex-cli -- serve # run the gateway
|
||||
cargo test # run all tests
|
||||
cargo clippy --workspace # lint
|
||||
cargo clippy --workspace --all-targets # lint (test code included)
|
||||
```
|
||||
|
||||
## CI
|
||||
@@ -177,7 +177,7 @@ pass before merging:
|
||||
|
||||
```sh
|
||||
cargo fmt --check --all # formatting
|
||||
cargo clippy --workspace -- -D warnings # lint (warnings are errors)
|
||||
cargo clippy --workspace --all-targets -- -D warnings # lint (warnings are errors)
|
||||
cargo test --workspace # tests
|
||||
```
|
||||
|
||||
@@ -191,7 +191,8 @@ Work each change on its own branch; `main` stays releasable.
|
||||
|
||||
1. Implement on a feature branch (`fix/<issue>-…`, `feat/<issue>-…`).
|
||||
2. Run the CI triad locally (`cargo fmt --check --all`,
|
||||
`cargo clippy --workspace -- -D warnings`, `cargo test --workspace`).
|
||||
`cargo clippy --workspace --all-targets -- -D warnings`,
|
||||
`cargo test --workspace`).
|
||||
Local builds are **CPU-only** — the `#[cfg(feature = "cuda")]` neuron/TP
|
||||
paths do NOT compile locally. The branch CI's **CUDA type-check** job is
|
||||
the only thing that validates them, so for any neuron change the push to
|
||||
|
||||
@@ -50,8 +50,48 @@ pub struct ModelLimit {
|
||||
/// back to `context − output`.
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub input: Option<usize>,
|
||||
/// Maximum number of generation tokens.
|
||||
/// The generation budget a request gets when it names none, and the
|
||||
/// figure `input = context − output` is derived from.
|
||||
///
|
||||
/// A *reserve*, not a ceiling — a request may ask for more and be
|
||||
/// served it. Named `output` for the shape's history; the number a
|
||||
/// client should plan against is [`ModelLimit::output_ceiling`].
|
||||
pub output: usize,
|
||||
/// The largest output a request may name and be served (#278).
|
||||
///
|
||||
/// Distinct from `output`, which is the default and the KV-planning
|
||||
/// reserve. Publishing the reserve as the ceiling meant a client
|
||||
/// that trusted the advertisement asked for a fraction of what the
|
||||
/// model would happily generate — on a reasoning model, often less
|
||||
/// than its own think block costs.
|
||||
///
|
||||
/// `#[serde(default)]` so a neuron predating this field still
|
||||
/// deserializes; zero means "not advertised", and consumers fall
|
||||
/// back to `output`.
|
||||
#[serde(default, skip_serializing_if = "is_zero")]
|
||||
pub output_ceiling: usize,
|
||||
}
|
||||
|
||||
fn is_zero(v: &usize) -> bool {
|
||||
*v == 0
|
||||
}
|
||||
|
||||
/// What each reasoning effort level costs, in reasoning tokens (#223).
|
||||
///
|
||||
/// OpenAI-shaped clients can only send `minimal|low|medium|high` — the
|
||||
/// ladder is their entire vocabulary, and they cannot express a token
|
||||
/// count. So the server picks the numbers, and has to say what they are:
|
||||
/// a client choosing `low` with no idea whether that buys 2k tokens or
|
||||
/// 20k is guessing, which is the failure #274 exists to end.
|
||||
///
|
||||
/// Ordered rungs rather than a map, so the ladder reads in the order a
|
||||
/// client would climb it.
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub struct ReasoningBudgetRung {
|
||||
/// The effort level as the caller spells it on the wire.
|
||||
pub effort: String,
|
||||
/// Reasoning tokens that level is allowed to spend.
|
||||
pub tokens: usize,
|
||||
}
|
||||
|
||||
/// Operator-set pricing, **USD per 1,000,000 tokens, as JSON numbers**
|
||||
@@ -144,6 +184,11 @@ pub struct ModelInfo {
|
||||
/// from its own routing table.
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub servable: Option<ModelServability>,
|
||||
/// The reasoning-effort ladder this deployment honours (#223).
|
||||
/// Absent when the model does not reason, or the harness has no
|
||||
/// budget to offer.
|
||||
#[serde(default, skip_serializing_if = "Vec::is_empty")]
|
||||
pub reasoning_budget: Vec<ReasoningBudgetRung>,
|
||||
}
|
||||
|
||||
/// Why a loaded model can or cannot be served right now (#245).
|
||||
|
||||
@@ -44,6 +44,12 @@ pub struct NodeState {
|
||||
/// just one whose total topology could. Empty until the first
|
||||
/// /health poll.
|
||||
pub device_health: Vec<crate::discovery::DeviceHealth>,
|
||||
/// The reasoning-effort ladder this neuron honours (#223), copied
|
||||
/// from its `/models` reply at poll time. Per-node rather than
|
||||
/// per-model because it is host configuration — every model on a
|
||||
/// neuron reports the same rungs — and duplicating it onto each
|
||||
/// model entry would invite the two to disagree.
|
||||
pub reasoning_budget: Vec<crate::harness::ReasoningBudgetRung>,
|
||||
}
|
||||
|
||||
/// A model registered on a node, with its runtime status.
|
||||
@@ -187,12 +193,26 @@ pub struct CortexModelEntry {
|
||||
/// Served max-seq-len in tokens — mirrors `limit.context`.
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub max_model_len: Option<usize>,
|
||||
/// The same window under the two names generic OpenAI clients look
|
||||
/// for (#278). pi-ai's provider discovery, for one, reads
|
||||
/// `context_window` then `context_length` and nothing else — so a
|
||||
/// client probing cortex found no window at all and fell back to a
|
||||
/// number its operator had to invent by hand.
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub context_window: Option<usize>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub context_length: Option<usize>,
|
||||
/// Usable input budget — mirrors `limit.input` when present.
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub max_input_tokens: Option<usize>,
|
||||
/// Maximum generation tokens — mirrors `limit.output`.
|
||||
/// The largest output a request may name — mirrors
|
||||
/// `limit.output_ceiling`, falling back to `limit.output` (#278).
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub max_output_tokens: Option<usize>,
|
||||
/// What each effort level buys, in reasoning tokens (#223) — so a
|
||||
/// client picking `low` knows what it asked for.
|
||||
#[serde(default, skip_serializing_if = "Vec::is_empty")]
|
||||
pub reasoning_budget: Vec<crate::harness::ReasoningBudgetRung>,
|
||||
}
|
||||
|
||||
impl CortexModelEntry {
|
||||
@@ -204,8 +224,22 @@ impl CortexModelEntry {
|
||||
/// dropped the limit.
|
||||
pub fn sync_flat_limit(&mut self) {
|
||||
self.max_model_len = self.limit.as_ref().map(|l| l.context);
|
||||
self.context_window = self.limit.as_ref().map(|l| l.context);
|
||||
self.context_length = self.limit.as_ref().map(|l| l.context);
|
||||
self.max_input_tokens = self.limit.as_ref().and_then(|l| l.input);
|
||||
self.max_output_tokens = self.limit.as_ref().map(|l| l.output);
|
||||
// The ceiling, not the reserve (#278): this field is what
|
||||
// OpenAI-ecosystem clients read to decide what to ask for, and
|
||||
// pi-ai/DSH turns it straight into a per-request cap. Advertising
|
||||
// the reserve here handed a reasoning model less budget than its
|
||||
// own think block needs. Falls back to the reserve for a neuron
|
||||
// too old to publish a ceiling.
|
||||
self.max_output_tokens = self.limit.as_ref().map(|l| {
|
||||
if l.output_ceiling > 0 {
|
||||
l.output_ceiling
|
||||
} else {
|
||||
l.output
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -145,12 +145,22 @@ pub enum ResponsesInputItem {
|
||||
/// string or an array of content parts; the translator renders
|
||||
/// either form to text rather than losing the tool result.
|
||||
FunctionCallOutput { call_id: String, output: Value },
|
||||
/// Reasoning items emitted by o-series models. Accepted but
|
||||
/// not forwarded to the model — neuron's candle path doesn't
|
||||
/// surface reasoning separately yet.
|
||||
/// A reasoning item the model emitted on an earlier turn, replayed
|
||||
/// by the client so the model can continue its own train of thought.
|
||||
///
|
||||
/// Both spellings are modelled because both are in the wild and one
|
||||
/// of them is ours: OpenAI's o-series carries the text in `content`
|
||||
/// as `reasoning_text` parts, while neuron emits `summary` with a
|
||||
/// `summary_text` part. Clients round-trip the completed item
|
||||
/// verbatim — pi-ai stores our `response.output_item.done` payload
|
||||
/// and replays it unchanged — so whichever field we emit is the
|
||||
/// field we get back, and reading only one of them silently loses
|
||||
/// the turn's reasoning.
|
||||
Reasoning {
|
||||
#[serde(default)]
|
||||
content: Vec<Value>,
|
||||
#[serde(default)]
|
||||
summary: Vec<Value>,
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
@@ -94,6 +94,28 @@ pub fn anthropic_to_openai(req: MessagesRequest) -> ChatCompletionRequest {
|
||||
if let Some(tc) = tool_choice {
|
||||
obj.insert("tool_choice".into(), tc);
|
||||
}
|
||||
// Anthropic's extended-thinking budget is the one quantitative
|
||||
// reasoning control any of our wire formats offers (#223).
|
||||
// Normalised onto `reasoning.max_tokens`, the shape the harness
|
||||
// resolves, so all three inlets converge on one field instead of
|
||||
// each generation loop learning three spellings.
|
||||
let budget = obj
|
||||
.get("thinking")
|
||||
.filter(|t| t.get("type").and_then(Value::as_str) != Some("disabled"))
|
||||
.and_then(|t| t.get("budget_tokens"))
|
||||
.and_then(Value::as_u64)
|
||||
.filter(|n| *n > 0);
|
||||
if let Some(budget) = budget {
|
||||
let reasoning = obj
|
||||
.entry("reasoning")
|
||||
.or_insert_with(|| Value::Object(Default::default()));
|
||||
if !reasoning.is_object() {
|
||||
*reasoning = Value::Object(Default::default());
|
||||
}
|
||||
if let Some(r) = reasoning.as_object_mut() {
|
||||
r.entry("max_tokens").or_insert_with(|| Value::from(budget));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
ChatCompletionRequest {
|
||||
|
||||
@@ -798,6 +798,10 @@ async fn list_models(State(fleet): State<Arc<CortexState>>) -> Json<Value> {
|
||||
max_model_len: None,
|
||||
max_input_tokens: None,
|
||||
max_output_tokens: None,
|
||||
context_window: None,
|
||||
context_length: None,
|
||||
// Catalogue-seeded: no neuron has reported rungs yet.
|
||||
reasoning_budget: Vec::new(),
|
||||
},
|
||||
);
|
||||
}
|
||||
@@ -839,6 +843,13 @@ async fn list_models(State(fleet): State<Arc<CortexState>>) -> Json<Value> {
|
||||
// so a client never overflows the most-constrained
|
||||
// deployment that might serve it.
|
||||
e.limit = tightest_limit(e.limit.take(), entry.limit.clone());
|
||||
// The reasoning ladder this neuron honours (#223).
|
||||
// First non-empty wins: a catalogue-seeded entry
|
||||
// starts with none, and neurons serving the same
|
||||
// model are configured alike.
|
||||
if e.reasoning_budget.is_empty() {
|
||||
e.reasoning_budget = node.reasoning_budget.clone();
|
||||
}
|
||||
})
|
||||
.or_insert_with(|| CortexModelEntry {
|
||||
id: model_id.clone(),
|
||||
@@ -858,6 +869,10 @@ async fn list_models(State(fleet): State<Arc<CortexState>>) -> Json<Value> {
|
||||
max_model_len: None,
|
||||
max_input_tokens: None,
|
||||
max_output_tokens: None,
|
||||
context_window: None,
|
||||
context_length: None,
|
||||
// What this neuron says each effort level buys (#223).
|
||||
reasoning_budget: node.reasoning_budget.clone(),
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -919,6 +934,10 @@ async fn list_models(State(fleet): State<Arc<CortexState>>) -> Json<Value> {
|
||||
max_model_len: None,
|
||||
max_input_tokens: None,
|
||||
max_output_tokens: None,
|
||||
context_window: None,
|
||||
context_length: None,
|
||||
// What this neuron says each effort level buys (#223).
|
||||
reasoning_budget: node.reasoning_budget.clone(),
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -956,6 +975,10 @@ async fn list_models(State(fleet): State<Arc<CortexState>>) -> Json<Value> {
|
||||
max_model_len: None,
|
||||
max_input_tokens: None,
|
||||
max_output_tokens: None,
|
||||
context_window: None,
|
||||
context_length: None,
|
||||
// Catalogue-seeded: no neuron has reported rungs yet.
|
||||
reasoning_budget: Vec::new(),
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
@@ -141,6 +141,15 @@ pub fn usage_sink(
|
||||
/// returns a guard the caller settles with actual usage; on refusal returns
|
||||
/// the #63 envelope (`rate_limit_exceeded` + `Retry-After` for a resetting
|
||||
/// window, `insufficient_quota` for a hard balance — never `402`).
|
||||
// The `Err` variant is the #63 envelope itself, which is a data-carrying
|
||||
// struct (status, type, code, message, param, retry hint, diagnostics) and
|
||||
// is passed by value everywhere it is built, matched and rendered. Boxing it
|
||||
// at this one boundary would push `Box` conversions through three handlers
|
||||
// and both HTTP adapters to save an allocation on a path taken once per
|
||||
// rejected request. clippy 1.98 lowered what it considers large enough to
|
||||
// flag; the shape is deliberate. Same call as the existing allows on the
|
||||
// figment-returning config loaders.
|
||||
#[allow(clippy::result_large_err)]
|
||||
pub async fn reserve_or_reject(
|
||||
provider: Arc<dyn EntitlementProvider>,
|
||||
principal: &Principal,
|
||||
@@ -279,7 +288,7 @@ mod tests {
|
||||
// Requested output dominates; prompt estimate is small for a tiny body.
|
||||
let body = br#"{"model":"m","max_tokens":1000}"#;
|
||||
let est = reservation_estimate(body, Some(8192));
|
||||
assert!(est >= 1000 && est < 1100, "est was {est}");
|
||||
assert!((1000..1100).contains(&est), "est was {est}");
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
||||
@@ -130,6 +130,17 @@ async fn poll_neuron(fleet: &CortexState, name: &str, endpoint: &str) {
|
||||
Ok(resp) if resp.status().is_success() => {
|
||||
match resp.json::<Vec<ModelInfo>>().await {
|
||||
Ok(models) => {
|
||||
// The reasoning-effort ladder is host configuration:
|
||||
// every model on a neuron reports the same rungs, so
|
||||
// it is held once per node rather than copied onto
|
||||
// each model entry where the two could drift (#223).
|
||||
// Taken from the first model that offers one — a
|
||||
// non-reasoning model advertises none.
|
||||
node.reasoning_budget = models
|
||||
.iter()
|
||||
.map(|m| m.reasoning_budget.clone())
|
||||
.find(|rungs| !rungs.is_empty())
|
||||
.unwrap_or_default();
|
||||
let mut seen = std::collections::HashSet::new();
|
||||
for upstream in &models {
|
||||
seen.insert(upstream.id.clone());
|
||||
|
||||
@@ -66,6 +66,9 @@ impl CortexState {
|
||||
endpoint: nc.endpoint.clone(),
|
||||
healthy: false,
|
||||
models: HashMap::new(),
|
||||
// Filled from the neuron's /models reply at the
|
||||
// first poll (#223).
|
||||
reasoning_budget: Vec::new(),
|
||||
lifecycle_cycles: 0,
|
||||
last_poll: None,
|
||||
discovery: None,
|
||||
|
||||
@@ -1,9 +1,9 @@
|
||||
//! Integration tests for budget enforcement (#52) — the A0 seatbelt.
|
||||
//!
|
||||
//! A reservation over the key's hard cap is refused *before* neuron is hit,
|
||||
//! with the #63 code matching the cap-window semantics (rate_limit_exceeded
|
||||
//! + Retry-After for a resetting window, insufficient_quota for a hard
|
||||
//! balance). Spend never exceeds the cap. No 402, ever.
|
||||
//! with the #63 code matching the cap-window semantics: `rate_limit_exceeded`
|
||||
//! plus `Retry-After` for a resetting window, `insufficient_quota` for a hard
|
||||
//! balance. Spend never exceeds the cap. No 402, ever.
|
||||
|
||||
use axum::Json;
|
||||
use axum::extract::Path;
|
||||
|
||||
@@ -69,6 +69,18 @@ capabilities = ["text"]
|
||||
let mut nodes = fleet.nodes.write().await;
|
||||
let node = nodes.get_mut("mock-node").expect("node exists");
|
||||
node.healthy = true;
|
||||
// The ladder the neuron reported at poll time (#223) — host
|
||||
// configuration, held once per node.
|
||||
node.reasoning_budget = ["minimal", "low", "medium", "high"]
|
||||
.into_iter()
|
||||
.zip([1024usize, 4096, 12288, 32768])
|
||||
.map(
|
||||
|(effort, tokens)| cortex_core::harness::ReasoningBudgetRung {
|
||||
effort: effort.into(),
|
||||
tokens,
|
||||
},
|
||||
)
|
||||
.collect();
|
||||
node.models.insert(
|
||||
"test-model".into(),
|
||||
ModelEntry {
|
||||
@@ -85,6 +97,9 @@ capabilities = ["text"]
|
||||
context: 49152,
|
||||
input: Some(40960),
|
||||
output: 8192,
|
||||
// The ceiling a request may name, distinct from the
|
||||
// 8192-token reserve above (#278).
|
||||
output_ceiling: 32768,
|
||||
}),
|
||||
servable: None,
|
||||
},
|
||||
@@ -147,7 +162,34 @@ capabilities = ["text"]
|
||||
// vLLM-convention probes (Hermes Agent) auto-detect the window.
|
||||
assert_eq!(entry["max_model_len"], 49152);
|
||||
assert_eq!(entry["max_input_tokens"], 40960);
|
||||
assert_eq!(entry["max_output_tokens"], 8192);
|
||||
// The ceiling, not the reserve (#278). pi-ai's discovery — and every
|
||||
// client like it — turns this field into the cap it sends on each
|
||||
// request; advertising the 8192 reserve told a reasoning model's
|
||||
// harness to stop below the cost of a single think block.
|
||||
assert_eq!(entry["max_output_tokens"], 32768);
|
||||
// What each effort level buys (#223). A client can only send the
|
||||
// rung names — it has no way to express a token count — so a ladder
|
||||
// it cannot see is a ladder it has to guess at.
|
||||
let rungs = entry["reasoning_budget"]
|
||||
.as_array()
|
||||
.expect("reasoning_budget advertised for a reasoning model");
|
||||
let names: Vec<&str> = rungs
|
||||
.iter()
|
||||
.map(|r| r["effort"].as_str().expect("effort"))
|
||||
.collect();
|
||||
assert_eq!(names, ["minimal", "low", "medium", "high"]);
|
||||
assert!(
|
||||
rungs
|
||||
.iter()
|
||||
.all(|r| r["tokens"].as_u64().is_some_and(|t| t > 0)),
|
||||
"every advertised rung must name a token count"
|
||||
);
|
||||
|
||||
// The window under the two names generic clients actually read.
|
||||
// pi-ai looks for `context_window` then `context_length` and gives
|
||||
// up if neither is present — `max_model_len` is invisible to it.
|
||||
assert_eq!(entry["context_window"], 49152);
|
||||
assert_eq!(entry["context_length"], 49152);
|
||||
|
||||
// No limit → flat fields omitted entirely, never 0 or a guess.
|
||||
let unknown = body["data"]
|
||||
@@ -157,7 +199,13 @@ capabilities = ["text"]
|
||||
.find(|m| m["id"] == "no-limit-model")
|
||||
.expect("no-limit-model present in /v1/models");
|
||||
assert!(unknown.get("limit").is_none());
|
||||
for key in ["max_model_len", "max_input_tokens", "max_output_tokens"] {
|
||||
for key in [
|
||||
"max_model_len",
|
||||
"max_input_tokens",
|
||||
"max_output_tokens",
|
||||
"context_window",
|
||||
"context_length",
|
||||
] {
|
||||
assert!(
|
||||
unknown.get(key).is_none(),
|
||||
"{key} must be omitted when the window is unknown"
|
||||
|
||||
@@ -96,6 +96,7 @@ async fn spawn_gateway(neuron: &str, context: usize) -> String {
|
||||
context,
|
||||
input: None,
|
||||
output: 16,
|
||||
output_ceiling: 0,
|
||||
}),
|
||||
servable: None,
|
||||
},
|
||||
|
||||
@@ -236,6 +236,7 @@ impl TargetClient {
|
||||
// A generic OpenAI endpoint says nothing about
|
||||
// serviceability, and absent means "no opinion".
|
||||
servable: None,
|
||||
reasoning_budget: Vec::new(),
|
||||
})
|
||||
.collect())
|
||||
}
|
||||
|
||||
@@ -80,10 +80,19 @@ fn router_entry(cortex: &str, e: &CortexModelEntry) -> CortexModelEntry {
|
||||
cost: e.cost.clone(),
|
||||
tool_call: e.tool_call,
|
||||
reasoning: e.reasoning,
|
||||
// Derived from `limit` by the final sync pass in aggregate_models.
|
||||
max_model_len: None,
|
||||
max_input_tokens: None,
|
||||
max_output_tokens: None,
|
||||
// Every flat ecosystem field is derived from `limit` by the
|
||||
// final sync pass in aggregate_models, so they are cleared here
|
||||
// rather than copied. Spread from `e` keeps a field added to the
|
||||
// entry later from being silently dropped at this hop.
|
||||
..CortexModelEntry {
|
||||
max_model_len: None,
|
||||
max_input_tokens: None,
|
||||
max_output_tokens: None,
|
||||
context_window: None,
|
||||
context_length: None,
|
||||
reasoning_budget: Vec::new(),
|
||||
..e.clone()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -164,6 +173,9 @@ mod tests {
|
||||
max_model_len: None,
|
||||
max_input_tokens: None,
|
||||
max_output_tokens: None,
|
||||
context_window: None,
|
||||
context_length: None,
|
||||
reasoning_budget: Vec::new(),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -224,6 +236,7 @@ mod tests {
|
||||
context: 32_768,
|
||||
input: None,
|
||||
output: 4096,
|
||||
output_ceiling: 0,
|
||||
});
|
||||
a.cost = Some(ModelCost {
|
||||
input: 0.50,
|
||||
@@ -236,6 +249,7 @@ mod tests {
|
||||
context: 16_384, // tighter
|
||||
input: None,
|
||||
output: 4096,
|
||||
output_ceiling: 0,
|
||||
});
|
||||
b.cost = Some(ModelCost {
|
||||
input: 0.20, // cheaper
|
||||
|
||||
@@ -42,6 +42,9 @@ fn model_entry(loaded: bool, feasible: bool) -> CortexModelEntry {
|
||||
max_model_len: None,
|
||||
max_input_tokens: None,
|
||||
max_output_tokens: None,
|
||||
context_window: None,
|
||||
context_length: None,
|
||||
reasoning_budget: Vec::new(),
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -58,6 +58,11 @@ mod tests {
|
||||
assert!(cfg.max_redirects > 0);
|
||||
}
|
||||
|
||||
// The closure's error type is `figment::Error`, which is 208 bytes and
|
||||
// not ours to shrink; `Jail::expect_with` dictates the signature. The
|
||||
// usual remedy (box the Err) is not available through a third-party
|
||||
// callback contract, and this is a test closure run once.
|
||||
#[allow(clippy::result_large_err)]
|
||||
#[test]
|
||||
fn loads_toml_with_env_override() {
|
||||
figment::Jail::expect_with(|jail| {
|
||||
|
||||
@@ -86,6 +86,16 @@ pub struct CandleHarnessConfig {
|
||||
#[serde(default)]
|
||||
pub context_limit: ContextLimitConfig,
|
||||
|
||||
/// What each OpenAI effort level costs, in reasoning tokens (#223).
|
||||
///
|
||||
/// OpenAI-shaped clients cannot send a token count — their whole
|
||||
/// vocabulary is the `minimal|low|medium|high` ladder — so the
|
||||
/// numbers live here and are advertised on `/models`. Operator
|
||||
/// configurable because the right values depend on the model and the
|
||||
/// hardware.
|
||||
#[serde(default)]
|
||||
pub reasoning_budget: crate::harness::reasoning_budget::ReasoningBudgetLadder,
|
||||
|
||||
/// Admission control (#53): bounds the per-model wait queue so a busy
|
||||
/// model returns a fast, retryable `429`/`503` instead of stalling new
|
||||
/// requests until their client times out.
|
||||
@@ -333,8 +343,24 @@ pub struct ContextLimitConfig {
|
||||
/// Generation reserve (tokens) left below the context wall:
|
||||
/// `input = context − output_reserve_tokens`. Defaults to neuron's
|
||||
/// default `max_tokens`.
|
||||
///
|
||||
/// This is the budget a request gets when it names none, and the
|
||||
/// figure the input/output split is derived from. It is **not** the
|
||||
/// most a request may ask for — see `max_output_tokens` (#278).
|
||||
#[serde(default = "default_output_reserve_tokens")]
|
||||
pub output_reserve_tokens: usize,
|
||||
|
||||
/// The largest output a request may name, advertised as
|
||||
/// `max_output_tokens` on `/v1/models` (#278).
|
||||
///
|
||||
/// `None` derives it as `min(max_position_embeddings,
|
||||
/// DEFAULT_OUTPUT_CEILING)`. Separate from the reserve because the
|
||||
/// two answer different questions and were previously the same
|
||||
/// number: publishing the reserve told every client that read the
|
||||
/// advertisement to cap itself at the default, which on a reasoning
|
||||
/// model can be less than one think block.
|
||||
#[serde(default)]
|
||||
pub max_output_tokens: Option<usize>,
|
||||
}
|
||||
|
||||
impl Default for ContextLimitConfig {
|
||||
@@ -346,6 +372,7 @@ impl Default for ContextLimitConfig {
|
||||
activation_headroom_mb: default_activation_headroom_mb(),
|
||||
min_free_floor_mb: default_context_min_free_floor_mb(),
|
||||
output_reserve_tokens: default_output_reserve_tokens(),
|
||||
max_output_tokens: None,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1051,7 +1051,7 @@ mod tests {
|
||||
|
||||
// Fill the single queue slot with a waiter that parks on the semaphore.
|
||||
let ctrl2 = Arc::clone(&ctrl);
|
||||
let waiter = tokio::spawn(async move { ctrl2.enter(None).await.map(|p| drop(p)) });
|
||||
let waiter = tokio::spawn(async move { ctrl2.enter(None).await.map(drop) });
|
||||
// Give the waiter a moment to occupy the queue slot.
|
||||
tokio::time::sleep(Duration::from_millis(50)).await;
|
||||
assert_eq!(ctrl.queue_depth(), 1);
|
||||
|
||||
@@ -562,9 +562,9 @@ mod tests {
|
||||
.forward_batch_decode(&input, &positions, mask.as_ref())
|
||||
.expect("batched step");
|
||||
assert_eq!(h.dims()[0], 3);
|
||||
for row in 0..3 {
|
||||
for (row, exp) in expected.iter().enumerate() {
|
||||
let got: Vec<f32> = h.i((row, 0, ..)).unwrap().to_vec1().unwrap();
|
||||
let diff = max_abs_diff(&expected[row][t], &got);
|
||||
let diff = max_abs_diff(&exp[t], &got);
|
||||
assert!(diff < 1e-4, "row {row} step {t} diverged: {diff}");
|
||||
}
|
||||
}
|
||||
@@ -610,9 +610,9 @@ mod tests {
|
||||
let h = model
|
||||
.forward_batch_decode(&input, &[3, 3], None)
|
||||
.expect("step");
|
||||
for row in 0..2 {
|
||||
for (row, exp) in expected.iter().enumerate() {
|
||||
let got: Vec<f32> = h.i((row, 0, ..)).unwrap().to_vec1().unwrap();
|
||||
let diff = max_abs_diff(&expected[row], &got);
|
||||
let diff = max_abs_diff(exp, &got);
|
||||
assert!(diff < 1e-4, "row {row} diverged: {diff}");
|
||||
}
|
||||
}
|
||||
@@ -674,9 +674,9 @@ mod tests {
|
||||
let h = model
|
||||
.forward_batch_decode(&input, &positions, mask.as_ref())
|
||||
.expect("batched step");
|
||||
for row in 0..2 {
|
||||
for (row, exp) in expected.iter().enumerate() {
|
||||
let got: Vec<f32> = h.i((row, 0, ..)).unwrap().to_vec1().unwrap();
|
||||
let diff = max_abs_diff(&expected[row][t], &got);
|
||||
let diff = max_abs_diff(&exp[t], &got);
|
||||
assert!(diff < 1e-4, "batched row {row} step {t}: {diff}");
|
||||
}
|
||||
}
|
||||
|
||||
@@ -98,6 +98,9 @@ pub struct CandleHarness {
|
||||
/// Context-limit derivation settings (#67), read in `list_models`
|
||||
/// to compute each model's advertised `limit{context,input,output}`.
|
||||
context_limit_cfg: crate::config::ContextLimitConfig,
|
||||
/// Effort-level → reasoning-token rungs (#223), resolved per request
|
||||
/// so a caller can bound how long the model thinks.
|
||||
reasoning_budget_cfg: super::reasoning_budget::ReasoningBudgetLadder,
|
||||
/// Admission-control settings (#53), used to build each loaded model's
|
||||
/// [`super::admission::AdmissionController`] at load time.
|
||||
admission_cfg: crate::config::AdmissionConfig,
|
||||
@@ -370,7 +373,7 @@ fn servability_from(
|
||||
});
|
||||
}
|
||||
{
|
||||
match check_vram(0, free_mb, profile, cfg) {
|
||||
match check_vram(0, None, free_mb, profile, cfg) {
|
||||
Ok(()) => Some(ModelServability {
|
||||
ok: true,
|
||||
reason: None,
|
||||
@@ -574,6 +577,33 @@ impl LoadedImageModel {
|
||||
/// model's `generation_config.json` and then to the built-in fallback.
|
||||
/// Kept in one place so a new knob is added once rather than at each of
|
||||
/// the four request-handling paths.
|
||||
/// The reasoning-token budget this request asked for, if any (#223).
|
||||
///
|
||||
/// Three inlets converge on `extra.reasoning`: Anthropic's
|
||||
/// `thinking.budget_tokens` (normalised in translation), the explicit
|
||||
/// `reasoning.max_tokens` extension, and OpenAI's `reasoning.effort`
|
||||
/// ladder resolved against the operator's configured rungs. An explicit
|
||||
/// token count wins over a ladder rung — it is the more specific thing
|
||||
/// the caller reached for.
|
||||
///
|
||||
/// `None` means unbounded, which is the default and exactly how the
|
||||
/// model behaved before this existed.
|
||||
fn requested_reasoning_budget(
|
||||
request: &ChatCompletionRequest,
|
||||
ladder: &super::reasoning_budget::ReasoningBudgetLadder,
|
||||
) -> Option<usize> {
|
||||
let reasoning = request.extra.get("reasoning")?;
|
||||
if let Some(n) = reasoning
|
||||
.get("max_tokens")
|
||||
.and_then(|v| v.as_u64())
|
||||
.filter(|n| *n > 0)
|
||||
{
|
||||
return Some(n as usize);
|
||||
}
|
||||
let effort = reasoning.get("effort").and_then(|v| v.as_str())?;
|
||||
ladder.for_effort(effort)
|
||||
}
|
||||
|
||||
fn requested_sampling(request: &ChatCompletionRequest) -> super::sampling::RequestedSampling {
|
||||
super::sampling::RequestedSampling {
|
||||
temperature: request.temperature,
|
||||
@@ -1231,17 +1261,6 @@ impl LlamaDense {
|
||||
}
|
||||
}
|
||||
|
||||
/// Repetition penalty applied to recently-generated tokens before
|
||||
/// sampling. 1.0 disables it; >1.0 makes recently-emitted tokens less
|
||||
/// likely. mistral.rs and llama.cpp default to 1.1, which is enough to
|
||||
/// stop small quantized models from degenerating into "Wait, no, no..."
|
||||
/// loops without distorting normal output.
|
||||
const REPEAT_PENALTY: f32 = 1.1;
|
||||
|
||||
/// Number of recently-generated tokens to feed into the repetition
|
||||
/// penalty. Matches the candle quantized-qwen3 example default.
|
||||
const REPEAT_LAST_N: usize = 64;
|
||||
|
||||
/// Architectures the dense safetensors path can construct. Keep
|
||||
/// alphabetical; one entry per supported `config.json#/model_type`
|
||||
/// value. New entries land alongside a new `ModelArch` variant + a
|
||||
@@ -1802,6 +1821,7 @@ fn check_vram_floor(vram_free_mb: u64) -> Result<(), InferenceError> {
|
||||
/// request, matching the fallback `derive_limit` takes.
|
||||
pub(crate) fn kv_reservation_mb(
|
||||
prompt_len: usize,
|
||||
requested_output: Option<usize>,
|
||||
profile: Option<&super::context_limit::ContextProfile>,
|
||||
cfg: &crate::config::ContextLimitConfig,
|
||||
) -> u64 {
|
||||
@@ -1809,7 +1829,13 @@ pub(crate) fn kv_reservation_mb(
|
||||
if profile.kv_bytes_per_token_per_card == 0 {
|
||||
return 0;
|
||||
}
|
||||
let tokens = (prompt_len as u64).saturating_add(cfg.output_reserve_tokens as u64);
|
||||
// Price the budget the request actually declared (#278). This used
|
||||
// to assume the reserve for every request, so a caller asking for
|
||||
// 24k tokens was admitted against an 8k reservation and the KV it
|
||||
// went on to occupy was never accounted for — the gate in #257 can
|
||||
// only be as honest as the number it is given.
|
||||
let output = requested_output.unwrap_or(cfg.output_reserve_tokens);
|
||||
let tokens = (prompt_len as u64).saturating_add(output as u64);
|
||||
profile
|
||||
.kv_bytes_per_token_per_card
|
||||
.saturating_mul(tokens)
|
||||
@@ -1858,6 +1884,7 @@ fn prefix_cache_reserved_mb(cfg: &crate::config::PrefixCacheConfig) -> u64 {
|
||||
/// copies of "can this serve" will always drift; one cannot.
|
||||
fn check_vram(
|
||||
prompt_len: usize,
|
||||
requested_output: Option<usize>,
|
||||
vram_free_mb: u64,
|
||||
profile: Option<&super::context_limit::ContextProfile>,
|
||||
cfg: &crate::config::ContextLimitConfig,
|
||||
@@ -1880,7 +1907,8 @@ fn check_vram(
|
||||
if let Some(profile) = profile
|
||||
&& profile.kv_bytes_per_token_per_card > 0
|
||||
{
|
||||
let tokens = (prompt_len as u64).saturating_add(cfg.output_reserve_tokens as u64);
|
||||
let tokens = (prompt_len as u64)
|
||||
.saturating_add(requested_output.unwrap_or(cfg.output_reserve_tokens) as u64);
|
||||
let kv_mb = profile.kv_bytes_per_token_per_card.saturating_mul(tokens) / (1024 * 1024);
|
||||
let required_mb = kv_mb
|
||||
.saturating_add(cfg.activation_headroom_mb)
|
||||
@@ -1990,18 +2018,39 @@ async fn acquire_pool_lock<'a>(
|
||||
/// Apply the repetition penalty (if any) to the prediction logits and
|
||||
/// then sample. Centralises the prefill / generation-loop call sites
|
||||
/// so they share identical sampling behaviour.
|
||||
///
|
||||
/// The penalty and its window come from the resolved [`SamplingParams`]
|
||||
/// rather than from constants: a caller that sets `repetition_penalty`
|
||||
/// or `repeat_last_n` must reach the sampler, or the request is
|
||||
/// accepted and discarded — the defect #271 exists to close.
|
||||
pub(crate) fn sample_with_penalty(
|
||||
logits: &Tensor,
|
||||
history: &[u32],
|
||||
logits_processor: &mut LogitsProcessor,
|
||||
sampling: &super::sampling::SamplingParams,
|
||||
governor: Option<&mut super::reasoning_budget::ReasoningGovernor>,
|
||||
) -> Result<u32> {
|
||||
let penalised = if (REPEAT_PENALTY - 1.0).abs() < f32::EPSILON || history.is_empty() {
|
||||
let penalised = if (sampling.repeat_penalty - 1.0).abs() < f32::EPSILON
|
||||
|| sampling.repeat_last_n == 0
|
||||
|| history.is_empty()
|
||||
{
|
||||
logits.clone()
|
||||
} else {
|
||||
let start = history.len().saturating_sub(REPEAT_LAST_N);
|
||||
candle_transformers::utils::apply_repeat_penalty(logits, REPEAT_PENALTY, &history[start..])?
|
||||
let start = history.len().saturating_sub(sampling.repeat_last_n);
|
||||
candle_transformers::utils::apply_repeat_penalty(
|
||||
logits,
|
||||
sampling.repeat_penalty,
|
||||
&history[start..],
|
||||
)?
|
||||
};
|
||||
Ok(logits_processor.sample(&penalised)?)
|
||||
let sampled = logits_processor.sample(&penalised)?;
|
||||
// The reasoning budget governs which token is *emitted*, not which
|
||||
// is sampled: over budget inside a think block, the model gets its
|
||||
// own close marker so it transitions to answering (#223).
|
||||
Ok(match governor {
|
||||
Some(g) => g.govern(sampled),
|
||||
None => sampled,
|
||||
})
|
||||
}
|
||||
|
||||
/// Chunked prefill against an in-process [`ModelArch`]. Splits
|
||||
@@ -2220,6 +2269,7 @@ impl CandleHarness {
|
||||
recovery_tx,
|
||||
prefix_cache_cfg: config.prefix_cache.clone(),
|
||||
context_limit_cfg: config.context_limit.clone(),
|
||||
reasoning_budget_cfg: config.reasoning_budget.clone(),
|
||||
admission_cfg: config.admission.clone(),
|
||||
image_cfg: config.image.clone(),
|
||||
});
|
||||
@@ -2354,6 +2404,55 @@ impl CandleHarness {
|
||||
})
|
||||
}
|
||||
|
||||
/// The files that describe **how to drive** a model, as opposed to
|
||||
/// its weights: the sampling its authors published, and the prompt
|
||||
/// format they wrote.
|
||||
///
|
||||
/// Every one is optional in a HuggingFace repo, and each is read
|
||||
/// from the snapshot directory beside `tokenizer.json` — so if the
|
||||
/// fetch does not pull it, the reader silently finds nothing and
|
||||
/// falls back to a built-in guess. That is the whole bug class this
|
||||
/// list exists to close (#272, #280): the loader used to fetch
|
||||
/// `config.json`, `tokenizer.json` and the weights, and nothing
|
||||
/// else, so a model neuron downloaded itself was served with our
|
||||
/// sampling defaults and our prompt format rather than its own.
|
||||
pub(crate) const SIDECAR_FILES: [&'static str; 4] = [
|
||||
// Sampling the model's authors published — read by
|
||||
// `ModelGenerationDefaults::load_from_dir` (#272).
|
||||
"generation_config.json",
|
||||
// The prompt format, in the three places HF puts it. Checked in
|
||||
// the same priority order `load_chat_template_alongside` reads
|
||||
// them, though all present ones are fetched (#280).
|
||||
"chat_template.jinja",
|
||||
"chat_template.json",
|
||||
"tokenizer_config.json",
|
||||
];
|
||||
|
||||
/// Pull the sidecars into the snapshot directory beside
|
||||
/// `tokenizer.json`, where their readers look at load time.
|
||||
///
|
||||
/// Best-effort by design: plenty of repos ship none of them, and a
|
||||
/// miss is a debug line rather than a load failure. But the fetch
|
||||
/// has to happen, or the file is never on disk to be read.
|
||||
async fn fetch_model_sidecars(repo: &hf_hub::api::tokio::ApiRepo, display_id: &str) {
|
||||
for name in Self::SIDECAR_FILES {
|
||||
match repo.get(name).await {
|
||||
Ok(path) => tracing::debug!(
|
||||
model = %display_id,
|
||||
file = %name,
|
||||
path = ?path,
|
||||
"fetched model sidecar"
|
||||
),
|
||||
Err(e) => tracing::debug!(
|
||||
model = %display_id,
|
||||
file = %name,
|
||||
error = %e,
|
||||
"model sidecar absent from repo; a built-in default applies"
|
||||
),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Resolve a dense (bf16/fp16 safetensors) model to its local file
|
||||
/// paths.
|
||||
///
|
||||
@@ -2379,6 +2478,10 @@ impl CandleHarness {
|
||||
.get("tokenizer.json")
|
||||
.await
|
||||
.with_context(|| format!("fetch tokenizer.json from {display_id}"))?;
|
||||
// Beside the tokenizer, so the sampling defaults and the
|
||||
// model's own prompt format are readable from the snapshot dir
|
||||
// at load (#272, #280).
|
||||
Self::fetch_model_sidecars(&repo, &display_id).await;
|
||||
|
||||
// Prefer the sharded layout (most HF dense models > 5B ship it).
|
||||
let safetensors_paths = match repo.get("model.safetensors.index.json").await {
|
||||
@@ -2681,6 +2784,9 @@ impl CandleHarness {
|
||||
.get("tokenizer.json")
|
||||
.await
|
||||
.with_context(|| format!("fetch tokenizer.json from {tokenizer_repo_path}"))?;
|
||||
// Same repo as the tokenizer — a GGUF-only repo carries none of
|
||||
// them, and the base repo carries them all (#272, #280).
|
||||
Self::fetch_model_sidecars(&tokenizer_repo, &tokenizer_repo_path).await;
|
||||
Ok((gguf_path, tokenizer_path))
|
||||
}
|
||||
|
||||
@@ -2846,7 +2952,24 @@ impl CandleHarness {
|
||||
&loaded.generation_defaults,
|
||||
unix_subsec_nanos(),
|
||||
);
|
||||
let max_new = request.max_tokens.unwrap_or(8192) as usize;
|
||||
// Bound the think block if the caller asked for it (#223),
|
||||
// so reasoning cannot consume the whole output budget.
|
||||
let governor = super::reasoning_budget::ReasoningGovernor::new(
|
||||
requested_reasoning_budget(&request, &self.reasoning_budget_cfg),
|
||||
loaded
|
||||
.reasoning_tokens
|
||||
.as_ref()
|
||||
.map(|p| (p.open_id, p.close_id)),
|
||||
prompt_opened_reasoning,
|
||||
);
|
||||
// The output budget the request declared, under either
|
||||
// spelling (#254), and the fallback when it declares none.
|
||||
// One value drives generation, KV pricing and admission — they
|
||||
// disagreed before, and the gate priced a budget nobody asked
|
||||
// for (#278).
|
||||
let requested_output = request.effective_max_tokens().map(|t| t as usize);
|
||||
let cfg_output_reserve = self.context_limit_cfg.output_reserve_tokens;
|
||||
let max_new = requested_output.unwrap_or(cfg_output_reserve);
|
||||
|
||||
let eos_id = loaded
|
||||
.tokenizer
|
||||
@@ -2905,6 +3028,7 @@ impl CandleHarness {
|
||||
*image_token_id,
|
||||
max_new,
|
||||
sampling.clone(),
|
||||
governor,
|
||||
eos_id,
|
||||
)
|
||||
.await
|
||||
@@ -2918,6 +3042,7 @@ impl CandleHarness {
|
||||
loaded.tokenizer.token_to_id("<|im_start|>"),
|
||||
max_new,
|
||||
sampling.clone(),
|
||||
governor,
|
||||
eos_id,
|
||||
)
|
||||
.await
|
||||
@@ -2968,6 +3093,7 @@ impl CandleHarness {
|
||||
im_start_id,
|
||||
max_new,
|
||||
sampling.clone(),
|
||||
governor,
|
||||
eos_id,
|
||||
)
|
||||
})
|
||||
@@ -3304,7 +3430,23 @@ impl CandleHarness {
|
||||
&loaded.generation_defaults,
|
||||
unix_subsec_nanos(),
|
||||
);
|
||||
let max_new = request.max_tokens.unwrap_or(8192) as usize;
|
||||
// Bound the think block if the caller asked for it (#223).
|
||||
let governor = super::reasoning_budget::ReasoningGovernor::new(
|
||||
requested_reasoning_budget(&request, &self.reasoning_budget_cfg),
|
||||
loaded
|
||||
.reasoning_tokens
|
||||
.as_ref()
|
||||
.map(|p| (p.open_id, p.close_id)),
|
||||
prompt_opens_reasoning(&prompt_tokens, loaded.reasoning_tokens.as_ref()),
|
||||
);
|
||||
// The output budget the request declared, under either
|
||||
// spelling (#254), and the fallback when it declares none.
|
||||
// One value drives generation, KV pricing and admission — they
|
||||
// disagreed before, and the gate priced a budget nobody asked
|
||||
// for (#278).
|
||||
let requested_output = request.effective_max_tokens().map(|t| t as usize);
|
||||
let cfg_output_reserve = self.context_limit_cfg.output_reserve_tokens;
|
||||
let max_new = requested_output.unwrap_or(cfg_output_reserve);
|
||||
|
||||
let eos_id = loaded
|
||||
.tokenizer
|
||||
@@ -3403,6 +3545,7 @@ impl CandleHarness {
|
||||
principal.as_deref(),
|
||||
kv_reservation_mb(
|
||||
prompt_len,
|
||||
requested_output,
|
||||
loaded.context_profile.as_ref(),
|
||||
&self.context_limit_cfg,
|
||||
),
|
||||
@@ -3423,6 +3566,12 @@ impl CandleHarness {
|
||||
prompt_tokens,
|
||||
max_new,
|
||||
sampling: sampling.clone(),
|
||||
// The engine builds its own per-slot governor from
|
||||
// this, since slots in a batch differ (#223).
|
||||
reasoning_budget: requested_reasoning_budget(
|
||||
&request,
|
||||
&self.reasoning_budget_cfg,
|
||||
),
|
||||
eos_id,
|
||||
tool_schemas,
|
||||
tx,
|
||||
@@ -3452,6 +3601,7 @@ impl CandleHarness {
|
||||
&loaded_for_task.prefill_rate,
|
||||
max_new,
|
||||
sampling.clone(),
|
||||
governor,
|
||||
eos_id,
|
||||
reasoning_tokens_inner,
|
||||
tool_call_tokens_inner,
|
||||
@@ -3514,6 +3664,7 @@ impl CandleHarness {
|
||||
loaded_for_task.prefix_cache.as_deref(),
|
||||
max_new,
|
||||
sampling.clone(),
|
||||
governor,
|
||||
eos_id,
|
||||
reasoning_tokens_inner.as_ref(),
|
||||
tool_call_tokens_inner.as_ref(),
|
||||
@@ -3787,6 +3938,24 @@ impl Harness for CandleHarness {
|
||||
tool_call: h.has_tool_call(),
|
||||
reasoning: h.has_reasoning(),
|
||||
servable: h.servability(&self.context_limit_cfg),
|
||||
// What each effort level buys (#223). Only meaningful
|
||||
// for a model that reasons — advertising rungs on a
|
||||
// model that never opens a think block would describe a
|
||||
// control that does nothing.
|
||||
reasoning_budget: if h.has_reasoning() {
|
||||
self.reasoning_budget_cfg
|
||||
.advertised()
|
||||
.into_iter()
|
||||
.map(
|
||||
|(effort, tokens)| cortex_core::harness::ReasoningBudgetRung {
|
||||
effort,
|
||||
tokens,
|
||||
},
|
||||
)
|
||||
.collect()
|
||||
} else {
|
||||
Vec::new()
|
||||
},
|
||||
});
|
||||
}
|
||||
// Models mid-recovery whose registry slot is absent (the
|
||||
@@ -3809,6 +3978,9 @@ impl Harness for CandleHarness {
|
||||
// Mid-recovery: the handle is gone, so there is
|
||||
// nothing to evaluate. No opinion, not "unservable".
|
||||
servable: None,
|
||||
// A recovering model is a trigger-time snapshot; it
|
||||
// is not serving, so it advertises no controls.
|
||||
reasoning_budget: Vec::new(),
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -4881,9 +5053,16 @@ impl CandleHarness {
|
||||
|
||||
let tp_for_marker = Arc::clone(&tp);
|
||||
let context_limit_cfg = self.context_limit_cfg.clone();
|
||||
let reasoning_budget_cfg = self.reasoning_budget_cfg.clone();
|
||||
let handle = tokio::spawn(
|
||||
chat_completion_tp_inner(tp, request, principal, context_limit_cfg)
|
||||
.instrument(span.clone()),
|
||||
chat_completion_tp_inner(
|
||||
tp,
|
||||
request,
|
||||
principal,
|
||||
context_limit_cfg,
|
||||
reasoning_budget_cfg,
|
||||
)
|
||||
.instrument(span.clone()),
|
||||
);
|
||||
match handle.await {
|
||||
Ok(Ok(resp)) => Ok(resp),
|
||||
@@ -5039,7 +5218,22 @@ impl CandleHarness {
|
||||
&tp.generation_defaults,
|
||||
unix_subsec_nanos(),
|
||||
);
|
||||
let max_new = request.max_tokens.unwrap_or(8192) as usize;
|
||||
// Bound the think block if the caller asked for it (#223).
|
||||
let mut governor = super::reasoning_budget::ReasoningGovernor::new(
|
||||
requested_reasoning_budget(&request, &self.reasoning_budget_cfg),
|
||||
tp.reasoning_tokens
|
||||
.as_ref()
|
||||
.map(|p| (p.open_id, p.close_id)),
|
||||
prompt_opens_reasoning(&prompt_tokens, tp.reasoning_tokens.as_ref()),
|
||||
);
|
||||
// The output budget the request declared, under either
|
||||
// spelling (#254), and the fallback when it declares none.
|
||||
// One value drives generation, KV pricing and admission — they
|
||||
// disagreed before, and the gate priced a budget nobody asked
|
||||
// for (#278).
|
||||
let requested_output = request.effective_max_tokens().map(|t| t as usize);
|
||||
let cfg_output_reserve = self.context_limit_cfg.output_reserve_tokens;
|
||||
let max_new = requested_output.unwrap_or(cfg_output_reserve);
|
||||
|
||||
let eos_id = tp
|
||||
.tokenizer
|
||||
@@ -5121,6 +5315,7 @@ impl CandleHarness {
|
||||
principal.as_deref(),
|
||||
kv_reservation_mb(
|
||||
prompt_len,
|
||||
requested_output,
|
||||
tp.context_profile.as_ref(),
|
||||
&self.context_limit_cfg,
|
||||
),
|
||||
@@ -5142,6 +5337,12 @@ impl CandleHarness {
|
||||
prompt_tokens,
|
||||
max_new,
|
||||
sampling: sampling.clone(),
|
||||
// The engine builds its own per-slot governor from
|
||||
// this, since slots in a batch differ (#223).
|
||||
reasoning_budget: requested_reasoning_budget(
|
||||
&request,
|
||||
&self.reasoning_budget_cfg,
|
||||
),
|
||||
eos_id,
|
||||
tool_schemas,
|
||||
tx,
|
||||
@@ -5349,7 +5550,7 @@ impl CandleHarness {
|
||||
}
|
||||
};
|
||||
let mut next_token =
|
||||
match sample_with_penalty(&logits, &all_tokens, &mut logits_processor) {
|
||||
match sample_with_penalty(&logits, &all_tokens, &mut logits_processor, &sampling, governor.as_mut()) {
|
||||
Ok(t) => t,
|
||||
Err(e) => {
|
||||
let health = logits_health_slice(&logits_vec);
|
||||
@@ -5477,6 +5678,8 @@ impl CandleHarness {
|
||||
&logits,
|
||||
&all_tokens,
|
||||
&mut logits_processor,
|
||||
&sampling,
|
||||
governor.as_mut(),
|
||||
) {
|
||||
Ok(t) => t,
|
||||
Err(e) => {
|
||||
@@ -5709,6 +5912,7 @@ async fn chat_completion_tp_inner(
|
||||
request: ChatCompletionRequest,
|
||||
principal: Option<String>,
|
||||
context_limit_cfg: crate::config::ContextLimitConfig,
|
||||
reasoning_budget_cfg: super::reasoning_budget::ReasoningBudgetLadder,
|
||||
) -> Result<ChatCompletionResponse, InferenceError> {
|
||||
let req_start = std::time::Instant::now();
|
||||
let model_id = request.model.clone();
|
||||
@@ -5776,7 +5980,22 @@ async fn chat_completion_tp_inner(
|
||||
&tp.generation_defaults,
|
||||
unix_subsec_nanos(),
|
||||
);
|
||||
let max_new = request.max_tokens.unwrap_or(8192) as usize;
|
||||
// Bound the think block if the caller asked for it (#223).
|
||||
let mut governor = super::reasoning_budget::ReasoningGovernor::new(
|
||||
requested_reasoning_budget(&request, &reasoning_budget_cfg),
|
||||
tp.reasoning_tokens
|
||||
.as_ref()
|
||||
.map(|p| (p.open_id, p.close_id)),
|
||||
prompt_opens_reasoning(&prompt_tokens, tp.reasoning_tokens.as_ref()),
|
||||
);
|
||||
// The output budget the request declared, under either
|
||||
// spelling (#254), and the fallback when it declares none.
|
||||
// One value drives generation, KV pricing and admission — they
|
||||
// disagreed before, and the gate priced a budget nobody asked
|
||||
// for (#278).
|
||||
let requested_output = request.effective_max_tokens().map(|t| t as usize);
|
||||
let cfg_output_reserve = context_limit_cfg.output_reserve_tokens;
|
||||
let max_new = requested_output.unwrap_or(cfg_output_reserve);
|
||||
|
||||
let eos_id = tp
|
||||
.tokenizer
|
||||
@@ -5815,7 +6034,12 @@ async fn chat_completion_tp_inner(
|
||||
.admission
|
||||
.enter_with_kv(
|
||||
principal.as_deref(),
|
||||
kv_reservation_mb(prompt_len, tp.context_profile.as_ref(), &context_limit_cfg),
|
||||
kv_reservation_mb(
|
||||
prompt_len,
|
||||
requested_output,
|
||||
tp.context_profile.as_ref(),
|
||||
&context_limit_cfg,
|
||||
),
|
||||
)
|
||||
.await
|
||||
.map_err(InferenceError::from)?;
|
||||
@@ -5926,7 +6150,13 @@ async fn chat_completion_tp_inner(
|
||||
// from CPU memory only.
|
||||
let logits = Tensor::new(logits_vec.as_slice(), &Device::Cpu)
|
||||
.map_err(|e| InferenceError::Other(anyhow::anyhow!("build cpu logits: {e:#}")))?;
|
||||
let mut next_token = match sample_with_penalty(&logits, &generated, &mut logits_processor) {
|
||||
let mut next_token = match sample_with_penalty(
|
||||
&logits,
|
||||
&generated,
|
||||
&mut logits_processor,
|
||||
&sampling,
|
||||
governor.as_mut(),
|
||||
) {
|
||||
Ok(t) => t,
|
||||
Err(e) => {
|
||||
// Logits health snapshot — the surrounding wrapper logs
|
||||
@@ -5963,7 +6193,13 @@ async fn chat_completion_tp_inner(
|
||||
let logits = Tensor::new(logits_vec.as_slice(), &Device::Cpu).map_err(|e| {
|
||||
InferenceError::Other(anyhow::anyhow!("build cpu logits step {index}: {e:#}"))
|
||||
})?;
|
||||
next_token = match sample_with_penalty(&logits, &generated, &mut logits_processor) {
|
||||
next_token = match sample_with_penalty(
|
||||
&logits,
|
||||
&generated,
|
||||
&mut logits_processor,
|
||||
&sampling,
|
||||
governor.as_mut(),
|
||||
) {
|
||||
Ok(t) => t,
|
||||
Err(e) => {
|
||||
let health = logits_health_slice(&logits_vec);
|
||||
@@ -6916,6 +7152,9 @@ async fn run_inference_with_images_via_worker(
|
||||
image_token_id: u32,
|
||||
max_new: usize,
|
||||
sampling: super::sampling::SamplingParams,
|
||||
// Bounds how long the model may think before it must answer (#223).
|
||||
// `None` = unbounded, which is the default and the prior behaviour.
|
||||
mut governor: Option<super::reasoning_budget::ReasoningGovernor>,
|
||||
eos_id: Option<u32>,
|
||||
) -> Result<(Vec<u32>, String)> {
|
||||
let mut logits_processor =
|
||||
@@ -6935,7 +7174,13 @@ async fn run_inference_with_images_via_worker(
|
||||
.await
|
||||
.map_err(|e| anyhow::anyhow!("forward_logits_with_images: {e:#}"))?;
|
||||
let logits = Tensor::new(logits_vec.as_slice(), &Device::Cpu)?;
|
||||
let mut next_token = match sample_with_penalty(&logits, &generated, &mut logits_processor) {
|
||||
let mut next_token = match sample_with_penalty(
|
||||
&logits,
|
||||
&generated,
|
||||
&mut logits_processor,
|
||||
&sampling,
|
||||
governor.as_mut(),
|
||||
) {
|
||||
Ok(t) => t,
|
||||
Err(e) => {
|
||||
let health = logits_health_slice(&logits_vec);
|
||||
@@ -6958,7 +7203,13 @@ async fn run_inference_with_images_via_worker(
|
||||
.await
|
||||
.map_err(|e| anyhow::anyhow!("decode step {index}: {e:#}"))?;
|
||||
let logits = Tensor::new(logits_vec.as_slice(), &Device::Cpu)?;
|
||||
next_token = match sample_with_penalty(&logits, &generated, &mut logits_processor) {
|
||||
next_token = match sample_with_penalty(
|
||||
&logits,
|
||||
&generated,
|
||||
&mut logits_processor,
|
||||
&sampling,
|
||||
governor.as_mut(),
|
||||
) {
|
||||
Ok(t) => t,
|
||||
Err(e) => {
|
||||
let health = logits_health_slice(&logits_vec);
|
||||
@@ -7269,6 +7520,9 @@ async fn run_inference_via_worker(
|
||||
im_start_id: Option<u32>,
|
||||
max_new: usize,
|
||||
sampling: super::sampling::SamplingParams,
|
||||
// Bounds how long the model may think before it must answer (#223).
|
||||
// `None` = unbounded, which is the default and the prior behaviour.
|
||||
mut governor: Option<super::reasoning_budget::ReasoningGovernor>,
|
||||
eos_id: Option<u32>,
|
||||
) -> Result<(Vec<u32>, String)> {
|
||||
let mut logits_processor =
|
||||
@@ -7311,7 +7565,13 @@ async fn run_inference_via_worker(
|
||||
None => chunked_prefill_via_worker(worker, handle, prompt_tokens, reused).await?,
|
||||
};
|
||||
let logits = Tensor::new(logits_vec.as_slice(), &Device::Cpu)?;
|
||||
let mut next_token = match sample_with_penalty(&logits, &generated, &mut logits_processor) {
|
||||
let mut next_token = match sample_with_penalty(
|
||||
&logits,
|
||||
&generated,
|
||||
&mut logits_processor,
|
||||
&sampling,
|
||||
governor.as_mut(),
|
||||
) {
|
||||
Ok(t) => t,
|
||||
Err(e) => {
|
||||
let health = logits_health_slice(&logits_vec);
|
||||
@@ -7334,7 +7594,13 @@ async fn run_inference_via_worker(
|
||||
.await
|
||||
.map_err(|e| anyhow::anyhow!("decode step {index}: {e:#}"))?;
|
||||
let logits = Tensor::new(logits_vec.as_slice(), &Device::Cpu)?;
|
||||
next_token = match sample_with_penalty(&logits, &generated, &mut logits_processor) {
|
||||
next_token = match sample_with_penalty(
|
||||
&logits,
|
||||
&generated,
|
||||
&mut logits_processor,
|
||||
&sampling,
|
||||
governor.as_mut(),
|
||||
) {
|
||||
Ok(t) => t,
|
||||
Err(e) => {
|
||||
let health = logits_health_slice(&logits_vec);
|
||||
@@ -7385,6 +7651,9 @@ async fn stream_inference_via_worker(
|
||||
prefill_rate: &super::context_limit::PrefillRateEma,
|
||||
max_new: usize,
|
||||
sampling: super::sampling::SamplingParams,
|
||||
// Bounds how long the model may think before it must answer (#223).
|
||||
// `None` = unbounded, which is the default and the prior behaviour.
|
||||
mut governor: Option<super::reasoning_budget::ReasoningGovernor>,
|
||||
eos_id: Option<u32>,
|
||||
reasoning_tokens: Option<ReasoningTokenPair>,
|
||||
tool_call_tokens: Option<ToolCallTokenPair>,
|
||||
@@ -7487,7 +7756,13 @@ async fn stream_inference_via_worker(
|
||||
let prefill_elapsed = prefill_start.elapsed();
|
||||
prefill_rate.record(prefill_prompt_len, prefill_elapsed);
|
||||
let logits = Tensor::new(logits_vec.as_slice(), &Device::Cpu)?;
|
||||
let mut next_token = match sample_with_penalty(&logits, &all_tokens, &mut logits_processor) {
|
||||
let mut next_token = match sample_with_penalty(
|
||||
&logits,
|
||||
&all_tokens,
|
||||
&mut logits_processor,
|
||||
&sampling,
|
||||
governor.as_mut(),
|
||||
) {
|
||||
Ok(t) => t,
|
||||
Err(e) => {
|
||||
let health = logits_health_slice(&logits_vec);
|
||||
@@ -7611,7 +7886,13 @@ async fn stream_inference_via_worker(
|
||||
.await
|
||||
.map_err(|e| anyhow::anyhow!("decode step {index}: {e:#}"))?;
|
||||
let logits = Tensor::new(logits_vec.as_slice(), &Device::Cpu)?;
|
||||
next_token = match sample_with_penalty(&logits, &all_tokens, &mut logits_processor) {
|
||||
next_token = match sample_with_penalty(
|
||||
&logits,
|
||||
&all_tokens,
|
||||
&mut logits_processor,
|
||||
&sampling,
|
||||
governor.as_mut(),
|
||||
) {
|
||||
Ok(t) => t,
|
||||
Err(e) => {
|
||||
let health = logits_health_slice(&logits_vec);
|
||||
@@ -7678,6 +7959,9 @@ fn run_inference(
|
||||
im_start_id: Option<u32>,
|
||||
max_new: usize,
|
||||
sampling: super::sampling::SamplingParams,
|
||||
// Bounds how long the model may think before it must answer (#223).
|
||||
// `None` = unbounded, which is the default and the prior behaviour.
|
||||
mut governor: Option<super::reasoning_budget::ReasoningGovernor>,
|
||||
eos_id: Option<u32>,
|
||||
) -> Result<(Vec<u32>, String)> {
|
||||
let mut logits_processor =
|
||||
@@ -7701,7 +7985,13 @@ fn run_inference(
|
||||
}
|
||||
None => chunked_prefill_local(arch, device, prompt_tokens, reused)?,
|
||||
};
|
||||
let mut next_token = sample_with_penalty(&logits, &generated, &mut logits_processor)?;
|
||||
let mut next_token = sample_with_penalty(
|
||||
&logits,
|
||||
&generated,
|
||||
&mut logits_processor,
|
||||
&sampling,
|
||||
governor.as_mut(),
|
||||
)?;
|
||||
|
||||
let mut finish_reason = "length";
|
||||
if Some(next_token) == eos_id {
|
||||
@@ -7711,7 +8001,13 @@ fn run_inference(
|
||||
for index in 0..max_new.saturating_sub(1) {
|
||||
let input = Tensor::new(&[next_token], device)?.unsqueeze(0)?;
|
||||
let logits = arch.forward(&input, prompt_tokens.len() + index)?;
|
||||
next_token = sample_with_penalty(&logits, &generated, &mut logits_processor)?;
|
||||
next_token = sample_with_penalty(
|
||||
&logits,
|
||||
&generated,
|
||||
&mut logits_processor,
|
||||
&sampling,
|
||||
governor.as_mut(),
|
||||
)?;
|
||||
if Some(next_token) == eos_id {
|
||||
finish_reason = "stop";
|
||||
break;
|
||||
@@ -7738,6 +8034,9 @@ fn run_inference_streaming(
|
||||
prefix_cache: Option<&ModelPrefixCache>,
|
||||
max_new: usize,
|
||||
sampling: super::sampling::SamplingParams,
|
||||
// Bounds how long the model may think before it must answer (#223).
|
||||
// `None` = unbounded, which is the default and the prior behaviour.
|
||||
mut governor: Option<super::reasoning_budget::ReasoningGovernor>,
|
||||
eos_id: Option<u32>,
|
||||
reasoning_tokens: Option<&ReasoningTokenPair>,
|
||||
tool_call_tokens: Option<&ToolCallTokenPair>,
|
||||
@@ -7796,7 +8095,13 @@ fn run_inference_streaming(
|
||||
}
|
||||
None => chunked_prefill_local(arch, device, prompt_tokens, reused)?,
|
||||
};
|
||||
let mut next_token = sample_with_penalty(&logits, &all_tokens, &mut logits_processor)?;
|
||||
let mut next_token = sample_with_penalty(
|
||||
&logits,
|
||||
&all_tokens,
|
||||
&mut logits_processor,
|
||||
&sampling,
|
||||
governor.as_mut(),
|
||||
)?;
|
||||
let prefill_elapsed = prefill_start.elapsed();
|
||||
let decode_start = std::time::Instant::now();
|
||||
|
||||
@@ -7891,7 +8196,13 @@ fn run_inference_streaming(
|
||||
for index in 0..max_new.saturating_sub(1) {
|
||||
let input = Tensor::new(&[next_token], device)?.unsqueeze(0)?;
|
||||
let logits = arch.forward(&input, prompt_tokens.len() + index)?;
|
||||
next_token = sample_with_penalty(&logits, &all_tokens, &mut logits_processor)?;
|
||||
next_token = sample_with_penalty(
|
||||
&logits,
|
||||
&all_tokens,
|
||||
&mut logits_processor,
|
||||
&sampling,
|
||||
governor.as_mut(),
|
||||
)?;
|
||||
if Some(next_token) == eos_id {
|
||||
finish_reason = FinishReason::Stop;
|
||||
break;
|
||||
@@ -7943,6 +8254,147 @@ fn unix_subsec_nanos() -> u64 {
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::super::context_limit::ContextProfile;
|
||||
use super::super::sampling::SamplingParams;
|
||||
use super::sample_with_penalty;
|
||||
use candle_core::{Device, Tensor};
|
||||
use candle_transformers::generation::{LogitsProcessor, Sampling};
|
||||
|
||||
/// A resolved `SamplingParams` with everything but the repetition
|
||||
/// knobs pinned to a deterministic sampler, so the assertions below
|
||||
/// are about the penalty and nothing else.
|
||||
fn greedy_with(repeat_penalty: f32, repeat_last_n: usize) -> SamplingParams {
|
||||
SamplingParams {
|
||||
temperature: 0.0,
|
||||
top_p: None,
|
||||
top_k: None,
|
||||
seed: 0,
|
||||
repeat_penalty,
|
||||
repeat_last_n,
|
||||
}
|
||||
}
|
||||
|
||||
/// The three inlets converge on one field (#223): an explicit token
|
||||
/// count, Anthropic's normalised `budget_tokens`, or an effort rung.
|
||||
#[test]
|
||||
fn the_reasoning_budget_resolves_from_every_inlet() {
|
||||
use cortex_core::openai::{ChatCompletionRequest, MessageContent};
|
||||
let ladder = crate::harness::reasoning_budget::ReasoningBudgetLadder::default();
|
||||
let req = |extra: serde_json::Value| ChatCompletionRequest {
|
||||
model: "m".into(),
|
||||
messages: vec![],
|
||||
temperature: None,
|
||||
top_p: None,
|
||||
top_k: None,
|
||||
seed: None,
|
||||
repetition_penalty: None,
|
||||
repeat_last_n: None,
|
||||
max_tokens: None,
|
||||
max_completion_tokens: None,
|
||||
stream: None,
|
||||
extra,
|
||||
};
|
||||
let _ = MessageContent::Null;
|
||||
|
||||
// Explicit count — also the shape Anthropic's `budget_tokens`
|
||||
// is normalised into during translation.
|
||||
assert_eq!(
|
||||
requested_reasoning_budget(
|
||||
&req(serde_json::json!({"reasoning": {"max_tokens": 5000}})),
|
||||
&ladder
|
||||
),
|
||||
Some(5000)
|
||||
);
|
||||
// Effort rung, resolved against the operator's configuration.
|
||||
assert_eq!(
|
||||
requested_reasoning_budget(
|
||||
&req(serde_json::json!({"reasoning": {"effort": "medium"}})),
|
||||
&ladder
|
||||
),
|
||||
Some(12_288)
|
||||
);
|
||||
// An explicit count beats a rung — it is the more specific ask.
|
||||
assert_eq!(
|
||||
requested_reasoning_budget(
|
||||
&req(serde_json::json!({"reasoning": {"effort": "low", "max_tokens": 999}})),
|
||||
&ladder
|
||||
),
|
||||
Some(999)
|
||||
);
|
||||
// No control at all stays unbounded, exactly as before #223.
|
||||
assert_eq!(
|
||||
requested_reasoning_budget(&req(serde_json::json!({})), &ladder),
|
||||
None
|
||||
);
|
||||
// `none` disables thinking via the template; it is not a budget.
|
||||
assert_eq!(
|
||||
requested_reasoning_budget(
|
||||
&req(serde_json::json!({"reasoning": {"effort": "none"}})),
|
||||
&ladder
|
||||
),
|
||||
None
|
||||
);
|
||||
}
|
||||
|
||||
/// The regression: `repetition_penalty` was modelled on the wire and
|
||||
/// resolved into `SamplingParams`, then the sampler read a module
|
||||
/// constant instead — so the knob was accepted and discarded, which
|
||||
/// is the exact defect #271 exists to close.
|
||||
#[test]
|
||||
fn the_requested_repetition_penalty_reaches_the_sampler() {
|
||||
let logits = Tensor::new(&[3.0f32, 2.0], &Device::Cpu).expect("logits");
|
||||
let history = [0u32];
|
||||
let mut lp = LogitsProcessor::from_sampling(0, Sampling::ArgMax);
|
||||
|
||||
// 1.0 disables the penalty: the leading token still wins.
|
||||
let plain = sample_with_penalty(&logits, &history, &mut lp, &greedy_with(1.0, 64), None)
|
||||
.expect("sample");
|
||||
assert_eq!(plain, 0);
|
||||
|
||||
// 2.0 halves the logit of the token already generated, which is
|
||||
// enough to hand the sample to its rival.
|
||||
let penalised =
|
||||
sample_with_penalty(&logits, &history, &mut lp, &greedy_with(2.0, 64), None)
|
||||
.expect("sample");
|
||||
assert_eq!(
|
||||
penalised, 1,
|
||||
"a penalty the caller set must change the sample"
|
||||
);
|
||||
}
|
||||
|
||||
/// `repeat_last_n` bounds how far back the penalty sees — the knob
|
||||
/// that matters for a long reasoning block, where a derivation
|
||||
/// repeated 500 tokens later sits well outside the 64-token default.
|
||||
#[test]
|
||||
fn the_penalty_window_bounds_the_history_it_sees() {
|
||||
let logits = Tensor::new(&[3.0f32, 2.0], &Device::Cpu).expect("logits");
|
||||
// Oldest first: token 1 was generated, then token 0.
|
||||
let history = [1u32, 0u32];
|
||||
let mut lp = LogitsProcessor::from_sampling(0, Sampling::ArgMax);
|
||||
|
||||
// Window of 1 sees only the most recent token (0), penalising it
|
||||
// to 1.5 and leaving token 1 at 2.0.
|
||||
let narrow = sample_with_penalty(&logits, &history, &mut lp, &greedy_with(2.0, 1), None)
|
||||
.expect("sample");
|
||||
assert_eq!(narrow, 1);
|
||||
|
||||
// Window of 2 reaches back far enough to penalise token 1 as
|
||||
// well (2.0 -> 1.0), so token 0 wins at 1.5.
|
||||
let wide = sample_with_penalty(&logits, &history, &mut lp, &greedy_with(2.0, 2), None)
|
||||
.expect("sample");
|
||||
assert_eq!(wide, 0, "the window must decide what is penalised");
|
||||
}
|
||||
|
||||
/// A zero window is the other way to say "no penalty", and must not
|
||||
/// slice the history backwards.
|
||||
#[test]
|
||||
fn a_zero_window_disables_the_penalty() {
|
||||
let logits = Tensor::new(&[3.0f32, 2.0], &Device::Cpu).expect("logits");
|
||||
let history = [0u32];
|
||||
let mut lp = LogitsProcessor::from_sampling(0, Sampling::ArgMax);
|
||||
let sampled = sample_with_penalty(&logits, &history, &mut lp, &greedy_with(2.0, 0), None)
|
||||
.expect("sample");
|
||||
assert_eq!(sampled, 0);
|
||||
}
|
||||
|
||||
/// Why the serviceability probe shares [`check_vram`] with the
|
||||
/// request path rather than re-implementing the floor.
|
||||
@@ -7968,7 +8420,7 @@ mod tests {
|
||||
// the reserve's KV plus activation headroom plus the floor
|
||||
// already exceeds what is free.
|
||||
assert!(
|
||||
check_vram(0, free_mb, Some(&profile), &cfg).is_err(),
|
||||
check_vram(0, None, free_mb, Some(&profile), &cfg).is_err(),
|
||||
"an empty prompt must still be rejected when the output \
|
||||
reserve cannot fit"
|
||||
);
|
||||
@@ -7984,7 +8436,7 @@ mod tests {
|
||||
kv_bytes_per_token_per_card: 65_536,
|
||||
world_size: 2,
|
||||
};
|
||||
assert!(check_vram(0, 60_000, Some(&profile), &cfg).is_ok());
|
||||
assert!(check_vram(0, None, 60_000, Some(&profile), &cfg).is_ok());
|
||||
}
|
||||
|
||||
/// CPU loads report free VRAM as the `0` sentinel and must not be
|
||||
@@ -7992,7 +8444,7 @@ mod tests {
|
||||
#[test]
|
||||
fn the_cpu_sentinel_is_never_a_rejection() {
|
||||
let cfg = crate::config::ContextLimitConfig::default();
|
||||
assert!(check_vram(0, 0, None, &cfg).is_ok());
|
||||
assert!(check_vram(0, None, 0, None, &cfg).is_ok());
|
||||
}
|
||||
|
||||
use super::*;
|
||||
@@ -8464,9 +8916,6 @@ mod tests {
|
||||
}
|
||||
}
|
||||
|
||||
/// A prompt under the cap with ample free VRAM passes; the same
|
||||
/// prompt over the cap is `PromptTooLong` before any VRAM math.
|
||||
#[test]
|
||||
/// Regression, observed live 2026-08-15. With the KV gate enabled,
|
||||
/// free VRAM is *supposed* to sit near zero while long sequences are
|
||||
/// resident — that is admission working, not the node dying. The
|
||||
@@ -8585,6 +9034,7 @@ mod tests {
|
||||
/// queues a request that is still served. Making the cache evict
|
||||
/// snapshots to admit a sequence — rather than the two silently
|
||||
/// competing — is the natural follow-up.
|
||||
#[test]
|
||||
fn kv_budget_prices_a_long_context_session() {
|
||||
let cfg = crate::config::ContextLimitConfig::default();
|
||||
let profile = super::super::context_limit::ContextProfile {
|
||||
@@ -8596,7 +9046,7 @@ mod tests {
|
||||
let budget = kv_budget_mb(9_286, &cfg, 1_024);
|
||||
assert_eq!(budget, 4_714);
|
||||
|
||||
let per_seq = kv_reservation_mb(67_881, Some(&profile), &cfg);
|
||||
let per_seq = kv_reservation_mb(67_881, None, Some(&profile), &cfg);
|
||||
// (67_881 + 8_192) * 32_768 / 1 MiB
|
||||
assert_eq!(per_seq, 2_377);
|
||||
|
||||
@@ -8629,13 +9079,33 @@ mod tests {
|
||||
#[test]
|
||||
fn unpriced_without_profile() {
|
||||
let cfg = crate::config::ContextLimitConfig::default();
|
||||
assert_eq!(kv_reservation_mb(50_000, None, &cfg), 0);
|
||||
assert_eq!(kv_reservation_mb(50_000, None, None, &cfg), 0);
|
||||
}
|
||||
|
||||
/// The gate can only be as honest as the number it is given (#278).
|
||||
/// A request declaring a large budget must reserve for it, or it is
|
||||
/// admitted against KV it will go on to occupy unaccounted.
|
||||
#[test]
|
||||
fn the_reservation_follows_the_declared_output_budget() {
|
||||
let cfg = crate::config::ContextLimitConfig::default();
|
||||
let profile = backstop_profile();
|
||||
let default_budget = kv_reservation_mb(1_000, None, Some(&profile), &cfg);
|
||||
let declared_small = kv_reservation_mb(1_000, Some(1_024), Some(&profile), &cfg);
|
||||
let declared_large = kv_reservation_mb(1_000, Some(32_768), Some(&profile), &cfg);
|
||||
assert!(
|
||||
declared_large > default_budget,
|
||||
"a 32k-token request must reserve more than the 8k default"
|
||||
);
|
||||
assert!(
|
||||
declared_small < default_budget,
|
||||
"a request asking for less must not be charged the reserve"
|
||||
);
|
||||
let degenerate = super::super::context_limit::ContextProfile {
|
||||
max_position_embeddings: 262_144,
|
||||
kv_bytes_per_token_per_card: 0,
|
||||
world_size: 1,
|
||||
};
|
||||
assert_eq!(kv_reservation_mb(50_000, Some(°enerate), &cfg), 0);
|
||||
assert_eq!(kv_reservation_mb(50_000, None, Some(°enerate), &cfg), 0);
|
||||
}
|
||||
|
||||
/// A card with less free VRAM than the fixed reserves yields a zero
|
||||
@@ -8646,6 +9116,9 @@ mod tests {
|
||||
assert_eq!(kv_budget_mb(100, &cfg, 1_024), 0);
|
||||
}
|
||||
|
||||
/// A prompt under the cap with ample free VRAM passes; the same
|
||||
/// prompt over the cap is `PromptTooLong` before any VRAM math.
|
||||
#[test]
|
||||
fn validate_request_cap_and_fit() {
|
||||
let cfg = crate::config::ContextLimitConfig::default();
|
||||
let profile = backstop_profile();
|
||||
@@ -8718,7 +9191,7 @@ mod tests {
|
||||
assert_eq!(cap, 87_040);
|
||||
|
||||
// (87_040 + 8_192) tokens × 32 KiB / 1 MiB
|
||||
let footprint = kv_reservation_mb(cap, Some(&profile), &cfg);
|
||||
let footprint = kv_reservation_mb(cap, None, Some(&profile), &cfg);
|
||||
assert_eq!(footprint, 2_976);
|
||||
|
||||
// `validate_request` keeps the cap and the static floor. It no
|
||||
|
||||
@@ -233,7 +233,24 @@ pub fn render_chat_template(
|
||||
MessageContent::Null => Value::Null,
|
||||
};
|
||||
let mut obj = serde_json::Map::new();
|
||||
obj.insert("role".into(), Value::String(m.role.clone()));
|
||||
// `developer` is OpenAI's successor to `system` and is what
|
||||
// a client sends once a model is declared reasoning-capable
|
||||
// (pi-ai: `model.reasoning && supportsDeveloperRole`). Most
|
||||
// chat templates — every one on this fleet — only know the
|
||||
// older spelling and `raise_exception` on anything else, so
|
||||
// an unmodified client got a 422 for a system prompt it sent
|
||||
// correctly.
|
||||
//
|
||||
// Rewritten only when the template does not handle the role
|
||||
// itself: a template that knows `developer` may treat it as
|
||||
// distinct from `system`, and substituting there would be us
|
||||
// deciding something the model's authors already decided.
|
||||
let role = if m.role == "developer" && !template.contains("developer") {
|
||||
"system"
|
||||
} else {
|
||||
m.role.as_str()
|
||||
};
|
||||
obj.insert("role".into(), Value::String(role.to_string()));
|
||||
obj.insert("content".into(), content_value);
|
||||
// Forward extras (e.g. tool_calls on assistant turns,
|
||||
// tool_call_id on tool result turns). HF templates that
|
||||
@@ -498,6 +515,78 @@ mod tests {
|
||||
/// Minimal Qwen3-style template — enough surface to confirm
|
||||
/// our renderer threads role + content correctly without
|
||||
/// loading a real model's tokenizer_config.json.
|
||||
/// A reasoning-capable model makes OpenAI clients send the system
|
||||
/// prompt as `developer` — and every chat template on this fleet
|
||||
/// raises on an unknown role, so the caller got a 422 for a prompt
|
||||
/// it had sent correctly. Observed live: pi with `reasoning: true`,
|
||||
/// `template_render_failed: Unexpected message role`.
|
||||
#[test]
|
||||
fn the_developer_role_reaches_a_template_that_only_knows_system() {
|
||||
const SYSTEM_ONLY: &str = "{%- for message in messages -%}\
|
||||
{%- if message.role == 'system' -%}SYS:{{ message.content }}\
|
||||
{%- elif message.role == 'user' -%}USR:{{ message.content }}\
|
||||
{%- else -%}{{ raise_exception('Unexpected message role.') }}\
|
||||
{%- endif -%}\
|
||||
{%- endfor -%}";
|
||||
let messages = vec![
|
||||
ChatMessage {
|
||||
role: "developer".into(),
|
||||
content: MessageContent::Text("be terse".into()),
|
||||
extra: Value::Object(Default::default()),
|
||||
},
|
||||
user_msg("hi"),
|
||||
];
|
||||
let out = render_chat_template(SYSTEM_ONLY, &messages, &Value::Null, &Value::Null)
|
||||
.expect("a developer role must not blow up a system-only template");
|
||||
assert!(out.contains("SYS:be terse"), "got {out:?}");
|
||||
}
|
||||
|
||||
/// A template that models `developer` itself keeps it: the role may
|
||||
/// mean something distinct there, and substituting would override a
|
||||
/// decision the model's authors already made.
|
||||
#[test]
|
||||
fn a_template_that_knows_developer_keeps_it() {
|
||||
const KNOWS_DEVELOPER: &str = "{%- for message in messages -%}\
|
||||
{%- if message.role == 'developer' -%}DEV:{{ message.content }}\
|
||||
{%- else -%}{{ message.role }}:{{ message.content }}\
|
||||
{%- endif -%}\
|
||||
{%- endfor -%}";
|
||||
let messages = vec![ChatMessage {
|
||||
role: "developer".into(),
|
||||
content: MessageContent::Text("be terse".into()),
|
||||
extra: Value::Object(Default::default()),
|
||||
}];
|
||||
let out = render_chat_template(KNOWS_DEVELOPER, &messages, &Value::Null, &Value::Null)
|
||||
.expect("render");
|
||||
assert!(out.contains("DEV:be terse"), "got {out:?}");
|
||||
}
|
||||
|
||||
/// The fetcher and the reader must agree on the filenames, or the
|
||||
/// reader looks for something nobody downloaded and falls back to a
|
||||
/// built-in prompt format without anything failing (#280 — observed
|
||||
/// live: every model neuron downloaded itself was served with the
|
||||
/// fallback because no sidecar was ever fetched).
|
||||
#[test]
|
||||
fn every_template_source_this_reads_is_one_the_loader_fetches() {
|
||||
for name in crate::harness::candle::CandleHarness::SIDECAR_FILES {
|
||||
assert!(
|
||||
!name.is_empty(),
|
||||
"sidecar names are the contract between fetch and read"
|
||||
);
|
||||
}
|
||||
// The three the reader consults, in its own priority order.
|
||||
for source in [
|
||||
"chat_template.jinja",
|
||||
"chat_template.json",
|
||||
"tokenizer_config.json",
|
||||
] {
|
||||
assert!(
|
||||
crate::harness::candle::CandleHarness::SIDECAR_FILES.contains(&source),
|
||||
"{source} is read by load_chat_template_alongside but never fetched"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
const QWEN3_LIKE: &str = "{%- for message in messages -%}\
|
||||
<|im_start|>{{ message.role }}\n{{ message.content }}<|im_end|>\n\
|
||||
{%- endfor -%}\
|
||||
|
||||
@@ -234,10 +234,23 @@ fn round_down(tokens: usize, granularity: usize) -> usize {
|
||||
|
||||
const CONTEXT_GRANULARITY: usize = 1024;
|
||||
|
||||
/// Default ceiling on what a request may name as its output budget.
|
||||
///
|
||||
/// Not derived from anything — a deliberate figure, and configurable
|
||||
/// per host (`[context_limit] max_output_tokens`). The evidence behind
|
||||
/// it: a reasoning model asked to build a small application front-loads
|
||||
/// one long think block before it acts, measured at 27,504 tokens for a
|
||||
/// frontier model on a task Qwen3.8-27B was cut off attempting at
|
||||
/// 24,433 (#271, #278). A ceiling below that turns a capable model into
|
||||
/// one that never reaches its first tool call, so the default sits
|
||||
/// above the observed cost of that class of turn.
|
||||
const DEFAULT_OUTPUT_CEILING: usize = 32_768;
|
||||
|
||||
/// Derive `limit{context,input,output}` for a loaded model.
|
||||
///
|
||||
/// ```text
|
||||
/// output = output_reserve_tokens
|
||||
/// output = output_reserve_tokens (default + KV planning)
|
||||
/// output_ceiling = min(max_output_tokens, max_position_embeddings)
|
||||
/// vram_ceiling = (free_tightest − activation_headroom − min_free_floor) / kv_bytes_per_token_per_card
|
||||
/// throughput_ceiling = target_prefill_latency_secs × prefill_tok_per_sec
|
||||
/// context = min(max_position_embeddings, vram_ceiling, throughput_ceiling) [clamped by `hard_ceiling` if set]
|
||||
@@ -318,10 +331,26 @@ pub fn derive_limit(
|
||||
);
|
||||
|
||||
let input = context.saturating_sub(output);
|
||||
// The ceiling a request may name, distinct from the reserve above
|
||||
// (#278). Bounded by the model's own position limit rather than by
|
||||
// the live context, because the advertisement is read once and
|
||||
// cached by clients: a ceiling that shrank with fleet VRAM would be
|
||||
// stale in whichever direction hurt — either rejecting what we said
|
||||
// we would serve, or leaving budget unclaimed. A request that will
|
||||
// not actually fit is refused at request time, with the numbers,
|
||||
// instead of being under-promised here.
|
||||
let output_ceiling = round_down(
|
||||
cfg.max_output_tokens
|
||||
.unwrap_or(DEFAULT_OUTPUT_CEILING)
|
||||
.min(profile.max_position_embeddings),
|
||||
CONTEXT_GRANULARITY,
|
||||
)
|
||||
.max(output);
|
||||
ModelLimit {
|
||||
context,
|
||||
input: Some(input),
|
||||
output,
|
||||
output_ceiling,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -329,6 +358,82 @@ pub fn derive_limit(
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
/// The ceiling a client is told to plan against must not be the
|
||||
/// reserve. Advertising the reserve is what told a reasoning model's
|
||||
/// harness to cap itself below the cost of one think block (#278).
|
||||
#[test]
|
||||
fn the_advertised_ceiling_is_not_the_reserve() {
|
||||
let cfg = ContextLimitConfig::default();
|
||||
let profile = ContextProfile {
|
||||
max_position_embeddings: 131_072,
|
||||
kv_bytes_per_token_per_card: 65_536,
|
||||
world_size: 2,
|
||||
};
|
||||
let limit = derive_limit(&profile, 0, 0.0, None, &cfg);
|
||||
assert_eq!(limit.output, cfg.output_reserve_tokens, "reserve unchanged");
|
||||
assert_eq!(limit.output_ceiling, DEFAULT_OUTPUT_CEILING);
|
||||
assert!(limit.output_ceiling > limit.output);
|
||||
}
|
||||
|
||||
/// A model whose own position limit is below the default ceiling
|
||||
/// cannot generate past it, so the advertisement must not claim it.
|
||||
#[test]
|
||||
fn the_ceiling_never_exceeds_the_models_own_position_limit() {
|
||||
let cfg = ContextLimitConfig::default();
|
||||
let profile = ContextProfile {
|
||||
max_position_embeddings: 16_384,
|
||||
kv_bytes_per_token_per_card: 65_536,
|
||||
world_size: 1,
|
||||
};
|
||||
let limit = derive_limit(&profile, 0, 0.0, None, &cfg);
|
||||
assert_eq!(limit.output_ceiling, 16_384);
|
||||
}
|
||||
|
||||
/// The operator's number wins, which is what makes the default a
|
||||
/// starting point rather than a guess baked into the binary.
|
||||
#[test]
|
||||
fn a_configured_ceiling_overrides_the_default() {
|
||||
let cfg = ContextLimitConfig {
|
||||
max_output_tokens: Some(65_536),
|
||||
..ContextLimitConfig::default()
|
||||
};
|
||||
let profile = ContextProfile {
|
||||
max_position_embeddings: 131_072,
|
||||
kv_bytes_per_token_per_card: 65_536,
|
||||
world_size: 2,
|
||||
};
|
||||
assert_eq!(
|
||||
derive_limit(&profile, 0, 0.0, None, &cfg).output_ceiling,
|
||||
65_536
|
||||
);
|
||||
}
|
||||
|
||||
/// The ceiling is deliberately *not* a function of live VRAM: it is
|
||||
/// read once and cached by clients, so it has to mean the same thing
|
||||
/// tomorrow. The live context still moves — that is what a request
|
||||
/// is judged against — but the contract does not.
|
||||
#[test]
|
||||
fn the_ceiling_holds_still_while_the_context_moves() {
|
||||
let cfg = ContextLimitConfig::default();
|
||||
let profile = ContextProfile {
|
||||
max_position_embeddings: 131_072,
|
||||
kv_bytes_per_token_per_card: 65_536,
|
||||
world_size: 2,
|
||||
};
|
||||
let roomy = derive_limit(&profile, 60_000, 5_000.0, None, &cfg);
|
||||
// Tight enough that the VRAM term binds — the same shape as the
|
||||
// fleet advertising 109568 one hour and 131072 the next.
|
||||
let tight = derive_limit(&profile, 6_000, 5_000.0, None, &cfg);
|
||||
assert!(
|
||||
tight.context < roomy.context,
|
||||
"the live window should shrink under VRAM pressure"
|
||||
);
|
||||
assert_eq!(
|
||||
tight.output_ceiling, roomy.output_ceiling,
|
||||
"the advertised ceiling must not move with fleet load"
|
||||
);
|
||||
}
|
||||
|
||||
/// beast Qwen3.6-27B: 16 full-attn layers, 4 kv heads, head_dim 256,
|
||||
/// f16 (2 B), TP=2 → 64 KiB/token total, 32 KiB/token/card.
|
||||
fn beast_profile() -> ContextProfile {
|
||||
|
||||
@@ -410,6 +410,10 @@ pub struct EngineRequest {
|
||||
/// → fallback (#272), rather than loose scalars each caller could
|
||||
/// default differently.
|
||||
pub sampling: crate::harness::sampling::SamplingParams,
|
||||
/// Reasoning-token budget for this request (#223). `None` is
|
||||
/// unbounded — the default, and the behaviour before it existed.
|
||||
/// Per-request because slots in one batch can differ.
|
||||
pub reasoning_budget: Option<usize>,
|
||||
pub eos_id: Option<u32>,
|
||||
pub tool_schemas: ToolSchemas,
|
||||
pub tx: mpsc::Sender<InferenceEvent>,
|
||||
@@ -467,6 +471,12 @@ struct Slot {
|
||||
max_new: usize,
|
||||
eos_id: Option<u32>,
|
||||
lp: LogitsProcessor,
|
||||
/// The slot's resolved sampling (#272). Held per-slot because the
|
||||
/// repetition penalty and its window are per-request knobs, and
|
||||
/// slots in one batch can carry different ones.
|
||||
sampling: crate::harness::sampling::SamplingParams,
|
||||
/// Bounds this slot's think block (#223); `None` = unbounded.
|
||||
governor: Option<crate::harness::reasoning_budget::ReasoningGovernor>,
|
||||
router: mpsc::Sender<RouterMsg>,
|
||||
/// Set by the router when the consumer hangs up; the engine stops
|
||||
/// feeding the slot and compacts it out.
|
||||
@@ -574,7 +584,13 @@ async fn run_engine(cfg: EngineConfig, mut rx: mpsc::Receiver<EngineRequest>) {
|
||||
break;
|
||||
}
|
||||
};
|
||||
let nt = match sample_with_penalty(&logits, &slot.generated, &mut slot.lp) {
|
||||
let nt = match sample_with_penalty(
|
||||
&logits,
|
||||
&slot.generated,
|
||||
&mut slot.lp,
|
||||
&slot.sampling,
|
||||
slot.governor.as_mut(),
|
||||
) {
|
||||
Ok(t) => t,
|
||||
Err(e) => {
|
||||
let health = logits_health_slice(&logits_vec);
|
||||
@@ -779,6 +795,7 @@ async fn prefill_join(
|
||||
prompt_tokens,
|
||||
max_new,
|
||||
sampling,
|
||||
reasoning_budget,
|
||||
eos_id,
|
||||
tool_schemas,
|
||||
tx,
|
||||
@@ -787,6 +804,15 @@ async fn prefill_join(
|
||||
} = req;
|
||||
|
||||
let mut lp = LogitsProcessor::from_sampling(sampling.seed, sampling.to_sampling());
|
||||
// Built before prefill's first sample so a prompt that opens the
|
||||
// think block is governed from its very first token (#223).
|
||||
let mut governor = crate::harness::reasoning_budget::ReasoningGovernor::new(
|
||||
reasoning_budget,
|
||||
cfg.reasoning_tokens
|
||||
.as_ref()
|
||||
.map(|p| (p.open_id, p.close_id)),
|
||||
prompt_opens_reasoning(&prompt_tokens, cfg.reasoning_tokens.as_ref()),
|
||||
);
|
||||
|
||||
let prompt_len = prompt_tokens.len();
|
||||
let prefill_start = std::time::Instant::now();
|
||||
@@ -796,17 +822,18 @@ async fn prefill_join(
|
||||
// First token from the prefill logits.
|
||||
let generated: Vec<u32> = Vec::new();
|
||||
let logits = Tensor::new(logits_vec.as_slice(), &Device::Cpu)?;
|
||||
let first = match sample_with_penalty(&logits, &generated, &mut lp) {
|
||||
Ok(t) => t,
|
||||
Err(e) => {
|
||||
let health = logits_health_slice(&logits_vec);
|
||||
tracing::warn!(
|
||||
?health,
|
||||
"batch engine: prefill sample failed; logits unhealthy"
|
||||
);
|
||||
return Err(e);
|
||||
}
|
||||
};
|
||||
let first =
|
||||
match sample_with_penalty(&logits, &generated, &mut lp, &sampling, governor.as_mut()) {
|
||||
Ok(t) => t,
|
||||
Err(e) => {
|
||||
let health = logits_health_slice(&logits_vec);
|
||||
tracing::warn!(
|
||||
?health,
|
||||
"batch engine: prefill sample failed; logits unhealthy"
|
||||
);
|
||||
return Err(e);
|
||||
}
|
||||
};
|
||||
|
||||
// Router task for this slot.
|
||||
let hangup = Arc::new(AtomicBool::new(false));
|
||||
@@ -835,6 +862,8 @@ async fn prefill_join(
|
||||
max_new,
|
||||
eos_id,
|
||||
lp,
|
||||
sampling,
|
||||
governor,
|
||||
router: router_tx,
|
||||
hangup,
|
||||
finished: None,
|
||||
@@ -1186,6 +1215,9 @@ mod tests {
|
||||
&crate::harness::sampling::ModelGenerationDefaults::default(),
|
||||
0,
|
||||
),
|
||||
// Unbounded: this fixture asserts exact token sequences,
|
||||
// so nothing may substitute a close marker (#223).
|
||||
reasoning_budget: None,
|
||||
eos_id: None,
|
||||
tool_schemas: ToolSchemas::new(),
|
||||
tx,
|
||||
|
||||
@@ -12,6 +12,7 @@ pub mod image;
|
||||
pub mod prefix_cache;
|
||||
pub mod preflight;
|
||||
pub mod preprocess;
|
||||
pub mod reasoning_budget;
|
||||
pub mod sampling;
|
||||
pub mod speculative;
|
||||
pub mod tp;
|
||||
|
||||
@@ -0,0 +1,279 @@
|
||||
//! Bounding how long a model may think before it must answer (#223).
|
||||
//!
|
||||
//! A reasoning model with no stopping criterion will fill whatever
|
||||
//! output budget it is given. Measured on Qwen3.8-27B: given 16,384
|
||||
//! tokens it used 16,384; given 32,768 it used 32,768; both times it was
|
||||
//! still opening new topics when the cap bit, and both times the caller
|
||||
//! received no answer at all.
|
||||
//!
|
||||
//! The fix is not to stop the model thinking — the run that did produce
|
||||
//! files designed for 27,157 tokens first, and that design was good. It
|
||||
//! is to bound the thinking so the rest of the output budget survives
|
||||
//! for the answer.
|
||||
//!
|
||||
//! Two halves live here:
|
||||
//!
|
||||
//! - [`ReasoningBudget`] — how many reasoning tokens this request may
|
||||
//! spend, resolved from whichever control the caller's wire format
|
||||
//! gave them.
|
||||
//! - [`ReasoningGovernor`] — the per-request state that enforces it, by
|
||||
//! substituting the model's own `</think>` token when the budget runs
|
||||
//! out so generation transitions to answering instead of being cut off.
|
||||
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
/// Per-model mapping from an OpenAI effort level to a token budget.
|
||||
///
|
||||
/// OpenAI-shaped clients cannot send a token count — pi-ai's vocabulary
|
||||
/// is the ladder `off | minimal | low | medium | high | xhigh | max` and
|
||||
/// nothing else — so the numbers have to live on the server. Operator
|
||||
/// configurable, because the right values depend on the model and the
|
||||
/// hardware; a derived default would be a guess baked into the binary.
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct ReasoningBudgetLadder {
|
||||
pub minimal: usize,
|
||||
pub low: usize,
|
||||
pub medium: usize,
|
||||
pub high: usize,
|
||||
/// `xhigh` and `max` are only offered by a client when the model
|
||||
/// declares them; absent here means "not offered", and a request
|
||||
/// naming one falls back to `high` rather than being rejected.
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub xhigh: Option<usize>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub max: Option<usize>,
|
||||
}
|
||||
|
||||
impl Default for ReasoningBudgetLadder {
|
||||
/// Rungs sized against observed behaviour rather than round numbers:
|
||||
/// a long agentic design turn on this class of model runs to ~27k
|
||||
/// tokens, so `high` sits above that (bounded but rarely binding),
|
||||
/// `medium` at roughly half (enough to plan a whole implementation),
|
||||
/// and `low` at the scale of a focused single-file change.
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
minimal: 1_024,
|
||||
low: 4_096,
|
||||
medium: 12_288,
|
||||
high: 32_768,
|
||||
xhigh: None,
|
||||
max: None,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl ReasoningBudgetLadder {
|
||||
/// Resolve an effort level to a token budget.
|
||||
///
|
||||
/// `none`/`off` are handled before this — they disable thinking
|
||||
/// outright via the chat template — so they have no budget here.
|
||||
/// An unrecognised level resolves to `None` (unbounded) rather than
|
||||
/// to a guess: silently substituting a budget nobody asked for is
|
||||
/// the failure mode this whole area has been full of.
|
||||
pub fn for_effort(&self, effort: &str) -> Option<usize> {
|
||||
match effort.trim().to_ascii_lowercase().as_str() {
|
||||
"minimal" => Some(self.minimal),
|
||||
"low" => Some(self.low),
|
||||
"medium" => Some(self.medium),
|
||||
"high" => Some(self.high),
|
||||
"xhigh" => Some(self.xhigh.unwrap_or(self.high)),
|
||||
"max" => Some(self.max.or(self.xhigh).unwrap_or(self.high)),
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
|
||||
/// The mapping as a client should see it on `/v1/models` — the rungs
|
||||
/// this deployment actually offers, so a caller picking `low` knows
|
||||
/// what it bought instead of guessing (#274).
|
||||
pub fn advertised(&self) -> Vec<(String, usize)> {
|
||||
let mut out = vec![
|
||||
("minimal".to_string(), self.minimal),
|
||||
("low".to_string(), self.low),
|
||||
("medium".to_string(), self.medium),
|
||||
("high".to_string(), self.high),
|
||||
];
|
||||
if let Some(v) = self.xhigh {
|
||||
out.push(("xhigh".to_string(), v));
|
||||
}
|
||||
if let Some(v) = self.max {
|
||||
out.push(("max".to_string(), v));
|
||||
}
|
||||
out
|
||||
}
|
||||
}
|
||||
|
||||
/// Enforces a resolved budget over one request's token stream.
|
||||
///
|
||||
/// Fed every sampled token; returns the token that should actually be
|
||||
/// emitted. While the model is inside a think block and over budget, it
|
||||
/// returns the close marker instead, exactly once — after which the
|
||||
/// model is answering and the governor is inert.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct ReasoningGovernor {
|
||||
limit: usize,
|
||||
open_id: u32,
|
||||
close_id: u32,
|
||||
inside: bool,
|
||||
used: usize,
|
||||
forced: bool,
|
||||
}
|
||||
|
||||
impl ReasoningGovernor {
|
||||
/// `None` when the request named no budget, or the model has no
|
||||
/// reasoning markers to enforce one with — in both cases generation
|
||||
/// proceeds exactly as it did before this existed.
|
||||
pub fn new(
|
||||
limit: Option<usize>,
|
||||
markers: Option<(u32, u32)>,
|
||||
prompt_opened_reasoning: bool,
|
||||
) -> Option<Self> {
|
||||
let limit = limit?;
|
||||
let (open_id, close_id) = markers?;
|
||||
Some(Self {
|
||||
limit,
|
||||
open_id,
|
||||
close_id,
|
||||
inside: prompt_opened_reasoning,
|
||||
used: 0,
|
||||
forced: false,
|
||||
})
|
||||
}
|
||||
|
||||
/// Whether the forced close has already been spent.
|
||||
pub fn forced(&self) -> bool {
|
||||
self.forced
|
||||
}
|
||||
|
||||
/// Reasoning tokens counted so far.
|
||||
pub fn used(&self) -> usize {
|
||||
self.used
|
||||
}
|
||||
|
||||
/// Govern one sampled token, returning the token to emit.
|
||||
pub fn govern(&mut self, sampled: u32) -> u32 {
|
||||
if sampled == self.open_id {
|
||||
self.inside = true;
|
||||
return sampled;
|
||||
}
|
||||
if sampled == self.close_id {
|
||||
self.inside = false;
|
||||
return sampled;
|
||||
}
|
||||
if !self.inside {
|
||||
return sampled;
|
||||
}
|
||||
if self.used >= self.limit {
|
||||
// Budget spent: hand the model its own close marker so it
|
||||
// transitions to answering. Cutting the stream here instead
|
||||
// would reproduce the very failure this exists to prevent.
|
||||
self.inside = false;
|
||||
self.forced = true;
|
||||
return self.close_id;
|
||||
}
|
||||
self.used += 1;
|
||||
sampled
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
const OPEN: u32 = 100;
|
||||
const CLOSE: u32 = 101;
|
||||
|
||||
fn governor(limit: usize, opened: bool) -> ReasoningGovernor {
|
||||
ReasoningGovernor::new(Some(limit), Some((OPEN, CLOSE)), opened).expect("governor")
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_request_with_no_budget_is_ungoverned() {
|
||||
assert!(ReasoningGovernor::new(None, Some((OPEN, CLOSE)), true).is_none());
|
||||
}
|
||||
|
||||
/// A model with no detected markers cannot be governed — better
|
||||
/// unbounded than truncated at an arbitrary token.
|
||||
#[test]
|
||||
fn a_model_without_markers_is_ungoverned() {
|
||||
assert!(ReasoningGovernor::new(Some(10), None, true).is_none());
|
||||
}
|
||||
|
||||
/// The case that motivated this: the chat template left the think
|
||||
/// block open, so the model is reasoning from the first token.
|
||||
#[test]
|
||||
fn a_prompt_opened_block_is_bounded_and_force_closed() {
|
||||
let mut g = governor(3, true);
|
||||
assert_eq!(g.govern(7), 7);
|
||||
assert_eq!(g.govern(8), 8);
|
||||
assert_eq!(g.govern(9), 9);
|
||||
assert_eq!(g.govern(10), CLOSE, "the fourth token must close the block");
|
||||
assert!(g.forced());
|
||||
assert_eq!(g.used(), 3);
|
||||
// Answering now — the governor is inert.
|
||||
assert_eq!(g.govern(11), 11);
|
||||
assert_eq!(g.govern(12), 12);
|
||||
}
|
||||
|
||||
/// Content before the block opens is never counted against the
|
||||
/// reasoning budget.
|
||||
#[test]
|
||||
fn only_tokens_inside_the_block_are_counted() {
|
||||
let mut g = governor(2, false);
|
||||
assert_eq!(g.govern(5), 5);
|
||||
assert_eq!(g.govern(6), 6);
|
||||
assert_eq!(g.used(), 0);
|
||||
assert_eq!(g.govern(OPEN), OPEN);
|
||||
assert_eq!(g.govern(7), 7);
|
||||
assert_eq!(g.govern(8), 8);
|
||||
assert_eq!(g.govern(9), CLOSE);
|
||||
assert_eq!(g.used(), 2);
|
||||
}
|
||||
|
||||
/// A model that finishes thinking on its own is untouched — the
|
||||
/// budget is a ceiling, not a target.
|
||||
#[test]
|
||||
fn a_model_that_stops_on_its_own_is_not_forced() {
|
||||
let mut g = governor(100, true);
|
||||
for t in 20..30 {
|
||||
assert_eq!(g.govern(t), t);
|
||||
}
|
||||
assert_eq!(g.govern(CLOSE), CLOSE);
|
||||
assert!(!g.forced());
|
||||
for t in 40..50 {
|
||||
assert_eq!(g.govern(t), t);
|
||||
}
|
||||
assert!(!g.forced(), "content after the block must not be governed");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn the_ladder_resolves_the_levels_clients_can_send() {
|
||||
let l = ReasoningBudgetLadder::default();
|
||||
assert_eq!(l.for_effort("minimal"), Some(1_024));
|
||||
assert_eq!(l.for_effort("LOW"), Some(4_096));
|
||||
assert_eq!(l.for_effort(" medium "), Some(12_288));
|
||||
assert_eq!(l.for_effort("high"), Some(32_768));
|
||||
// Not offered by default → falls back to high rather than
|
||||
// rejecting a level the client was told it could send.
|
||||
assert_eq!(l.for_effort("xhigh"), Some(32_768));
|
||||
assert_eq!(l.for_effort("max"), Some(32_768));
|
||||
// Unknown → unbounded, never a substituted guess.
|
||||
assert_eq!(l.for_effort("enthusiastic"), None);
|
||||
assert_eq!(l.for_effort("none"), None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn the_advertised_ladder_lists_only_what_is_offered() {
|
||||
let l = ReasoningBudgetLadder::default();
|
||||
let names: Vec<_> = l.advertised().into_iter().map(|(k, _)| k).collect();
|
||||
assert_eq!(names, ["minimal", "low", "medium", "high"]);
|
||||
let l2 = ReasoningBudgetLadder {
|
||||
xhigh: Some(49_152),
|
||||
..ReasoningBudgetLadder::default()
|
||||
};
|
||||
assert!(
|
||||
l2.advertised()
|
||||
.iter()
|
||||
.any(|(k, v)| k == "xhigh" && *v == 49_152)
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -2445,7 +2445,7 @@ mod tests {
|
||||
}
|
||||
|
||||
#[cfg(not(feature = "cuda"))]
|
||||
fn vb_over(path: &std::path::Path) -> ShardedVarBuilder {
|
||||
fn vb_over(path: &std::path::Path) -> ShardedVarBuilder<'_> {
|
||||
// SAFETY: mmap of a file the test just wrote; nothing mutates it.
|
||||
unsafe {
|
||||
candle_nn::var_builder::ShardedSafeTensors::var_builder(
|
||||
|
||||
@@ -113,8 +113,30 @@ pub fn request_to_chat(req: ResponsesRequest) -> Result<ChatCompletionRequest, T
|
||||
});
|
||||
}
|
||||
ResponsesInput::Items(items) => {
|
||||
// A reasoning item describes the assistant turn that
|
||||
// follows it, so it is held until that turn arrives rather
|
||||
// than becoming a message of its own (#277).
|
||||
let mut pending_reasoning: Option<String> = None;
|
||||
for element in items {
|
||||
let msg = match element {
|
||||
ResponsesInputElement::Typed(ResponsesInputItem::Reasoning {
|
||||
content,
|
||||
summary,
|
||||
}) => {
|
||||
if let Some(text) = reasoning_item_text(&content, &summary) {
|
||||
// Two reasoning items with no assistant turn
|
||||
// between them: keep both, in order, rather
|
||||
// than letting the second overwrite the first.
|
||||
match &mut pending_reasoning {
|
||||
Some(existing) => {
|
||||
existing.push_str("\n\n");
|
||||
existing.push_str(&text);
|
||||
}
|
||||
slot => *slot = Some(text),
|
||||
}
|
||||
}
|
||||
None
|
||||
}
|
||||
ResponsesInputElement::Typed(item) => input_item_to_chat(item),
|
||||
// Bare `{role, content}` (OpenAI EasyInputMessage —
|
||||
// what litellm/agent-zero emit). `content: null`
|
||||
@@ -128,29 +150,86 @@ pub fn request_to_chat(req: ResponsesRequest) -> Result<ChatCompletionRequest, T
|
||||
extra: Value::Object(Default::default()),
|
||||
}),
|
||||
// Forward-compat: an item shape we don't model.
|
||||
// Dropped rather than rejected (see
|
||||
// `ResponsesInputElement::Other`).
|
||||
ResponsesInputElement::Other(_) => None,
|
||||
// Counted and logged rather than vanishing silently —
|
||||
// a shape we drop must be discoverable from the
|
||||
// outside (#277), which is how this class of defect
|
||||
// stayed invisible for so long.
|
||||
ResponsesInputElement::Other(other) => {
|
||||
// Resolved before the macro: `tracing`'s own
|
||||
// `Value` trait is in scope inside it, and would
|
||||
// shadow `serde_json::Value` here.
|
||||
let item_type = other
|
||||
.get("type")
|
||||
.and_then(|t| t.as_str())
|
||||
.unwrap_or("<untyped>")
|
||||
.to_string();
|
||||
tracing::warn!(
|
||||
%item_type,
|
||||
"responses: dropping an input item shape this build does not model"
|
||||
);
|
||||
None
|
||||
}
|
||||
};
|
||||
if let Some(msg) = msg {
|
||||
if let Some(mut msg) = msg {
|
||||
if msg.role == "assistant"
|
||||
&& let Some(text) = pending_reasoning.take()
|
||||
{
|
||||
attach_reasoning(&mut msg, text);
|
||||
}
|
||||
messages.push(msg);
|
||||
}
|
||||
}
|
||||
// Reasoning with no assistant turn after it — the exact
|
||||
// shape of a turn truncated mid-think and replayed with
|
||||
// "continue". It is the whole payload of that turn, so it
|
||||
// becomes an assistant turn of its own rather than being
|
||||
// discarded for lack of a message to ride on.
|
||||
if let Some(text) = pending_reasoning.take() {
|
||||
let mut msg = ChatMessage {
|
||||
role: "assistant".into(),
|
||||
content: MessageContent::Text(String::new()),
|
||||
extra: Value::Object(Default::default()),
|
||||
};
|
||||
attach_reasoning(&mut msg, text);
|
||||
messages.push(msg);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Forward tool definitions to the harness (#158). The chat path
|
||||
// reads `extra["tools"]` for both the Jinja prompt render and
|
||||
// argument-type coercion, so normalizing into the chat-wrapped
|
||||
// shape here is all the plumbing tools need.
|
||||
let mut extra = serde_json::Map::new();
|
||||
if let Some(tools) = req.extra.get("tools").and_then(Value::as_array) {
|
||||
let normalized: Vec<Value> = tools.iter().filter_map(normalize_tool).collect();
|
||||
if !normalized.is_empty() {
|
||||
extra.insert("tools".into(), Value::Array(normalized));
|
||||
// Carry the caller's extension fields across the hop (#277).
|
||||
//
|
||||
// This used to start from an empty map and insert exactly one key,
|
||||
// which meant every extension a caller sent to `/v1/responses` was
|
||||
// discarded — including `chat_template_kwargs`, the only route by
|
||||
// which `enable_thinking` can reach the template, which is why
|
||||
// `/no_think` worked on chat/completions and not here (#223).
|
||||
// Keys the chat path does not model are inert, so forwarding them
|
||||
// costs nothing and stops the next control needing its own plumbing.
|
||||
let mut extra = match req.extra {
|
||||
Value::Object(map) => map,
|
||||
_ => serde_json::Map::new(),
|
||||
};
|
||||
// Tool definitions are the exception: Responses flattens the
|
||||
// function fields and the chat path (#158) reads the nested shape
|
||||
// for both the Jinja render and argument coercion, so this key is
|
||||
// rewritten rather than forwarded.
|
||||
match extra.get("tools").and_then(Value::as_array) {
|
||||
Some(tools) => {
|
||||
let normalized: Vec<Value> = tools.iter().filter_map(normalize_tool).collect();
|
||||
if normalized.is_empty() {
|
||||
extra.remove("tools");
|
||||
} else {
|
||||
extra.insert("tools".into(), Value::Array(normalized));
|
||||
}
|
||||
}
|
||||
None => {
|
||||
extra.remove("tools");
|
||||
}
|
||||
}
|
||||
|
||||
// A caller that asked for no reasoning must get none (#223).
|
||||
apply_reasoning_effort(&mut extra);
|
||||
|
||||
Ok(ChatCompletionRequest {
|
||||
model: req.model,
|
||||
messages,
|
||||
@@ -243,13 +322,102 @@ fn input_item_to_chat(item: ResponsesInputItem) -> Option<ChatMessage> {
|
||||
extra: Value::Object(extra),
|
||||
})
|
||||
}
|
||||
// Reasoning items don't have a chat-completions equivalent
|
||||
// we can faithfully forward. Silently drop — the alternative
|
||||
// is rejecting a well-formed request, which is worse UX.
|
||||
// Handled before this function is reached: a reasoning item
|
||||
// belongs to the assistant turn that follows it, so the caller
|
||||
// holds it and attaches it there (#277).
|
||||
ResponsesInputItem::Reasoning { .. } => None,
|
||||
}
|
||||
}
|
||||
|
||||
/// Map the Responses reasoning control onto the template kwarg neuron
|
||||
/// already honours (#223).
|
||||
///
|
||||
/// `reasoning: {"effort": "none"}` is how a Responses client says "do
|
||||
/// not think" — it is what pi-ai emits for thinking-level *off*. The
|
||||
/// chat-completions path couples generation to surfacing via
|
||||
/// `default_enable_thinking`; this surface never did, so the switch had
|
||||
/// no route at all and a client asking for no reasoning got a full think
|
||||
/// block anyway. Observed live: a harness set thinking off and received
|
||||
/// 32,768 tokens of reasoning and no answer.
|
||||
///
|
||||
/// Only `none` and `off` map. `minimal` / `low` deliberately do not:
|
||||
/// they ask for *less* thinking, not none, and the template's switch is
|
||||
/// binary — silently turning "minimal" into "off" would be the same
|
||||
/// class of lie this whole area has been full of. They pass through and
|
||||
/// the model thinks as usual.
|
||||
///
|
||||
/// An explicit `chat_template_kwargs.enable_thinking` from the caller
|
||||
/// wins, matching `default_enable_thinking`'s precedence: the more
|
||||
/// specific control is the one the caller reached for.
|
||||
fn apply_reasoning_effort(extra: &mut serde_json::Map<String, Value>) {
|
||||
if extra
|
||||
.get("chat_template_kwargs")
|
||||
.and_then(|k| k.get("enable_thinking"))
|
||||
.is_some()
|
||||
{
|
||||
return;
|
||||
}
|
||||
let effort = extra
|
||||
.get("reasoning")
|
||||
.and_then(|r| r.get("effort"))
|
||||
.and_then(|e| e.as_str())
|
||||
.map(|e| e.trim().to_ascii_lowercase());
|
||||
if !matches!(effort.as_deref(), Some("none") | Some("off")) {
|
||||
return;
|
||||
}
|
||||
let kwargs = extra
|
||||
.entry("chat_template_kwargs")
|
||||
.or_insert_with(|| json!({}));
|
||||
if !kwargs.is_object() {
|
||||
*kwargs = json!({});
|
||||
}
|
||||
if let Some(kw) = kwargs.as_object_mut() {
|
||||
kw.insert("enable_thinking".into(), Value::Bool(false));
|
||||
}
|
||||
}
|
||||
|
||||
/// The text of a replayed reasoning item, under either spelling.
|
||||
///
|
||||
/// `summary` is what neuron emits (`summary_text` parts) and `content`
|
||||
/// is what OpenAI's o-series emits (`reasoning_text` parts); clients
|
||||
/// replay the completed item verbatim, so the field we read has to be
|
||||
/// the field the far end wrote. Parts are joined with a blank line,
|
||||
/// matching how pi-ai flattens the same item for display.
|
||||
///
|
||||
/// Returns `None` for an item with no usable text — an
|
||||
/// encrypted-content-only item, or an empty summary — so an assistant
|
||||
/// turn is not annotated with an empty think block.
|
||||
fn reasoning_item_text(content: &[Value], summary: &[Value]) -> Option<String> {
|
||||
fn join(parts: &[Value]) -> String {
|
||||
parts
|
||||
.iter()
|
||||
.filter_map(|p| p.get("text").and_then(Value::as_str))
|
||||
.filter(|t| !t.is_empty())
|
||||
.collect::<Vec<_>>()
|
||||
.join("\n\n")
|
||||
}
|
||||
let text = match join(summary) {
|
||||
s if !s.is_empty() => s,
|
||||
_ => join(content),
|
||||
};
|
||||
(!text.is_empty()).then_some(text)
|
||||
}
|
||||
|
||||
/// Hang a turn's reasoning on the assistant message it belongs to.
|
||||
///
|
||||
/// `reasoning_content` is the field HF chat templates read — Qwen3.8's
|
||||
/// renders it back inside `<think>` — and `render_chat_template`
|
||||
/// forwards message extras verbatim, so this is the whole of what the
|
||||
/// round-trip needs on our side.
|
||||
fn attach_reasoning(msg: &mut ChatMessage, text: String) {
|
||||
if !msg.extra.is_object() {
|
||||
msg.extra = Value::Object(Default::default());
|
||||
}
|
||||
if let Some(obj) = msg.extra.as_object_mut() {
|
||||
obj.insert("reasoning_content".into(), Value::String(text));
|
||||
}
|
||||
}
|
||||
|
||||
fn message_content_to_chat(content: ResponsesMessageContent) -> MessageContent {
|
||||
match content {
|
||||
ResponsesMessageContent::Text(s) => MessageContent::Text(s),
|
||||
@@ -1239,12 +1407,18 @@ mod tests {
|
||||
));
|
||||
}
|
||||
|
||||
/// An item carrying no readable text — encrypted-content-only, or
|
||||
/// an empty summary — must not manufacture an empty assistant turn
|
||||
/// with a blank think block.
|
||||
#[test]
|
||||
fn reasoning_items_are_silently_dropped() {
|
||||
fn a_reasoning_item_with_no_text_adds_no_turn() {
|
||||
let req = ResponsesRequest {
|
||||
model: "m".into(),
|
||||
input: typed_items(vec![
|
||||
ResponsesInputItem::Reasoning { content: vec![] },
|
||||
ResponsesInputItem::Reasoning {
|
||||
content: vec![],
|
||||
summary: vec![],
|
||||
},
|
||||
ResponsesInputItem::Message {
|
||||
role: "user".into(),
|
||||
content: ResponsesMessageContent::Text("hi".into()),
|
||||
@@ -1267,6 +1441,299 @@ mod tests {
|
||||
assert_eq!(chat.messages[0].role, "user");
|
||||
}
|
||||
|
||||
/// The shape neuron itself emits — `summary` with a `summary_text`
|
||||
/// part — replayed by a client that round-trips the completed item
|
||||
/// verbatim, which is what pi-ai/DSH does.
|
||||
#[test]
|
||||
fn replayed_reasoning_rides_on_the_assistant_turn_it_belongs_to() {
|
||||
let raw = r#"{
|
||||
"model": "Qwen/Qwen3.8-27B",
|
||||
"input": [
|
||||
{"type": "message", "role": "user", "content": "build it"},
|
||||
{"type": "reasoning", "id": "rs_1", "status": "completed",
|
||||
"summary": [{"type": "summary_text", "text": "the plan so far"}]},
|
||||
{"type": "message", "role": "assistant",
|
||||
"content": [{"type": "output_text", "text": "partial answer"}]},
|
||||
{"type": "message", "role": "user", "content": "continue"}
|
||||
]
|
||||
}"#;
|
||||
let req: ResponsesRequest = serde_json::from_str(raw).expect("parse");
|
||||
let chat = request_to_chat(req).expect("translate");
|
||||
assert_eq!(chat.messages.len(), 3);
|
||||
assert_eq!(chat.messages[1].role, "assistant");
|
||||
assert_eq!(
|
||||
chat.messages[1].extra.get("reasoning_content"),
|
||||
Some(&Value::String("the plan so far".into())),
|
||||
"the assistant turn must carry the thinking that produced it"
|
||||
);
|
||||
}
|
||||
|
||||
/// OpenAI's o-series spelling: the text lives in `content` as
|
||||
/// `reasoning_text` parts. Same item, different field.
|
||||
#[test]
|
||||
fn the_content_spelling_of_a_reasoning_item_is_read_too() {
|
||||
let raw = r#"{
|
||||
"model": "m",
|
||||
"input": [
|
||||
{"type": "reasoning", "id": "rs_1",
|
||||
"content": [{"type": "reasoning_text", "text": "step one"},
|
||||
{"type": "reasoning_text", "text": "step two"}]},
|
||||
{"type": "message", "role": "assistant", "content": "done"}
|
||||
]
|
||||
}"#;
|
||||
let req: ResponsesRequest = serde_json::from_str(raw).expect("parse");
|
||||
let chat = request_to_chat(req).expect("translate");
|
||||
assert_eq!(
|
||||
chat.messages[0].extra.get("reasoning_content"),
|
||||
Some(&Value::String("step one\n\nstep two".into()))
|
||||
);
|
||||
}
|
||||
|
||||
/// The case that motivated #277: a turn truncated mid-think, then
|
||||
/// `continue`. There is no assistant *message* to hang the
|
||||
/// reasoning on — the reasoning is the entire turn — so it must
|
||||
/// still reach the model, or the model restarts from nothing.
|
||||
#[test]
|
||||
fn a_truncated_turn_replayed_as_reasoning_alone_still_reaches_the_model() {
|
||||
let raw = r#"{
|
||||
"model": "Qwen/Qwen3.8-27B",
|
||||
"input": [
|
||||
{"type": "message", "role": "user", "content": "build a game"},
|
||||
{"type": "reasoning", "id": "rs_1", "status": "incomplete",
|
||||
"summary": [{"type": "summary_text", "text": "24k tokens of design"}]},
|
||||
{"type": "message", "role": "user", "content": "continue"}
|
||||
]
|
||||
}"#;
|
||||
let req: ResponsesRequest = serde_json::from_str(raw).expect("parse");
|
||||
let chat = request_to_chat(req).expect("translate");
|
||||
let assistant: Vec<_> = chat
|
||||
.messages
|
||||
.iter()
|
||||
.filter(|m| m.role == "assistant")
|
||||
.collect();
|
||||
assert_eq!(
|
||||
assistant.len(),
|
||||
1,
|
||||
"the truncated turn must survive as a turn"
|
||||
);
|
||||
assert_eq!(
|
||||
assistant[0].extra.get("reasoning_content"),
|
||||
Some(&Value::String("24k tokens of design".into()))
|
||||
);
|
||||
}
|
||||
|
||||
/// Consecutive reasoning items with no assistant turn between them
|
||||
/// accumulate rather than the later one erasing the earlier.
|
||||
#[test]
|
||||
fn consecutive_reasoning_items_accumulate() {
|
||||
let raw = r#"{
|
||||
"model": "m",
|
||||
"input": [
|
||||
{"type": "reasoning", "summary": [{"type": "summary_text", "text": "first"}]},
|
||||
{"type": "reasoning", "summary": [{"type": "summary_text", "text": "second"}]},
|
||||
{"type": "message", "role": "assistant", "content": "ok"}
|
||||
]
|
||||
}"#;
|
||||
let req: ResponsesRequest = serde_json::from_str(raw).expect("parse");
|
||||
let chat = request_to_chat(req).expect("translate");
|
||||
assert_eq!(
|
||||
chat.messages[0].extra.get("reasoning_content"),
|
||||
Some(&Value::String("first\n\nsecond".into()))
|
||||
);
|
||||
}
|
||||
|
||||
/// The control a Responses client actually reaches for (#223): pi-ai
|
||||
/// emits `reasoning: {"effort": "none"}` for thinking-level *off*.
|
||||
/// Before this it went nowhere, and a harness that asked for no
|
||||
/// thinking got 32,768 tokens of it and no answer.
|
||||
#[test]
|
||||
fn reasoning_effort_none_turns_thinking_off() {
|
||||
for effort in ["none", "off", "NONE", " off "] {
|
||||
let raw =
|
||||
format!(r#"{{"model":"m","input":"hi","reasoning":{{"effort":"{effort}"}}}}"#);
|
||||
let req: ResponsesRequest = serde_json::from_str(&raw).expect("parse");
|
||||
let chat = request_to_chat(req).expect("translate");
|
||||
assert_eq!(
|
||||
chat.extra
|
||||
.get("chat_template_kwargs")
|
||||
.and_then(|k| k.get("enable_thinking")),
|
||||
Some(&Value::Bool(false)),
|
||||
"effort {effort:?} must reach the template as enable_thinking=false"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/// `minimal` and `low` ask for *less* thinking, not none. The
|
||||
/// template switch is binary, so mapping them to off would be a
|
||||
/// silent substitution — they pass through untouched instead.
|
||||
#[test]
|
||||
fn a_reduced_effort_is_not_silently_treated_as_none() {
|
||||
for effort in ["minimal", "low", "medium", "high"] {
|
||||
let raw =
|
||||
format!(r#"{{"model":"m","input":"hi","reasoning":{{"effort":"{effort}"}}}}"#);
|
||||
let req: ResponsesRequest = serde_json::from_str(&raw).expect("parse");
|
||||
let chat = request_to_chat(req).expect("translate");
|
||||
assert!(
|
||||
chat.extra
|
||||
.get("chat_template_kwargs")
|
||||
.and_then(|k| k.get("enable_thinking"))
|
||||
.is_none(),
|
||||
"effort {effort:?} must not be rewritten to off"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/// The more specific control wins, matching `default_enable_thinking`.
|
||||
#[test]
|
||||
fn an_explicit_template_kwarg_beats_the_effort_mapping() {
|
||||
let raw = r#"{"model":"m","input":"hi",
|
||||
"reasoning":{"effort":"none"},
|
||||
"chat_template_kwargs":{"enable_thinking":true}}"#;
|
||||
let req: ResponsesRequest = serde_json::from_str(raw).expect("parse");
|
||||
let chat = request_to_chat(req).expect("translate");
|
||||
assert_eq!(
|
||||
chat.extra
|
||||
.get("chat_template_kwargs")
|
||||
.and_then(|k| k.get("enable_thinking")),
|
||||
Some(&Value::Bool(true))
|
||||
);
|
||||
}
|
||||
|
||||
/// A request with no reasoning control is untouched — the model
|
||||
/// thinks by default, which is the Responses-native expectation.
|
||||
#[test]
|
||||
fn no_reasoning_control_leaves_thinking_alone() {
|
||||
let raw = r#"{"model":"m","input":"hi"}"#;
|
||||
let req: ResponsesRequest = serde_json::from_str(raw).expect("parse");
|
||||
let chat = request_to_chat(req).expect("translate");
|
||||
assert!(chat.extra.get("chat_template_kwargs").is_none());
|
||||
}
|
||||
|
||||
/// End to end, on the rendered prompt: Qwen3's template emits an
|
||||
/// empty think block when `enable_thinking` is false, and opens one
|
||||
/// otherwise. This asserts the caller's control reaches that branch.
|
||||
#[test]
|
||||
fn reasoning_effort_none_reaches_the_template_branch() {
|
||||
const QWEN_THINK_SWITCH: &str = "{%- if enable_thinking is defined and enable_thinking is false -%}\
|
||||
<think>\n\n</think>\n\n\
|
||||
{%- else -%}\
|
||||
<think>\n\
|
||||
{%- endif -%}";
|
||||
let raw = r#"{"model":"m","input":"hi","reasoning":{"effort":"none"}}"#;
|
||||
let req: ResponsesRequest = serde_json::from_str(raw).expect("parse");
|
||||
let chat = request_to_chat(req).expect("translate");
|
||||
let kwargs = chat
|
||||
.extra
|
||||
.get("chat_template_kwargs")
|
||||
.cloned()
|
||||
.unwrap_or(Value::Null);
|
||||
let out = crate::harness::chat_template::render_chat_template(
|
||||
QWEN_THINK_SWITCH,
|
||||
&chat.messages,
|
||||
&Value::Null,
|
||||
&kwargs,
|
||||
)
|
||||
.expect("render");
|
||||
assert!(
|
||||
out.contains("<think>\n\n</think>"),
|
||||
"the closed think block is what tells the model not to reason; got {out:?}"
|
||||
);
|
||||
}
|
||||
|
||||
/// #223's root cause: `chat_template_kwargs` is how `enable_thinking`
|
||||
/// reaches the template, and the translator used to rebuild `extra`
|
||||
/// from an allowlist of one key, so it never arrived.
|
||||
#[test]
|
||||
fn extension_fields_survive_the_hop() {
|
||||
let raw = r#"{
|
||||
"model": "m",
|
||||
"input": "hello",
|
||||
"chat_template_kwargs": {"enable_thinking": false},
|
||||
"some_future_control": 7
|
||||
}"#;
|
||||
let req: ResponsesRequest = serde_json::from_str(raw).expect("parse");
|
||||
let chat = request_to_chat(req).expect("translate");
|
||||
assert_eq!(
|
||||
chat.extra
|
||||
.get("chat_template_kwargs")
|
||||
.and_then(|k| k.get("enable_thinking")),
|
||||
Some(&Value::Bool(false)),
|
||||
"the thinking switch must reach the chat path"
|
||||
);
|
||||
assert_eq!(chat.extra.get("some_future_control"), Some(&Value::from(7)));
|
||||
}
|
||||
|
||||
/// The end-to-end claim, asserted where it actually matters: on the
|
||||
/// rendered prompt. A translator test proves we set a field; this
|
||||
/// proves the model sees the thinking, through the same
|
||||
/// `reasoning_content` branch Qwen3.8's real template uses.
|
||||
///
|
||||
/// The same shape as the #179 system-prompt tests, and for the same
|
||||
/// reason — the contract is what reaches the model, not what the
|
||||
/// intermediate struct holds.
|
||||
#[test]
|
||||
fn replayed_reasoning_reaches_the_rendered_prompt() {
|
||||
// The assistant branch of Qwen3.8's chat_template, reduced to
|
||||
// the part under test.
|
||||
const QWEN_REASONING_LIKE: &str = "{%- for message in messages -%}\
|
||||
{%- if message.role == 'assistant' -%}\
|
||||
<|im_start|>assistant\n<think>\n{{ message.reasoning_content }}\n</think>\n\n{{ message.content }}<|im_end|>\n\
|
||||
{%- else -%}\
|
||||
<|im_start|>{{ message.role }}\n{{ message.content }}<|im_end|>\n\
|
||||
{%- endif -%}\
|
||||
{%- endfor -%}";
|
||||
|
||||
let raw = r#"{
|
||||
"model": "Qwen/Qwen3.8-27B",
|
||||
"input": [
|
||||
{"type": "message", "role": "user", "content": "build a game"},
|
||||
{"type": "reasoning", "id": "rs_1",
|
||||
"summary": [{"type": "summary_text", "text": "the design so far"}]},
|
||||
{"type": "message", "role": "user", "content": "continue"}
|
||||
]
|
||||
}"#;
|
||||
let req: ResponsesRequest = serde_json::from_str(raw).expect("parse");
|
||||
let chat = request_to_chat(req).expect("translate");
|
||||
let prompt = crate::harness::chat_template::render_chat_template(
|
||||
QWEN_REASONING_LIKE,
|
||||
&chat.messages,
|
||||
&Value::Null,
|
||||
&Value::Null,
|
||||
)
|
||||
.expect("render");
|
||||
assert!(
|
||||
prompt.contains("<think>\nthe design so far\n</think>"),
|
||||
"the replayed reasoning must reach the prompt; got:\n{prompt}"
|
||||
);
|
||||
}
|
||||
|
||||
/// Forwarding `extra` wholesale must not smuggle the Responses-shaped
|
||||
/// tool array through — the chat path reads the nested spelling.
|
||||
#[test]
|
||||
fn tools_are_still_rewritten_not_forwarded_raw() {
|
||||
let raw = r#"{
|
||||
"model": "m",
|
||||
"input": "hi",
|
||||
"tools": [{"type": "function", "name": "run_code",
|
||||
"parameters": {"type": "object"}}]
|
||||
}"#;
|
||||
let req: ResponsesRequest = serde_json::from_str(raw).expect("parse");
|
||||
let chat = request_to_chat(req).expect("translate");
|
||||
let tools = chat
|
||||
.extra
|
||||
.get("tools")
|
||||
.and_then(Value::as_array)
|
||||
.expect("tools");
|
||||
assert_eq!(
|
||||
tools[0]["function"]["name"],
|
||||
Value::String("run_code".into())
|
||||
);
|
||||
assert!(
|
||||
tools[0].get("name").is_none(),
|
||||
"the flat spelling must not survive"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn bare_easy_messages_translate_like_typed_messages() {
|
||||
// The agent-zero / litellm shape: bare `{role, content}` items
|
||||
|
||||
@@ -60,7 +60,9 @@ fn read_f32(path: &Path) -> Vec<f32> {
|
||||
let bytes = std::fs::read(path).unwrap_or_else(|e| panic!("read {path:?}: {e}"));
|
||||
assert!(bytes.len().is_multiple_of(4), "truncated f32 file {path:?}");
|
||||
bytes
|
||||
.chunks_exact(4)
|
||||
.as_chunks::<4>()
|
||||
.0
|
||||
.iter()
|
||||
.map(|b| f32::from_le_bytes([b[0], b[1], b[2], b[3]]))
|
||||
.collect()
|
||||
}
|
||||
|
||||
@@ -37,7 +37,9 @@ fn fixture_dir() -> PathBuf {
|
||||
fn read_f32(path: &Path) -> Vec<f32> {
|
||||
let bytes = std::fs::read(path).unwrap_or_else(|e| panic!("read {path:?}: {e}"));
|
||||
bytes
|
||||
.chunks_exact(4)
|
||||
.as_chunks::<4>()
|
||||
.0
|
||||
.iter()
|
||||
.map(|c| f32::from_le_bytes([c[0], c[1], c[2], c[3]]))
|
||||
.collect()
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user