Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
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