diff --git a/crates/adaptive/src/acg/ir_builder.rs b/crates/adaptive/src/acg/ir_builder.rs index a40702d63..ce03c97bf 100644 --- a/crates/adaptive/src/acg/ir_builder.rs +++ b/crates/adaptive/src/acg/ir_builder.rs @@ -35,19 +35,35 @@ use crate::acg::prompt_ir::{ pub fn build_prompt_ir(request: &AnnotatedLlmRequest) -> Result { let mut blocks: Vec = Vec::new(); let mut sequence_index: u32 = 0; - let mut inserted_tool_blocks = false; + let mut inserted_scaffold_blocks = false; + let structured_output = request + .extra + .get("response_format") + .filter(|value| !value.is_null()) + .map(canonicalize_value) + .transpose()?; for message in &request.messages { - if should_insert_tool_blocks_before_message(inserted_tool_blocks, request, message) { + if !inserted_scaffold_blocks && !matches!(message, Message::System { .. }) { append_tool_schema_blocks(&mut blocks, &mut sequence_index, request.tools.as_deref())?; - inserted_tool_blocks = true; + append_structured_output_block( + &mut blocks, + &mut sequence_index, + structured_output.as_deref(), + ); + inserted_scaffold_blocks = true; } append_message_blocks(&mut blocks, &mut sequence_index, message)?; } - if !inserted_tool_blocks { + if !inserted_scaffold_blocks { append_tool_schema_blocks(&mut blocks, &mut sequence_index, request.tools.as_deref())?; + append_structured_output_block( + &mut blocks, + &mut sequence_index, + structured_output.as_deref(), + ); } let tool_schema_hashes = match &request.tools { @@ -60,20 +76,12 @@ pub fn build_prompt_ir(request: &AnnotatedLlmRequest) -> Result { ir_id: Uuid::new_v4(), blocks, tool_schema_hashes, - structured_output_schema_id: None, + structured_output_schema_id: structured_output.as_deref().map(sha256_hex), source_request_hash, created_at: Utc::now(), }) } -fn should_insert_tool_blocks_before_message( - inserted_tool_blocks: bool, - request: &AnnotatedLlmRequest, - message: &Message, -) -> bool { - !inserted_tool_blocks && !matches!(message, Message::System { .. }) && request.tools.is_some() -} - fn append_message_blocks( blocks: &mut Vec, sequence_index: &mut u32, @@ -252,6 +260,29 @@ fn append_tool_schema_blocks( Ok(()) } +fn append_structured_output_block( + blocks: &mut Vec, + seq: &mut u32, + response_format: Option<&str>, +) { + let Some(content) = response_format else { + return; + }; + + let index = *seq; + *seq += 1; + blocks.push(PromptBlock { + span_id: generate_span_id(PromptRole::System, index, Some("structured-output")), + sequence_index: index, + role: PromptRole::System, + content: content.to_string(), + content_type: BlockContentType::StructuredOutput, + provenance: ProvenanceLabel::System, + sensitivity: SensitivityLabel::default(), + token_metadata: None, + }); +} + fn build_tool_schema_block(seq: &mut u32, tool: &ToolDefinition) -> Result { let tool_value = serde_json::to_value(tool)?; let canonical = canonicalize_value(&tool_value)?; diff --git a/crates/adaptive/src/acg/stability.rs b/crates/adaptive/src/acg/stability.rs index 6141ffa54..de8afc7ea 100644 --- a/crates/adaptive/src/acg/stability.rs +++ b/crates/adaptive/src/acg/stability.rs @@ -39,6 +39,11 @@ pub struct StabilityAnalysisResult { pub scores: Vec, /// Number of leading blocks that were classified as stable. pub stable_prefix_length: usize, + /// Fingerprint of the dominant stable prefix. Generic analysis hashes the + /// prompt IR; ACG profile persistence additionally binds it to the learning + /// key and, beyond the leading scaffold, the full source request. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub stable_prefix_fingerprint: Option, /// Total number of observations included in the analysis. pub total_observations: u32, } @@ -69,6 +74,7 @@ pub fn analyze_stability( return StabilityAnalysisResult { scores: Vec::new(), stable_prefix_length: 0, + stable_prefix_fingerprint: None, total_observations: 0, }; } @@ -89,14 +95,109 @@ pub fn analyze_stability( let scores: Vec = indexed_scores.into_iter().map(|(_, score)| score).collect(); let stable_prefix_length = find_stable_prefix_length(&scores); + let stable_prefix_fingerprint = observations + .iter() + .filter_map(|observation| prompt_prefix_fingerprint(observation, stable_prefix_length)) + .fold(HashMap::new(), |mut counts, fingerprint| { + *counts.entry(fingerprint).or_insert(0_u32) += 1; + counts + }) + .into_iter() + .max_by(|(left_hash, left_count), (right_hash, right_count)| { + left_count + .cmp(right_count) + .then_with(|| right_hash.cmp(left_hash)) + }) + .map(|(fingerprint, _)| fingerprint); StabilityAnalysisResult { scores, stable_prefix_length, + stable_prefix_fingerprint, total_observations, } } +pub(crate) fn prompt_prefix_fingerprint( + observation: &PromptIR, + prefix_length: usize, +) -> Option { + if prefix_length == 0 || observation.blocks.len() < prefix_length { + return None; + } + + let prefix = observation + .blocks + .iter() + .take(prefix_length) + .map(|block| { + serde_json::to_string(&( + &block.span_id, + block.role, + block.content_type, + &block.content, + )) + }) + .collect::, _>>() + .ok()? + .join("\n"); + Some(sha256_hex(&prefix)) +} + +pub(crate) fn profile_prefix_fingerprint( + observation: &PromptIR, + prefix_length: usize, + learning_key: &str, +) -> Option { + let prefix_fingerprint = prompt_prefix_fingerprint(observation, prefix_length)?; + let scaffold_length = observation + .blocks + .iter() + .take_while(|block| { + block.role == crate::acg::prompt_ir::PromptRole::System + || matches!( + block.content_type, + crate::acg::prompt_ir::BlockContentType::ToolSchema + | crate::acg::prompt_ir::BlockContentType::StructuredOutput + ) + }) + .count(); + // Conservatively bind deeper prefixes to the complete request because the + // normalized IR does not retain a lossless provider-prefix representation. + let request_fingerprint = if prefix_length > scaffold_length { + observation.source_request_hash.as_deref()? + } else { + "stable-scaffold" + }; + + Some(sha256_hex( + &[learning_key, &prefix_fingerprint, request_fingerprint].join("\n"), + )) +} + +pub(crate) fn dominant_profile_prefix_fingerprint( + observations: &[PromptIR], + prefix_length: usize, + learning_key: &str, +) -> Option { + observations + .iter() + .filter_map(|observation| { + profile_prefix_fingerprint(observation, prefix_length, learning_key) + }) + .fold(HashMap::new(), |mut counts, fingerprint| { + *counts.entry(fingerprint).or_insert(0_u32) += 1; + counts + }) + .into_iter() + .max_by(|(left_hash, left_count), (right_hash, right_count)| { + left_count + .cmp(right_count) + .then_with(|| right_hash.cmp(left_hash)) + }) + .map(|(fingerprint, _)| fingerprint) +} + fn record_observation(observation: &PromptIR, span_map: &mut HashMap) { let mut seen_in_observation: HashSet = HashSet::new(); diff --git a/crates/adaptive/src/acg_component.rs b/crates/adaptive/src/acg_component.rs index 443e376ff..6f51c13e5 100644 --- a/crates/adaptive/src/acg_component.rs +++ b/crates/adaptive/src/acg_component.rs @@ -133,6 +133,34 @@ fn build_intent_bundle( return None; } + let learning_key = derive_acg_learning_key(agent_id, annotated_request); + let live_fingerprint = crate::acg::stability::profile_prefix_fingerprint( + prompt_ir, + stability.stable_prefix_length, + &learning_key, + ); + let Some(stored_fingerprint) = stability.stable_prefix_fingerprint.as_ref() else { + acg_debug::emit( + "build_intent_bundle_skipped", + json!({"reason": "stable_prefix_fingerprint_missing"}), + ); + return None; + }; + let Some(live_fingerprint) = live_fingerprint.as_ref() else { + acg_debug::emit( + "build_intent_bundle_skipped", + json!({"reason": "stable_prefix_fingerprint_unavailable"}), + ); + return None; + }; + if stored_fingerprint != live_fingerprint { + acg_debug::emit( + "build_intent_bundle_skipped", + json!({"reason": "stable_prefix_fingerprint_mismatch"}), + ); + return None; + } + let toolset_hash = annotated_request .tools .as_ref() diff --git a/crates/adaptive/src/acg_learner.rs b/crates/adaptive/src/acg_learner.rs index 5c0a31ae3..9232966a1 100644 --- a/crates/adaptive/src/acg_learner.rs +++ b/crates/adaptive/src/acg_learner.rs @@ -105,7 +105,13 @@ impl Learner for AcgLearner { .store_observations(&profile_key, &observations_vec) .await?; - let stability_result = analyze_stability(&observations_vec, &self.thresholds); + let mut stability_result = analyze_stability(&observations_vec, &self.thresholds); + stability_result.stable_prefix_fingerprint = + crate::acg::stability::dominant_profile_prefix_fingerprint( + &observations_vec, + stability_result.stable_prefix_length, + &profile_key, + ); backend .store_stability(&profile_key, &stability_result) .await?; diff --git a/crates/adaptive/src/acg_profile.rs b/crates/adaptive/src/acg_profile.rs index 01e5c3b29..65fcf476f 100644 --- a/crates/adaptive/src/acg_profile.rs +++ b/crates/adaptive/src/acg_profile.rs @@ -15,6 +15,8 @@ struct AcgKeyParts<'a> { model: &'a str, system_hash: String, tool_hash: String, + response_format_hash: Option, + has_stable_scaffold: bool, } /// Derive the stable ACG learning key used to bucket observations and hot-cache state. @@ -28,12 +30,20 @@ pub(crate) fn derive_acg_learning_key( annotated_request: &AnnotatedLlmRequest, ) -> String { let parts = derive_key_parts(annotated_request); - let seed_fingerprint = learning_seed_fingerprint(annotated_request); - let seed_hash = short_hash(&seed_fingerprint); - format!( + let seed_fingerprint = + (!parts.has_stable_scaffold).then(|| learning_seed_fingerprint(annotated_request)); + let seed_hash = seed_fingerprint + .as_deref() + .map(short_hash) + .unwrap_or("stable-scaffold"); + let key = format!( "{agent_id}::model={}::seed={seed_hash}::system={}::tools={}", parts.model, parts.system_hash, parts.tool_hash - ) + ); + parts + .response_format_hash + .map(|hash| format!("{key}::response_format={hash}")) + .unwrap_or(key) } /// Derive the exact ACG profile key used for diagnostics and debug output. @@ -58,18 +68,38 @@ pub(crate) fn derive_acg_profile_key( .join("."); format!( "{agent_id}::model={}::roles={role_signature}::system={}::anchor={}::tools={}", - parts.model, parts.system_hash, anchor_hash, parts.tool_hash + parts.model, + short_hash(&parts.system_hash), + anchor_hash, + short_hash(&parts.tool_hash) ) } fn derive_key_parts(annotated_request: &AnnotatedLlmRequest) -> AcgKeyParts<'_> { let system_fingerprint = system_prompt_fingerprint(annotated_request); let tool_fingerprint = tool_schema_fingerprint(annotated_request.tools.as_deref()); + let response_format_fingerprint = response_format_fingerprint(annotated_request); + let has_stable_scaffold = system_fingerprint != "no-system" + || annotated_request + .tools + .as_ref() + .is_some_and(|tools| !tools.is_empty()) + || response_format_fingerprint.is_some(); AcgKeyParts { model: annotated_request.model.as_deref().unwrap_or("unknown"), - system_hash: short_hash(&system_fingerprint).to_string(), - tool_hash: short_hash(&tool_fingerprint).to_string(), + system_hash: if has_stable_scaffold { + system_fingerprint + } else { + short_hash(&system_fingerprint).to_string() + }, + tool_hash: if has_stable_scaffold { + tool_fingerprint + } else { + short_hash(&tool_fingerprint).to_string() + }, + response_format_hash: response_format_fingerprint, + has_stable_scaffold, } } @@ -183,6 +213,15 @@ fn tool_schema_fingerprint(tools: Option<&[ToolDefinition]>) -> String { } } +fn response_format_fingerprint(annotated_request: &AnnotatedLlmRequest) -> Option { + annotated_request + .extra + .get("response_format") + .filter(|value| !value.is_null()) + .and_then(|value| canonicalize_value(value).ok()) + .map(|value| sha256_hex(&value)) +} + fn extract_text(content: &MessageContent) -> String { match content { MessageContent::Text(text) => text.clone(), diff --git a/crates/adaptive/tests/integration/redis_tests.rs b/crates/adaptive/tests/integration/redis_tests.rs index 9cd730ae6..27792a5fc 100644 --- a/crates/adaptive/tests/integration/redis_tests.rs +++ b/crates/adaptive/tests/integration/redis_tests.rs @@ -501,3 +501,54 @@ async fn redis_integration_persists_runtime_seed_entries_and_manifest_cleanup() "adaptive manifest should not depend directly on the compatibility shim" ); } + +#[tokio::test] +async fn redis_integration_persists_scaffold_profile_across_backend_restart() { + let Some((backend, prefix)) = get_test_redis_with_prefix().await else { + return; + }; + let agent_id = "agent-redis-scaffold-restart"; + let learner = AcgLearner::new(agent_id, 8, StabilityThresholds::default()); + let first = sample_annotated_request("claude-3-5-sonnet"); + let mut second = first.clone(); + second.messages[1] = Message::User { + content: MessageContent::Text("Plan a different task".to_string()), + name: None, + }; + let hot_cache = empty_hot_cache(); + + for request in [first, second] { + learner + .process_run( + &sample_run_with_request(agent_id, request), + &backend, + &hot_cache, + ) + .await + .unwrap(); + } + let learning_key = { + let guard = hot_cache.read().unwrap(); + assert_eq!(guard.acg_profiles.len(), 1); + guard.acg_profiles.keys().next().unwrap().clone() + }; + drop(backend); + + let restarted = RedisBackend::new("redis://127.0.0.1/", prefix) + .await + .unwrap(); + let observations = restarted + .load_observations(&learning_key) + .await + .unwrap() + .unwrap(); + let stability = restarted + .load_stability(&learning_key) + .await + .unwrap() + .unwrap(); + + assert_eq!(observations.len(), 2); + assert_eq!(stability.total_observations, 2); + assert!(stability.stable_prefix_fingerprint.is_some()); +} diff --git a/crates/adaptive/tests/integration/runtime_integration_tests.rs b/crates/adaptive/tests/integration/runtime_integration_tests.rs index 16da0444f..5c80685a2 100644 --- a/crates/adaptive/tests/integration/runtime_integration_tests.rs +++ b/crates/adaptive/tests/integration/runtime_integration_tests.rs @@ -61,10 +61,6 @@ fn enable_operational_logs() { }); } -fn short_hash(value: &str) -> &str { - value.get(..16).unwrap_or(value) -} - fn reset_global() { enable_operational_logs(); let _ = clear_plugin_configuration(); @@ -517,14 +513,8 @@ async fn runtime_integration_acg_learner_reuses_learning_buckets_across_growing_ let requests = sample_growing_chat_requests("claude-3-5-sonnet"); let learner = AcgLearner::new(agent_id, 8, StabilityThresholds::default()); let learning_key = format!( - "{agent_id}::model=claude-3-5-sonnet::seed={}::system={}::tools=no-tools", - short_hash(&format!( - "user:{}", - nemo_relay_adaptive::acg::sha256_hex("Summarize the latest findings") - )), - short_hash(&nemo_relay_adaptive::acg::sha256_hex( - "You are a careful planner" - )), + "{agent_id}::model=claude-3-5-sonnet::seed=stable-scaffold::system={}::tools=no-tools", + nemo_relay_adaptive::acg::sha256_hex("You are a careful planner"), ); learner diff --git a/crates/adaptive/tests/unit/acg/economics_internal_tests.rs b/crates/adaptive/tests/unit/acg/economics_internal_tests.rs index 7a6a01364..69053c425 100644 --- a/crates/adaptive/tests/unit/acg/economics_internal_tests.rs +++ b/crates/adaptive/tests/unit/acg/economics_internal_tests.rs @@ -121,6 +121,7 @@ fn economics_internal_build_prefix_stats_stops_at_first_non_stable_block() { score(1, StabilityClass::Variable, 0.4, 0.9), ], stable_prefix_length: 2, + stable_prefix_fingerprint: None, total_observations: 6, }; diff --git a/crates/adaptive/tests/unit/acg/economics_policy_tests.rs b/crates/adaptive/tests/unit/acg/economics_policy_tests.rs index 1d4a00966..254508ee3 100644 --- a/crates/adaptive/tests/unit/acg/economics_policy_tests.rs +++ b/crates/adaptive/tests/unit/acg/economics_policy_tests.rs @@ -84,6 +84,7 @@ fn stability_result(scores: &[(f64, f64)], observation_count: u32) -> StabilityA }) .collect(), stable_prefix_length: scores.len(), + stable_prefix_fingerprint: None, total_observations: observation_count, } } diff --git a/crates/adaptive/tests/unit/acg/ir_builder_tests.rs b/crates/adaptive/tests/unit/acg/ir_builder_tests.rs index 2fee6ba62..c03629715 100644 --- a/crates/adaptive/tests/unit/acg/ir_builder_tests.rs +++ b/crates/adaptive/tests/unit/acg/ir_builder_tests.rs @@ -187,3 +187,65 @@ fn build_prompt_ir_omits_tool_schema_hashes_when_no_tools_are_present() { assert!(prompt_ir.tool_schema_hashes.is_none()); assert_eq!(prompt_ir.blocks[0].span_id.0, "user-0"); } + +#[test] +fn build_prompt_ir_canonicalizes_structured_output_before_user_content() { + let mut request = AnnotatedLlmRequest { + messages: vec![Message::User { + content: MessageContent::Text("variable task".to_string()), + name: None, + }], + model: Some("gpt-4o".to_string()), + params: None, + tools: None, + tool_choice: None, + store: None, + previous_response_id: None, + truncation: None, + reasoning: None, + include: None, + user: None, + metadata: None, + service_tier: None, + parallel_tool_calls: None, + max_output_tokens: None, + max_tool_calls: None, + top_logprobs: None, + stream: None, + extra: serde_json::Map::new(), + }; + request.extra.insert( + "response_format".to_string(), + serde_json::json!({"type":"json_schema","json_schema":{"required":["answer"],"type":"object"}}), + ); + + let first = build_prompt_ir(&request).unwrap(); + request.extra.insert( + "response_format".to_string(), + serde_json::json!({"json_schema":{"type":"object","required":["answer"]},"type":"json_schema"}), + ); + let reordered = build_prompt_ir(&request).unwrap(); + + assert_eq!( + first.structured_output_schema_id, + reordered.structured_output_schema_id + ); + assert!(first.structured_output_schema_id.is_some()); + assert_eq!( + first.blocks[0].content_type, + BlockContentType::StructuredOutput + ); + assert_eq!(first.blocks[1].role, PromptRole::User); + + request + .extra + .insert("response_format".to_string(), serde_json::Value::Null); + let null_contract = build_prompt_ir(&request).unwrap(); + assert!(null_contract.structured_output_schema_id.is_none()); + assert!( + null_contract + .blocks + .iter() + .all(|block| block.content_type != BlockContentType::StructuredOutput) + ); +} diff --git a/crates/adaptive/tests/unit/acg/multi_breakpoint_tests.rs b/crates/adaptive/tests/unit/acg/multi_breakpoint_tests.rs index f4c228482..ddf8b0017 100644 --- a/crates/adaptive/tests/unit/acg/multi_breakpoint_tests.rs +++ b/crates/adaptive/tests/unit/acg/multi_breakpoint_tests.rs @@ -81,6 +81,7 @@ fn layered_stability(observation_count: u32, layers: usize) -> StabilityAnalysis }) .collect(), stable_prefix_length: layers, + stable_prefix_fingerprint: None, total_observations: observation_count, } } diff --git a/crates/adaptive/tests/unit/acg/stability_internal_tests.rs b/crates/adaptive/tests/unit/acg/stability_internal_tests.rs index c39843586..fe9155ab4 100644 --- a/crates/adaptive/tests/unit/acg/stability_internal_tests.rs +++ b/crates/adaptive/tests/unit/acg/stability_internal_tests.rs @@ -78,3 +78,58 @@ fn stability_internal_effective_score_handles_zero_present_count() { assert_eq!(effective_stability_score(&observations, 3), 0.0); } + +#[test] +fn stability_internal_fingerprints_the_exact_dominant_prefix() { + let first = prompt(vec![ + block("system-0", 0, "stable policy"), + block("user-1", 1, "task alpha"), + ]); + let second = prompt(vec![ + block("system-0", 0, "stable policy"), + block("user-1", 1, "task beta"), + ]); + let changed = prompt(vec![ + block("system-0", 0, "changed policy"), + block("user-1", 1, "task gamma"), + ]); + + let result = analyze_stability(&[first.clone(), second], &StabilityThresholds::default()); + + assert_eq!(result.stable_prefix_length, 1); + assert_eq!( + result.stable_prefix_fingerprint, + prompt_prefix_fingerprint(&first, 1) + ); + assert_ne!( + result.stable_prefix_fingerprint, + prompt_prefix_fingerprint(&changed, 1) + ); + assert_eq!(prompt_prefix_fingerprint(&first, 0), None); +} + +#[test] +fn stability_internal_deserializes_persisted_state_without_a_prefix_fingerprint() { + let restored: StabilityAnalysisResult = serde_json::from_value(serde_json::json!({ + "scores": [], + "stable_prefix_length": 0, + "total_observations": 4 + })) + .unwrap(); + + assert_eq!(restored.stable_prefix_fingerprint, None); +} + +#[test] +fn profile_fingerprint_requires_source_hash_beyond_the_scaffold() { + let mut observation = prompt(vec![ + block("system-0", 0, "stable policy"), + block("user-1", 1, "stable task"), + ]); + observation.blocks[1].role = PromptRole::User; + + assert_eq!( + profile_prefix_fingerprint(&observation, 2, "learning-key"), + None + ); +} diff --git a/crates/adaptive/tests/unit/acg_component_tests.rs b/crates/adaptive/tests/unit/acg_component_tests.rs index fcf09c01a..1e4bf21b4 100644 --- a/crates/adaptive/tests/unit/acg_component_tests.rs +++ b/crates/adaptive/tests/unit/acg_component_tests.rs @@ -47,6 +47,7 @@ fn sample_hot_cache() -> Arc> { }, ], stable_prefix_length: 2, + stable_prefix_fingerprint: None, total_observations: 8, }), acg_observation_count: 8, @@ -231,6 +232,7 @@ fn stability_with_prefix(prefix_len: u32, observations: u32) -> StabilityAnalysi }) .collect(), stable_prefix_length: prefix_len as usize, + stable_prefix_fingerprint: None, total_observations: observations, } } @@ -261,6 +263,7 @@ fn layered_stability_result(observation_count: u32) -> StabilityAnalysisResult { }, ], stable_prefix_length: 3, + stable_prefix_fingerprint: None, total_observations: observation_count, } } @@ -285,6 +288,27 @@ fn sample_prompt_ir(span: &str) -> PromptIR { } } +fn stability_for_request( + agent_id: &str, + request: &LlmRequest, + mut stability: StabilityAnalysisResult, +) -> StabilityAnalysisResult { + let view = build_semantic_request_view(request).expect("fixture request should decode"); + let prompt_ir = crate::acg::ir_builder::build_prompt_ir(&view.annotated_request) + .expect("fixture request should build prompt ir"); + stability.stable_prefix_length = stability.stable_prefix_length.min(prompt_ir.blocks.len()); + let learning_key = derive_acg_learning_key(agent_id, &view.annotated_request); + stability.stable_prefix_fingerprint = Some( + crate::acg::stability::profile_prefix_fingerprint( + &prompt_ir, + stability.stable_prefix_length, + &learning_key, + ) + .expect("fixture stable prefix should exist"), + ); + stability +} + #[test] fn acg_component_default_config_and_thresholds_are_stable() { let config = AcgComponentConfig::default(); @@ -427,6 +451,11 @@ fn acg_component_translate_request_degrades_when_provider_semantics_do_not_match fn acg_component_translate_request_applies_openai_semantics_on_resolved_request_surface() { let request = sample_openai_responses_request(); let hot_cache = sample_hot_cache(); + hot_cache.write().unwrap().acg_stability = Some(stability_for_request( + "agent-1", + &request, + stability_with_prefix(2, 8), + )); let plugin = build_provider_plugin("openai").expect("openai plugin should build"); let translated = translate_request(&request, "agent-1", "openai", plugin.as_ref(), &hot_cache) @@ -502,7 +531,11 @@ fn acg_component_translate_request_applies_multiple_ordered_anthropic_breakpoint agent_hints_default: None, acg_profiles: std::collections::HashMap::new(), acg_profile_observation_counts: std::collections::HashMap::new(), - acg_stability: Some(layered_stability_result(6)), + acg_stability: Some(stability_for_request( + "agent-1", + &request, + layered_stability_result(6), + )), acg_observation_count: 6, })); @@ -548,7 +581,11 @@ fn rewrite_request_with_hot_cache_adaptive_placement_differs_by_state() { agent_hints_default: None, acg_profiles: std::collections::HashMap::new(), acg_profile_observation_counts: std::collections::HashMap::new(), - acg_stability: Some(stability_with_prefix(1, 8)), + acg_stability: Some(stability_for_request( + "agent-1", + &request, + stability_with_prefix(1, 8), + )), acg_observation_count: 8, })); let translated_short = @@ -561,7 +598,11 @@ fn rewrite_request_with_hot_cache_adaptive_placement_differs_by_state() { agent_hints_default: None, acg_profiles: std::collections::HashMap::new(), acg_profile_observation_counts: std::collections::HashMap::new(), - acg_stability: Some(stability_with_prefix(3, 8)), + acg_stability: Some(stability_for_request( + "agent-1", + &request, + stability_with_prefix(3, 8), + )), acg_observation_count: 8, })); let translated_long = @@ -602,7 +643,7 @@ fn acg_component_translate_request_uses_learning_key_for_profile_lookup() { agent_hints_default: None, acg_profiles: std::collections::HashMap::from([( learning_key.clone(), - layered_stability_result(6), + stability_for_request("agent-1", &request, layered_stability_result(6)), )]), acg_profile_observation_counts: std::collections::HashMap::from([(learning_key, 6)]), acg_stability: None, @@ -633,7 +674,11 @@ async fn acg_component_stream_execution_intercept_rewrites_streaming_requests() agent_hints_default: None, acg_profiles: std::collections::HashMap::new(), acg_profile_observation_counts: std::collections::HashMap::new(), - acg_stability: Some(layered_stability_result(6)), + acg_stability: Some(stability_for_request( + "agent-1", + &request, + layered_stability_result(6), + )), acg_observation_count: 6, })); @@ -689,6 +734,7 @@ fn acg_component_build_intent_bundle_requires_at_least_two_observations() { }, ], stable_prefix_length: 2, + stable_prefix_fingerprint: None, total_observations: 1, }; @@ -766,6 +812,39 @@ async fn acg_component_load_persisted_state_prefers_stability_and_falls_back_to_ assert_eq!(guard.acg_observation_count, 2); } +#[tokio::test] +async fn acg_component_rehydrates_matching_profile_and_rejects_changed_scaffold() { + let agent_id = "agent-restarted"; + let request = sample_openai_responses_request(); + let stability = stability_for_request( + agent_id, + &request, + stability_with_prefix(2, MIN_ACG_OBSERVATIONS), + ); + let backend = InMemoryBackend::new(); + backend.store_stability(agent_id, &stability).await.unwrap(); + + let hot_cache = Arc::new(RwLock::new(HotCache { + plan: None, + trie: None, + agent_hints_default: None, + acg_profiles: std::collections::HashMap::new(), + acg_profile_observation_counts: std::collections::HashMap::new(), + acg_stability: None, + acg_observation_count: 0, + })); + load_persisted_acg_state(agent_id, &backend, &hot_cache) + .await + .unwrap(); + let plugin = build_provider_plugin("openai").unwrap(); + + assert!(translate_request(&request, agent_id, "openai", plugin.as_ref(), &hot_cache).is_some()); + + let mut changed = request; + changed.content["instructions"] = json!("A different system policy."); + assert!(translate_request(&changed, agent_id, "openai", plugin.as_ref(), &hot_cache).is_none()); +} + #[tokio::test] async fn acg_component_load_persisted_state_handles_empty_backend_and_poisoned_cache() { let backend = InMemoryBackend::new(); @@ -863,7 +942,12 @@ fn acg_component_build_intent_bundle_supports_openai_and_rejects_unknown_provide let request = sample_annotated_request("gpt-4o"); let prompt_ir = crate::acg::ir_builder::build_prompt_ir(&request).unwrap(); let plugin = build_provider_plugin("openai").unwrap(); - let stability = layered_stability_result(4); + let mut stability = stability_with_prefix(2, 4); + stability.stable_prefix_fingerprint = crate::acg::stability::profile_prefix_fingerprint( + &prompt_ir, + stability.stable_prefix_length, + &derive_acg_learning_key("agent-openai", &request), + ); let bundle = build_intent_bundle( "agent-openai", @@ -895,6 +979,157 @@ fn acg_component_build_intent_bundle_supports_openai_and_rejects_unknown_provide ); } +#[test] +fn acg_component_requires_the_exact_learned_prefix_before_reuse() { + let request = sample_annotated_request("gpt-4o"); + let prompt_ir = crate::acg::ir_builder::build_prompt_ir(&request).unwrap(); + let mut stability = crate::acg::stability::analyze_stability( + std::slice::from_ref(&prompt_ir), + &crate::acg::stability::StabilityThresholds::default(), + ); + let learning_key = derive_acg_learning_key("agent-openai", &request); + stability.stable_prefix_fingerprint = crate::acg::stability::profile_prefix_fingerprint( + &prompt_ir, + stability.stable_prefix_length, + &learning_key, + ); + let plugin = build_provider_plugin("openai").unwrap(); + + assert!( + build_intent_bundle( + "agent-openai", + "openai", + plugin.as_ref(), + RequestSurface::OpenAIChat, + &request, + &prompt_ir, + &stability, + MIN_ACG_OBSERVATIONS, + ) + .is_some() + ); + + let mut changed_prefix = prompt_ir.clone(); + changed_prefix.blocks[0].content = "changed policy".to_string(); + assert!( + build_intent_bundle( + "agent-openai", + "openai", + plugin.as_ref(), + RequestSurface::OpenAIChat, + &request, + &changed_prefix, + &stability, + MIN_ACG_OBSERVATIONS, + ) + .is_none() + ); + + let mut legacy_stability = stability; + legacy_stability.stable_prefix_fingerprint = None; + assert!( + build_intent_bundle( + "agent-openai", + "openai", + plugin.as_ref(), + RequestSurface::OpenAIChat, + &request, + &prompt_ir, + &legacy_stability, + MIN_ACG_OBSERVATIONS, + ) + .is_none() + ); + + legacy_stability.stable_prefix_length = prompt_ir.blocks.len() + 1; + assert!( + build_intent_bundle( + "agent-openai", + "openai", + plugin.as_ref(), + RequestSurface::OpenAIChat, + &request, + &prompt_ir, + &legacy_stability, + MIN_ACG_OBSERVATIONS, + ) + .is_none() + ); +} + +#[test] +fn acg_component_rejects_raw_user_changes_beyond_the_stable_scaffold() { + let request = sample_annotated_request("gpt-4o"); + let prompt_ir = crate::acg::ir_builder::build_prompt_ir(&request).unwrap(); + let mut stability = crate::acg::stability::analyze_stability( + std::slice::from_ref(&prompt_ir), + &crate::acg::stability::StabilityThresholds::default(), + ); + let learning_key = derive_acg_learning_key("agent-openai", &request); + stability.stable_prefix_fingerprint = crate::acg::stability::profile_prefix_fingerprint( + &prompt_ir, + stability.stable_prefix_length, + &learning_key, + ); + + let mut changed_request = request.clone(); + changed_request.messages[1] = Message::User { + content: MessageContent::Text("Hello ".to_string()), + name: None, + }; + let changed_ir = crate::acg::ir_builder::build_prompt_ir(&changed_request).unwrap(); + let plugin = build_provider_plugin("openai").unwrap(); + + assert!( + build_intent_bundle( + "agent-openai", + "openai", + plugin.as_ref(), + RequestSurface::OpenAIChat, + &changed_request, + &changed_ir, + &stability, + MIN_ACG_OBSERVATIONS, + ) + .is_none() + ); +} + +#[test] +fn acg_component_rejects_raw_whitespace_changes_inside_the_stable_scaffold() { + let request = sample_annotated_request("gpt-4o"); + let prompt_ir = crate::acg::ir_builder::build_prompt_ir(&request).unwrap(); + let mut stability = stability_with_prefix(1, MIN_ACG_OBSERVATIONS); + stability.stable_prefix_fingerprint = crate::acg::stability::profile_prefix_fingerprint( + &prompt_ir, + 1, + &derive_acg_learning_key("agent-openai", &request), + ); + + let mut changed_request = request.clone(); + changed_request.messages[0] = Message::System { + content: MessageContent::Text("You are helpful. ".to_string()), + name: None, + }; + let changed_ir = crate::acg::ir_builder::build_prompt_ir(&changed_request).unwrap(); + assert_eq!(prompt_ir.blocks[0].content, changed_ir.blocks[0].content); + + let plugin = build_provider_plugin("openai").unwrap(); + assert!( + build_intent_bundle( + "agent-openai", + "openai", + plugin.as_ref(), + RequestSurface::OpenAIChat, + &changed_request, + &changed_ir, + &stability, + MIN_ACG_OBSERVATIONS, + ) + .is_none() + ); +} + #[test] fn acg_component_anthropic_cache_intents_fail_open_when_surface_or_model_do_not_match() { let plugin = build_provider_plugin("anthropic").unwrap(); @@ -947,7 +1182,7 @@ fn acg_component_build_hint_translation_and_apply_hint_translation_cover_passthr let prompt_ir = crate::acg::ir_builder::build_prompt_ir(&semantic_view.annotated_request).unwrap(); let plugin = build_provider_plugin("openai").unwrap(); - let stability = layered_stability_result(4); + let stability = stability_for_request("agent-openai", &request, layered_stability_result(4)); let bundle = build_intent_bundle( "agent-openai", "openai", @@ -1002,7 +1237,7 @@ fn acg_component_translate_request_uses_profile_specific_stability_and_fails_ope agent_hints_default: None, acg_profiles: std::collections::HashMap::from([( learning_key.clone(), - layered_stability_result(6), + stability_for_request("agent-profile", &request, layered_stability_result(6)), )]), acg_profile_observation_counts: std::collections::HashMap::from([(learning_key, 6)]), acg_stability: None, @@ -1049,7 +1284,11 @@ async fn acg_component_execution_intercept_rewrites_non_streaming_requests() { agent_hints_default: None, acg_profiles: std::collections::HashMap::new(), acg_profile_observation_counts: std::collections::HashMap::new(), - acg_stability: Some(layered_stability_result(6)), + acg_stability: Some(stability_for_request( + "agent-1", + &request, + layered_stability_result(6), + )), acg_observation_count: 6, })); @@ -1148,7 +1387,7 @@ fn acg_component_build_hint_translation_succeeds_for_openai_and_anthropic() { RequestSurface::OpenAIChat, &openai_view.annotated_request, &openai_prompt_ir, - &layered_stability_result(4), + &stability_for_request("agent-openai", &openai_request, stability_with_prefix(1, 4)), 4, ) .unwrap(); @@ -1189,7 +1428,11 @@ fn acg_component_build_hint_translation_succeeds_for_openai_and_anthropic() { RequestSurface::AnthropicMessages, &anthropic_view.annotated_request, &anthropic_prompt_ir, - &layered_stability_result(6), + &stability_for_request( + "agent-anthropic", + &anthropic_request, + layered_stability_result(6), + ), 6, ) .unwrap(); @@ -1323,7 +1566,11 @@ fn acg_component_request_intercept_rewrites_annotation_without_mutating_provider agent_hints_default: None, acg_profiles: std::collections::HashMap::new(), acg_profile_observation_counts: std::collections::HashMap::new(), - acg_stability: Some(layered_stability_result(6)), + acg_stability: Some(stability_for_request( + "agent-1", + &request, + layered_stability_result(6), + )), acg_observation_count: 6, })); let intercept = create_acg_llm_request_intercept( diff --git a/crates/adaptive/tests/unit/acg_learner_tests.rs b/crates/adaptive/tests/unit/acg_learner_tests.rs index 664ff5126..c960c24d4 100644 --- a/crates/adaptive/tests/unit/acg_learner_tests.rs +++ b/crates/adaptive/tests/unit/acg_learner_tests.rs @@ -228,6 +228,44 @@ async fn acg_learner_returns_early_without_llm_requests() { .unwrap(); } +#[tokio::test(flavor = "current_thread")] +async fn acg_learner_accumulates_same_scaffold_calls_from_one_run_in_one_profile() { + let first = sample_request("gpt-4o", "Stable system", "task alpha"); + let second = sample_request("gpt-4o", "Stable system", "task beta"); + let learning_key = derive_acg_learning_key("agent-a", &first); + assert_eq!(learning_key, derive_acg_learning_key("agent-a", &second)); + + let learner = AcgLearner::new("agent-a", 8, StabilityThresholds::default()); + let backend = crate::storage::memory::InMemoryBackend::new(); + let hot_cache = empty_cache(); + learner + .process_run(&sample_run(vec![first, second]), &backend, &hot_cache) + .await + .unwrap(); + + assert_eq!( + backend + .load_observations(&learning_key) + .await + .unwrap() + .unwrap() + .len(), + 2 + ); + let guard = hot_cache.read().unwrap(); + assert_eq!(guard.acg_profiles.len(), 1); + assert_eq!(guard.acg_profile_observation_counts[&learning_key], 2); + let stored = &guard.acg_profiles[&learning_key]; + assert_eq!( + stored.stable_prefix_fingerprint, + crate::acg::stability::profile_prefix_fingerprint( + &build_prompt_ir(&sample_request("gpt-4o", "Stable system", "task alpha")).unwrap(), + stored.stable_prefix_length, + &learning_key, + ) + ); +} + #[tokio::test(flavor = "current_thread")] async fn acg_learner_trims_observation_windows_and_updates_agent_seed() { let learner = AcgLearner::new("agent-a", 2, StabilityThresholds::default()); diff --git a/crates/adaptive/tests/unit/acg_profile_tests.rs b/crates/adaptive/tests/unit/acg_profile_tests.rs index 3d5e23f7e..f3bb01790 100644 --- a/crates/adaptive/tests/unit/acg_profile_tests.rs +++ b/crates/adaptive/tests/unit/acg_profile_tests.rs @@ -46,6 +46,104 @@ fn sample_tool(name: &str) -> ToolDefinition { } } +fn scaffolded_request(task: &str) -> AnnotatedLlmRequest { + request( + vec![ + Message::System { + content: MessageContent::Text("Follow policy".to_string()), + name: None, + }, + Message::User { + content: MessageContent::Text(task.to_string()), + name: None, + }, + ], + Some(vec![sample_tool("search")]), + ) +} + +#[test] +fn acg_learning_key_groups_variable_tasks_by_stable_scaffold() { + let first = scaffolded_request("find alpha"); + let second = scaffolded_request("find beta"); + + assert_eq!( + derive_acg_learning_key("agent-a", &first), + derive_acg_learning_key("agent-a", &second) + ); +} + +#[test] +fn acg_learning_key_separates_scaffold_and_output_contract_changes() { + let baseline = scaffolded_request("find alpha"); + + let mut changed_system = baseline.clone(); + changed_system.messages[0] = Message::System { + content: MessageContent::Text("Follow a different policy".to_string()), + name: None, + }; + + let mut changed_tool = baseline.clone(); + changed_tool.tools = Some(vec![sample_tool("lookup")]); + + let mut contract_a = baseline.clone(); + contract_a.extra.insert( + "response_format".to_string(), + json!({"type":"json_schema","json_schema":{"type":"object","properties":{"a":{"type":"string"}}}}), + ); + let mut contract_b = baseline.clone(); + contract_b.extra.insert( + "response_format".to_string(), + json!({"json_schema":{"properties":{"b":{"type":"string"}},"type":"object"},"type":"json_schema"}), + ); + + let baseline_key = derive_acg_learning_key("agent-a", &baseline); + assert_ne!( + baseline_key, + derive_acg_learning_key("agent-a", &changed_system) + ); + assert_ne!( + baseline_key, + derive_acg_learning_key("agent-a", &changed_tool) + ); + assert_ne!( + derive_acg_learning_key("agent-a", &contract_a), + derive_acg_learning_key("agent-a", &contract_b) + ); + + let mut null_contract = baseline.clone(); + null_contract + .extra + .insert("response_format".to_string(), serde_json::Value::Null); + assert_eq!( + baseline_key, + derive_acg_learning_key("agent-a", &null_contract) + ); +} + +#[test] +fn acg_learning_key_keeps_task_seed_without_a_stable_scaffold() { + let first = request( + vec![Message::User { + content: MessageContent::Text("alpha".to_string()), + name: None, + }], + None, + ); + let second = request( + vec![Message::User { + content: MessageContent::Text("beta".to_string()), + name: None, + }], + None, + ); + + assert_ne!( + derive_acg_learning_key("agent-a", &first), + derive_acg_learning_key("agent-a", &second) + ); +} + #[test] fn acg_profile_derivation_covers_anchor_hash_system_fallback_and_empty_tools() { let layered = request( diff --git a/crates/adaptive/tests/unit/cache_diagnostics_tests.rs b/crates/adaptive/tests/unit/cache_diagnostics_tests.rs index b46c7b846..a369ec716 100644 --- a/crates/adaptive/tests/unit/cache_diagnostics_tests.rs +++ b/crates/adaptive/tests/unit/cache_diagnostics_tests.rs @@ -82,6 +82,7 @@ fn make_hot_cache(stable_prefix_length: Option) -> HotCache { }) .collect(), stable_prefix_length, + stable_prefix_fingerprint: None, total_observations: 4, }), acg_observation_count: 4, diff --git a/crates/adaptive/tests/unit/intercepts_tests.rs b/crates/adaptive/tests/unit/intercepts_tests.rs index 971d71c60..f19048d3a 100644 --- a/crates/adaptive/tests/unit/intercepts_tests.rs +++ b/crates/adaptive/tests/unit/intercepts_tests.rs @@ -50,6 +50,7 @@ fn make_hot_cache( acg_stability: Some(StabilityAnalysisResult { scores: vec![], stable_prefix_length, + stable_prefix_fingerprint: None, total_observations: observation_count, }), acg_observation_count: observation_count, diff --git a/crates/adaptive/tests/unit/runtime_features_tests.rs b/crates/adaptive/tests/unit/runtime_features_tests.rs index 5a53c7734..7c88fe1d1 100644 --- a/crates/adaptive/tests/unit/runtime_features_tests.rs +++ b/crates/adaptive/tests/unit/runtime_features_tests.rs @@ -87,6 +87,14 @@ fn layered_acg_request() -> LlmRequest { } fn layered_acg_stability_result(observation_count: u32) -> StabilityAnalysisResult { + let request = layered_acg_request(); + let annotated_request = nemo_relay::codec::resolve::request_codec( + nemo_relay::codec::resolve::ProviderSurface::AnthropicMessages, + ) + .decode(&request) + .expect("fixture request should decode"); + let prompt_ir = crate::acg::ir_builder::build_prompt_ir(&annotated_request) + .expect("fixture request should build prompt ir"); StabilityAnalysisResult { scores: vec![ BlockStabilityScore { @@ -112,6 +120,11 @@ fn layered_acg_stability_result(observation_count: u32) -> StabilityAnalysisResu }, ], stable_prefix_length: 3, + stable_prefix_fingerprint: crate::acg::stability::profile_prefix_fingerprint( + &prompt_ir, + 3, + &crate::acg_profile::derive_acg_learning_key("agent-acg", &annotated_request), + ), total_observations: observation_count, } } diff --git a/crates/adaptive/tests/unit/runtime_tests.rs b/crates/adaptive/tests/unit/runtime_tests.rs index 77231ccdb..baff03e69 100644 --- a/crates/adaptive/tests/unit/runtime_tests.rs +++ b/crates/adaptive/tests/unit/runtime_tests.rs @@ -31,10 +31,6 @@ fn reset_runtime_context() { set_thread_scope_stack(create_scope_stack()); } -fn short_hash(value: &str) -> &str { - value.get(..16).unwrap_or(value) -} - fn sample_annotated_request(model: Option<&str>) -> AnnotatedLlmRequest { AnnotatedLlmRequest { messages: vec![ @@ -417,12 +413,8 @@ fn adaptive_acg_defaults_and_profile_key_behavior_stay_stable() { &sample_annotated_request(Some("claude-sonnet-4")), ); let expected_learning_key = format!( - "agent-1::model=claude-sonnet-4::seed={}::system={}::tools=no-tools", - short_hash(&format!( - "user:{}", - crate::acg::sha256_hex("Summarize the latest findings") - )), - short_hash(&crate::acg::sha256_hex("You are a careful planner")), + "agent-1::model=claude-sonnet-4::seed=stable-scaffold::system={}::tools=no-tools", + crate::acg::sha256_hex("You are a careful planner"), ); assert_eq!(learning_key, expected_learning_key,); @@ -498,9 +490,9 @@ fn adaptive_acg_defaults_and_profile_key_behavior_stay_stable() { "agent-1", &sample_layered_request(Some("claude-sonnet-4"), "Python review guide"), ); - assert_ne!( + assert_eq!( rust_learning_key, python_learning_key, - "layered requests should still separate learning buckets when the stable anchor differs", + "variable first-user guide content should share the stable system scaffold", ); let rust_bundle_variant = AnnotatedLlmRequest { diff --git a/crates/adaptive/tests/unit/storage_memory_internal_tests.rs b/crates/adaptive/tests/unit/storage_memory_internal_tests.rs index 311f007dc..2a95452f0 100644 --- a/crates/adaptive/tests/unit/storage_memory_internal_tests.rs +++ b/crates/adaptive/tests/unit/storage_memory_internal_tests.rs @@ -84,6 +84,7 @@ fn sample_stability_record() -> StabilityAnalysisResult { observation_count: 1, }], stable_prefix_length: 1, + stable_prefix_fingerprint: None, total_observations: 1, } } diff --git a/crates/adaptive/tests/unit/storage_tests.rs b/crates/adaptive/tests/unit/storage_tests.rs index 955dc8c76..7d48ea3fe 100644 --- a/crates/adaptive/tests/unit/storage_tests.rs +++ b/crates/adaptive/tests/unit/storage_tests.rs @@ -89,6 +89,7 @@ fn sample_stability(agent_id: &str) -> StabilityAnalysisResult { observation_count: 3, }], stable_prefix_length: 1, + stable_prefix_fingerprint: None, total_observations: 3, } } diff --git a/crates/adaptive/tests/unit/types_tests.rs b/crates/adaptive/tests/unit/types_tests.rs index c31b0c58d..e121c5390 100644 --- a/crates/adaptive/tests/unit/types_tests.rs +++ b/crates/adaptive/tests/unit/types_tests.rs @@ -41,6 +41,7 @@ fn sample_stability_result() -> StabilityAnalysisResult { observation_count: 4, }], stable_prefix_length: 1, + stable_prefix_fingerprint: Some("stable-prefix-1".to_string()), total_observations: 4, } } @@ -196,4 +197,13 @@ fn hot_cache_serialization_keeps_acg_field_names_stable() { decoded.acg_stability.as_ref().unwrap().total_observations, 4 ); + assert_eq!( + decoded + .acg_stability + .as_ref() + .unwrap() + .stable_prefix_fingerprint + .as_deref(), + Some("stable-prefix-1") + ); } diff --git a/crates/python/tests/coverage/py_storage_coverage_tests.rs b/crates/python/tests/coverage/py_storage_coverage_tests.rs index 0661dcf90..2a212f8a4 100644 --- a/crates/python/tests/coverage/py_storage_coverage_tests.rs +++ b/crates/python/tests/coverage/py_storage_coverage_tests.rs @@ -140,6 +140,7 @@ fn sample_stability_result() -> StabilityAnalysisResult { observation_count: 2, }], stable_prefix_length: 1, + stable_prefix_fingerprint: Some("stable-prefix-1".to_string()), total_observations: 2, } }