Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -45,3 +45,6 @@ uuid = "=1.18.1"

[workspace.lints.rust]
missing_docs = "deny"

[workspace.lints.clippy]
cognitive_complexity = "deny"
4 changes: 4 additions & 0 deletions clippy.toml
Original file line number Diff line number Diff line change
@@ -0,0 +1,4 @@
# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0

cognitive-complexity-threshold = 18
152 changes: 82 additions & 70 deletions crates/adaptive/src/plugin_component.rs
Original file line number Diff line number Diff line change
Expand Up @@ -215,37 +215,7 @@ fn validate_adaptive_plugin_config_with_policy(
);
}

if let Some(state_json) = plugin_config.get("state").and_then(Json::as_object) {
validate_unknown_fields(
&mut diagnostics,
&config.policy,
Some("state".to_string()),
state_json,
&["backend"],
);
if let Some(backend_json) = state_json.get("backend").and_then(Json::as_object) {
validate_unknown_fields(
&mut diagnostics,
&config.policy,
Some("backend".to_string()),
backend_json,
&["kind", "config"],
);
let backend_kind = backend_json
.get("kind")
.and_then(Json::as_str)
.unwrap_or_default();
if let Some(backend_config_json) = backend_json.get("config").and_then(Json::as_object)
{
validate_backend_config_fields(
&mut diagnostics,
&config.policy,
backend_kind,
backend_config_json,
);
}
}
}
validate_adaptive_state_section(&mut diagnostics, &config.policy, plugin_config);

if let Some(telemetry_json) = plugin_config.get("telemetry").and_then(Json::as_object) {
validate_unknown_fields(
Expand Down Expand Up @@ -303,52 +273,94 @@ fn validate_adaptive_plugin_config_with_policy(
);
}

if let Some(response_cache_json) = plugin_config
validate_response_cache_section(&mut diagnostics, &config.policy, plugin_config);

diagnostics.extend(AdaptiveRuntime::validate_config(&config).diagnostics);
diagnostics
}

fn validate_adaptive_state_section(
diagnostics: &mut Vec<ConfigDiagnostic>,
policy: &ConfigPolicy,
plugin_config: &Map<String, Json>,
) {
let Some(state_json) = plugin_config.get("state").and_then(Json::as_object) else {
return;
};
validate_unknown_fields(
diagnostics,
policy,
Some("state".to_string()),
state_json,
&["backend"],
);
let Some(backend_json) = state_json.get("backend").and_then(Json::as_object) else {
return;
};
validate_unknown_fields(
diagnostics,
policy,
Some("backend".to_string()),
backend_json,
&["kind", "config"],
);
let backend_kind = backend_json
.get("kind")
.and_then(Json::as_str)
.unwrap_or_default();
if let Some(backend_config_json) = backend_json.get("config").and_then(Json::as_object) {
validate_backend_config_fields(diagnostics, policy, backend_kind, backend_config_json);
}
}

fn validate_response_cache_section(
diagnostics: &mut Vec<ConfigDiagnostic>,
policy: &ConfigPolicy,
plugin_config: &Map<String, Json>,
) {
let Some(response_cache_json) = plugin_config
.get("response_cache")
.and_then(Json::as_object)
{
else {
return;
};
validate_unknown_fields(
diagnostics,
policy,
Some("response_cache".to_string()),
response_cache_json,
&[
"ttl_seconds",
"namespace",
"priority",
"bypass_rate",
"cache_nondeterministic",
"key_strategy",
"header_allowlist",
"backend",
],
);
if let Some(backend_json) = response_cache_json.get("backend").and_then(Json::as_object) {
validate_unknown_fields(
&mut diagnostics,
&config.policy,
Some("response_cache".to_string()),
response_cache_json,
&[
"ttl_seconds",
"namespace",
"priority",
"bypass_rate",
"cache_nondeterministic",
"key_strategy",
"header_allowlist",
"backend",
],
diagnostics,
policy,
Some("response_cache.backend".to_string()),
backend_json,
&["kind", "config"],
);
if let Some(backend_json) = response_cache_json.get("backend").and_then(Json::as_object) {
validate_unknown_fields(
&mut diagnostics,
&config.policy,
Some("response_cache.backend".to_string()),
backend_json,
&["kind", "config"],
let backend_kind = backend_json
.get("kind")
.and_then(Json::as_str)
.unwrap_or("in_memory");
if let Some(backend_config_json) = backend_json.get("config").and_then(Json::as_object) {
validate_response_cache_backend_config_fields(
diagnostics,
policy,
backend_kind,
backend_config_json,
);
let backend_kind = backend_json
.get("kind")
.and_then(Json::as_str)
.unwrap_or("in_memory");
if let Some(backend_config_json) = backend_json.get("config").and_then(Json::as_object)
{
validate_response_cache_backend_config_fields(
&mut diagnostics,
&config.policy,
backend_kind,
backend_config_json,
);
}
}
}

diagnostics.extend(AdaptiveRuntime::validate_config(&config).diagnostics);
diagnostics
}

fn validate_response_cache_backend_config_fields(
Expand Down
82 changes: 40 additions & 42 deletions crates/adaptive/src/response_cache/key.rs
Original file line number Diff line number Diff line change
Expand Up @@ -56,48 +56,8 @@ pub fn build_cache_key(
request: &LlmRequest,
config: &ResponseCacheConfig,
) -> KeyOutcome {
// Unparseable bodies arrive as null; they would all share one key.
if request.content.is_null() {
return KeyOutcome::Bypass("unparseable_body");
}
// Cacheability gates run on the RAW request, so they are correct regardless
// of which codec (if any) decodes the body — a chat codec may park `store`
// in `extra` rather than the typed field, so we must not rely on the decode.
if let Some(object) = request.content.as_object() {
// Any present, non-`false` `store` opts into server-side persistence —
// bypass even a malformed non-boolean rather than risk caching a stateful
// call (whose result is otherwise keyed with `store` stripped).
if object
.get("store")
.is_some_and(|value| !matches!(value, Json::Bool(false) | Json::Null))
{
return KeyOutcome::Bypass("stateful_store");
}
if object.contains_key("previous_response_id") {
return KeyOutcome::Bypass("stateful_previous_response_id");
}
// Server-side conversation state the key cannot see.
if object.contains_key("conversation") || object.contains_key("container") {
return KeyOutcome::Bypass("stateful_conversation");
}
// Responses persists by default; only an explicit opt-out is stateless.
// A `prompt` object is the Responses prompt-template reference; a bare
// string `prompt` is a completions body with no server-side state.
if (object.contains_key("input")
|| object.contains_key("instructions")
|| object.get("prompt").is_some_and(Json::is_object))
&& !object
.get("store")
.is_some_and(|store| store == &Json::Bool(false))
{
return KeyOutcome::Bypass("stateful_store");
}
}
// Toggle off = explicit temperature 0 only; absent defaults to sampling.
if !config.cache_nondeterministic
&& request_temperature(&request.content).is_none_or(|temperature| temperature > 0.0)
{
return KeyOutcome::Bypass("nondeterministic_temperature");
if let Some(reason) = cache_bypass_reason(request, config) {
return KeyOutcome::Bypass(reason);
}

// Body to fingerprint: the decoded/normalized form when a surface resolves
Expand Down Expand Up @@ -136,6 +96,44 @@ pub fn build_cache_key(
}
}

fn cache_bypass_reason(request: &LlmRequest, config: &ResponseCacheConfig) -> Option<&'static str> {
if request.content.is_null() {
return Some("unparseable_body");
}
if let Some(reason) = request
.content
.as_object()
.and_then(stateful_request_bypass_reason)
{
return Some(reason);
}
(!config.cache_nondeterministic
&& request_temperature(&request.content).is_none_or(|temperature| temperature > 0.0))
.then_some("nondeterministic_temperature")
}

fn stateful_request_bypass_reason(object: &Map<String, Json>) -> Option<&'static str> {
if object
.get("store")
.is_some_and(|value| !matches!(value, Json::Bool(false) | Json::Null))
{
return Some("stateful_store");
}
if object.contains_key("previous_response_id") {
return Some("stateful_previous_response_id");
}
if object.contains_key("conversation") || object.contains_key("container") {
return Some("stateful_conversation");
}
let responses_surface = object.contains_key("input")
|| object.contains_key("instructions")
|| object.get("prompt").is_some_and(Json::is_object);
let explicitly_stateless = object
.get("store")
.is_some_and(|store| store == &Json::Bool(false));
(responses_surface && !explicitly_stateless).then_some("stateful_store")
}

/// Preserves which OpenAI Chat token-cap field the caller sent.
///
/// The Chat codec normalizes both spellings into `GenerationParams.max_tokens`,
Expand Down
86 changes: 48 additions & 38 deletions crates/adaptive/src/response_cache/replay.rs
Original file line number Diff line number Diff line change
Expand Up @@ -192,44 +192,7 @@ fn synthesize_chat_chunks(aggregate: &Json) -> Vec<Json> {
.cloned()
.unwrap_or_default();
for (position, choice) in choices.iter().enumerate() {
let index = choice
.get("index")
.and_then(Json::as_u64)
.unwrap_or(position as u64);
let message = choice.get("message").cloned().unwrap_or(json!({}));
if let Some(role) = message.get("role") {
chunks.push(base(
json!([{"index": index, "delta": {"role": role}, "finish_reason": null}]),
));
}
if let Some(content) = message.get("content").and_then(Json::as_str)
&& !content.is_empty()
{
chunks.push(base(
json!([{"index": index, "delta": {"content": content}, "finish_reason": null}]),
));
}
if let Some(tool_calls) = message.get("tool_calls").and_then(Json::as_array) {
let deltas: Vec<Json> = tool_calls
.iter()
.enumerate()
.map(|(call_index, call)| {
let mut delta = call.clone();
if let Some(map) = delta.as_object_mut() {
map.entry("index".to_string())
.or_insert(json!(call_index as u64));
}
delta
})
.collect();
chunks.push(base(
json!([{"index": index, "delta": {"tool_calls": deltas}, "finish_reason": null}]),
));
}
let finish = choice.get("finish_reason").cloned().unwrap_or(Json::Null);
chunks.push(base(
json!([{"index": index, "delta": {}, "finish_reason": finish}]),
));
chunks.extend(synthesize_chat_choice_chunks(&base, position, choice));
}
if let Some(usage) = aggregate.get("usage") {
let mut usage_chunk = base(json!([]));
Expand All @@ -241,6 +204,53 @@ fn synthesize_chat_chunks(aggregate: &Json) -> Vec<Json> {
chunks
}

fn synthesize_chat_choice_chunks(
base: &impl Fn(Json) -> Json,
position: usize,
choice: &Json,
) -> Vec<Json> {
let index = choice
.get("index")
.and_then(Json::as_u64)
.unwrap_or(position as u64);
let message = choice.get("message").cloned().unwrap_or(json!({}));
let mut chunks = Vec::new();
if let Some(role) = message.get("role") {
chunks.push(base(
json!([{"index": index, "delta": {"role": role}, "finish_reason": null}]),
));
}
if let Some(content) = message.get("content").and_then(Json::as_str)
&& !content.is_empty()
{
chunks.push(base(
json!([{"index": index, "delta": {"content": content}, "finish_reason": null}]),
));
}
if let Some(tool_calls) = message.get("tool_calls").and_then(Json::as_array) {
let deltas = tool_calls
.iter()
.enumerate()
.map(|(call_index, call)| {
let mut delta = call.clone();
if let Some(map) = delta.as_object_mut() {
map.entry("index".to_string())
.or_insert(json!(call_index as u64));
}
delta
})
.collect::<Vec<_>>();
chunks.push(base(
json!([{"index": index, "delta": {"tool_calls": deltas}, "finish_reason": null}]),
));
}
let finish = choice.get("finish_reason").cloned().unwrap_or(Json::Null);
chunks.push(base(
json!([{"index": index, "delta": {}, "finish_reason": finish}]),
));
chunks
}

/// OpenAI Responses: a `response.created` snapshot, one `response.output_item.done`
/// per output item, and a `response.completed` carrying the full stored aggregate
/// (the collector keeps the last snapshot wholesale, so reassembly is exact).
Expand Down
Loading
Loading