diff --git a/CHANGELOG.md b/CHANGELOG.md index 684ece187..ee774cbea 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,22 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). ## [Unreleased] +### Removed + +- `zeph-core`: removed the `#[cfg(test)]`-only legacy sequential-dispatch harness in + `agent/tool_execution/tool_result.rs` (`handle_tool_result`, `record_tool_output_outcome`, + `process_successful_tool_output`, `handle_confirmation_required`) and the parallel legacy LLM + dispatch harness in `agent/tool_execution/llm_dispatch.rs` (`call_llm_with_timeout`, + `call_llm_non_streaming`, `call_llm_with_retry`) — both were superseded in production by + `process_one_tool_result`/`handle_confirmation_phase` (tier_loop.rs) and + `call_chat_with_tools_retry`/`call_chat_with_tools` respectively, but were kept around only to + keep their dependent tests green, which previously masked a real production gap (issue #6364). + All ~20 dependent tests across `parallel_and_handle_tests.rs`, `retry_and_skill_env_tests.rs`, + and `compaction_e2e.rs` were repointed to exercise the real production entry points + (`process_one_tool_result`, `stamp_and_send_tier_start`, `call_chat_with_tools_retry`, + `process_response`); response-cache tests were consolidated into the existing native-tool-use + cache coverage in `sanitize_and_native_tests.rs` (issue #6560). + ### Fixed - `zeph-core`/`zeph-skills`: the failure-driven self-learning path (`store_improved_version`) diff --git a/book/src/advanced/context.md b/book/src/advanced/context.md index 1e5bdbdd0..99e301fce 100644 --- a/book/src/advanced/context.md +++ b/book/src/advanced/context.md @@ -633,7 +633,7 @@ Text previews use safe UTF-8 truncation (`truncate_chars()`) that never splits a ## Reactive Retry on Context Length Errors -LLM calls in the agent loop (`call_llm_with_retry()` and `call_chat_with_tools_retry()`) intercept context length errors and automatically compact before retrying. The flow: +LLM calls in the agent loop (`call_chat_with_tools_retry()`) intercept context length errors and automatically compact before retrying. The flow: 1. Send messages to the LLM provider 2. If the provider returns a context length error, trigger `compact_context()` diff --git a/book/src/reference/security/untrusted-content-isolation.md b/book/src/reference/security/untrusted-content-isolation.md index 0549a501d..e6dd87684 100644 --- a/book/src/reference/security/untrusted-content-isolation.md +++ b/book/src/reference/security/untrusted-content-isolation.md @@ -102,10 +102,10 @@ The sanitizer is applied at every untrusted boundary: | Source | Trust Level | Integration Point | |--------|------------|-------------------| -| Shell / file tool results | `LocalUntrusted` | `handle_tool_result()` — both normal and confirmation-required paths | -| Web scrape output | `ExternalUntrusted` | `handle_tool_result()` | -| MCP tool responses | `ExternalUntrusted` | `handle_tool_result()` | -| A2A messages | `ExternalUntrusted` | `handle_tool_result()` | +| Shell / file tool results | `LocalUntrusted` | `process_one_tool_result()` — both normal and confirmation-resolved paths (`handle_confirmation_phase()`) | +| Web scrape output | `ExternalUntrusted` | `process_one_tool_result()` | +| MCP tool responses | `ExternalUntrusted` | `process_one_tool_result()` | +| A2A messages | `ExternalUntrusted` | `process_one_tool_result()` | | Native tool-use results (Claude provider) | `LocalUntrusted` or `ExternalUntrusted` | `handle_native_tool_calls()` — routes through `sanitize_tool_output()` before placing output in `ToolResult` parts | | Semantic memory recall | `ExternalUntrusted` | `prepare_context()` | | Cross-session memory | `ExternalUntrusted` | `prepare_context()` | diff --git a/crates/zeph-acp/README.md b/crates/zeph-acp/README.md index f1480a629..a1c09afb1 100644 --- a/crates/zeph-acp/README.md +++ b/crates/zeph-acp/README.md @@ -120,7 +120,7 @@ The agent loop emits `StopHint` values that `ZephAcpAgent` maps to protocol-leve 1. **Before execution** — `SessionUpdate::ToolCall` with `status: InProgress` is sent as soon as tool invocation begins, enabling the IDE to display a running indicator. 2. **After execution** — `SessionUpdate::ToolCallUpdate` with `status: Completed` (or `Failed` on error) carries the output content and optional file locations. -Each tool call is identified by a UUID generated per invocation. The UUID is threaded through `LoopbackEvent::ToolStart` / `LoopbackEvent::ToolOutput` so the update correctly references the original announcement. Both the fenced-block execution path (`handle_tool_result`) and the structured parallel tool-call path emit this full two-step sequence unconditionally — output content always appears inside a tool call block in the IDE regardless of which path handled the tool. +Each tool call is identified by a UUID generated per invocation. The UUID is threaded through `LoopbackEvent::ToolStart` / `LoopbackEvent::ToolOutput` so the update correctly references the original announcement. The structured parallel tool-call path emits this full two-step sequence unconditionally — output content always appears inside a tool call block in the IDE. **Terminal release ordering** — when a shell tool call embeds a terminal via `ToolCallContent::Terminal`, the ACP spec requires the terminal to remain alive until the IDE has processed the `tool_call_update` notification. `ZephAcpAgent` defers `terminal/release` until after all notifications for that event are dispatched. The deferred release is triggered from the `prompt()` event loop via `AcpShellExecutor::release_terminal()`, which is retained in `SessionEntry` for exactly this purpose. diff --git a/crates/zeph-core/README.md b/crates/zeph-core/README.md index 8e359eab3..39a6d2d66 100644 --- a/crates/zeph-core/README.md +++ b/crates/zeph-core/README.md @@ -21,7 +21,7 @@ Core orchestration crate for the Zeph agent. Manages the main agent loop, bootst | `agent::learning_engine` | `LearningEngine` — owns `LearningConfig`, tracks per-turn reflection state; delegates self-learning decisions to `is_enabled()` / `mark_reflection_used()` | | `agent::feedback_detector` | `FeedbackDetector` (regex) and `JudgeDetector` (LLM-backed) — implicit correction detection from user messages with multi-language support (7 languages: English, Russian, Spanish, German, French, Portuguese, Chinese); `JudgeDetector` runs in background via `tokio::spawn` with sliding-window rate limiter (5 calls / 60 s) and XML-escaped adversarial-defense prompt; adaptive threshold gates judge invocation to the regex uncertainty zone | | `agent::persistence` | Message persistence and background graph extraction integration; `maybe_spawn_graph_extraction` fires a `SemanticMemory::spawn_graph_extraction` task per user turn with injection-flag guard and last-4-user-messages context window | -| `agent::tool_execution` | Tool call handling, redaction, result processing; both the fenced-block path (`handle_tool_result`) and the structured tool-call path unconditionally emit `LoopbackEvent::ToolStart` (UUID generated per call) before execution and `LoopbackEvent::ToolOutput` (matching UUID, `is_error` flag) after; `call_llm_with_retry()` / `call_chat_with_tools_retry()` — auto-detect `ContextLengthExceeded`, compact context, and retry (max 2 attempts); `prune_stale_tool_outputs` invokes `count_tokens` once per `ToolResult` part | +| `agent::tool_execution` | Tool call handling, redaction, result processing; the structured tool-call path unconditionally emits `LoopbackEvent::ToolStart` (UUID generated per call, via `stamp_and_send_tier_start`) before execution and `LoopbackEvent::ToolOutput` (matching UUID, `is_error` flag, via `process_one_tool_result`) after; `call_chat_with_tools_retry()` — auto-detects `ContextLengthExceeded`, compacts context, and retries (max 2 attempts); `prune_stale_tool_outputs` invokes `count_tokens` once per `ToolResult` part | | `agent::message_queue` | Message queue management | | `agent::builder` | Agent builder API | | `agent::commands` | Chat command dispatch (skills, feedback, skill management via `/skill install`, `/skill remove`, `/skill reject `, sub-agent management via `/agent`, etc.) | diff --git a/crates/zeph-core/src/agent/tests/compaction_e2e.rs b/crates/zeph-core/src/agent/tests/compaction_e2e.rs index 78af96e4f..3c54a5f7a 100644 --- a/crates/zeph-core/src/agent/tests/compaction_e2e.rs +++ b/crates/zeph-core/src/agent/tests/compaction_e2e.rs @@ -33,14 +33,9 @@ async fn agent_recovers_from_context_length_exceeded_and_produces_response() { metadata: MessageMetadata::default(), }); - // call_llm_with_retry is the direct entry point for the retry/compact flow - let result = agent.call_llm_with_retry(2).await.unwrap(); - - assert!( - result.is_some(), - "agent must produce a response after recovering from context length error" - ); - assert_eq!(result.as_deref(), Some("final answer")); + // process_response is the real production entry point for the native tool loop, which + // internally retries via call_chat_with_tools_retry(_, 2) on a context-length error. + agent.process_response().await.unwrap(); // Verify the channel received the recovered response let sent = agent.channel.sent_messages(); diff --git a/crates/zeph-core/src/agent/tool_execution/llm_dispatch.rs b/crates/zeph-core/src/agent/tool_execution/llm_dispatch.rs index 2a359e3dc..ae001b137 100644 --- a/crates/zeph-core/src/agent/tool_execution/llm_dispatch.rs +++ b/crates/zeph-core/src/agent/tool_execution/llm_dispatch.rs @@ -2,8 +2,6 @@ // SPDX-License-Identifier: MIT OR Apache-2.0 use tracing::Instrument; -#[cfg(test)] -use zeph_common::text::estimate_tokens; use zeph_llm::provider::{ ChatResponse, LlmProvider, MessagePart, Role, ThinkingBlock, ToolDefinition, }; @@ -352,201 +350,4 @@ impl Agent { last.rebuild_content(); } } - - #[cfg(test)] - pub(super) async fn call_llm_with_timeout( - &mut self, - ) -> Result, crate::agent::error::AgentError> { - if self.runtime.lifecycle.cancel_token.is_cancelled() { - return Ok(None); - } - - if let Some(ref tracker) = self.runtime.metrics.cost_tracker - && let Err(e) = tracker.check_budget() - { - self.update_metrics(|m| m.cost_budget_exhausted += 1); - self.channel - .send(&format!("Budget limit reached: {e}")) - .await?; - return Ok(None); - } - - let query_embedding = match self.check_response_cache().await? { - super::CacheCheckResult::Hit(resp) => return Ok(Some(resp)), - super::CacheCheckResult::Miss { query_embedding } => query_embedding, - }; - - let llm_timeout = std::time::Duration::from_secs(self.runtime.config.timeouts.llm_seconds); - let start = std::time::Instant::now(); - let prompt_estimate = self.runtime.providers.cached_prompt_tokens; - - let memcot_state = match self.services.memory.extraction.memcot_accumulator.as_ref() { - Some(acc) => acc.current_state().await, - None => None, - }; - let dump_id = - self.runtime - .debug - .debug_dumper - .as_ref() - .map(|d: &crate::debug_dump::DebugDumper| { - let provider_request = if d.is_trace_format() { - serde_json::Value::Null - } else { - self.provider.debug_request_json( - &self.msg.messages, - &[], - self.provider.supports_streaming(), - ) - }; - d.dump_request(&crate::debug_dump::RequestDebugDump { - model_name: &self.runtime.config.model_name, - messages: &self.msg.messages, - tools: &[], - provider_request, - memcot_state: memcot_state.as_deref(), - }) - }); - - let trace_guard = self.runtime.debug.trace_collector.as_ref().and_then(|tc| { - self.runtime - .debug - .current_iteration_span_id - .map(|id| tc.begin_llm_request(id)) - }); - - let llm_span = tracing::info_span!( - "llm.turn_call", - model = %self.runtime.config.model_name, - provider = self.provider.name(), - ); - let result = self - .call_llm_non_streaming( - llm_timeout, - start, - prompt_estimate, - dump_id, - llm_span, - query_embedding, - ) - .await; - - if let Some(guard) = trace_guard - && let Some(ref mut tc) = self.runtime.debug.trace_collector - { - let latency = u64::try_from(start.elapsed().as_millis()).unwrap_or(u64::MAX); - let (prompt_tokens, completion_tokens) = - self.provider.last_usage().unwrap_or((prompt_estimate, 0)); - tc.end_llm_request( - guard, - &crate::debug_dump::trace::LlmAttributes { - model: self.runtime.config.model_name.clone(), - prompt_tokens, - completion_tokens, - latency_ms: latency, - streaming: false, - cache_hit: false, - }, - ); - } - - result - } - - #[cfg(test)] - async fn call_llm_non_streaming( - &mut self, - llm_timeout: std::time::Duration, - start: std::time::Instant, - prompt_estimate: u64, - dump_id: Option, - llm_span: tracing::Span, - query_embedding: Option>, - ) -> Result, crate::agent::error::AgentError> { - let cancel = self.runtime.lifecycle.cancel_token.clone(); - let chat_fut = self.provider.chat(&self.msg.messages).instrument(llm_span); - let result = tokio::select! { - r = tokio::time::timeout(llm_timeout, chat_fut) => r, - () = cancel.cancelled() => { - tracing::info!("LLM call cancelled by user"); - self.update_metrics(|m| m.cancellations += 1); - self.channel.send("[Cancelled]").await?; - return Ok(None); - } - }; - match result { - Ok(Ok(resp)) => { - let elapsed = start.elapsed(); - let latency = u64::try_from(elapsed.as_millis()).unwrap_or(u64::MAX); - let completion_heuristic = estimate_tokens(&resp) as u64; - let (final_prompt, final_completion) = self - .provider - .last_usage() - .unwrap_or((prompt_estimate, completion_heuristic)); - let reasoning = self.provider.last_reasoning_tokens().unwrap_or(0); - self.update_metrics(|m| { - m.api_calls += 1; - m.last_llm_latency_ms = latency; - m.context_tokens = final_prompt; - m.prompt_tokens += final_prompt; - m.completion_tokens += final_completion; - m.total_tokens = m.prompt_tokens + m.completion_tokens; - m.reasoning_tokens += reasoning; - }); - self.record_cost_and_cache(final_prompt, final_completion); - self.record_successful_task(); - if let Some(ref recorder) = self.runtime.metrics.histogram_recorder { - recorder.observe_llm_latency(elapsed); - } - if self.run_response_verification(&resp) { - let _ = self - .channel - .send("[security] Response blocked by injection detection.") - .await; - return Ok(None); - } - let cleaned = self.scan_output_and_warn(&resp); - if let (Some(d), Some(id)) = (self.runtime.debug.debug_dumper.as_ref(), dump_id) { - d.dump_response(id, &cleaned); - } - let display = self.maybe_redact(&cleaned); - self.channel.send(&display).await?; - self.store_response_in_cache(&cleaned, query_embedding) - .await; - Ok(Some(cleaned)) - } - Ok(Err(e)) => Err(e.into()), - Err(_) => { - self.channel - .send("LLM request timed out. Please try again.") - .await?; - Ok(None) - } - } - } - - #[cfg(test)] - pub(in crate::agent) async fn call_llm_with_retry( - &mut self, - max_attempts: usize, - ) -> Result, crate::agent::error::AgentError> { - for attempt in 0..max_attempts { - match self.call_llm_with_timeout().await { - Ok(result) => return Ok(result), - Err(e) if e.is_context_length_error() && attempt + 1 < max_attempts => { - tracing::warn!( - attempt, - "LLM context length exceeded, compacting and retrying" - ); - self.channel - .send_status_best_effort("context too long, compacting...") - .await; - let _ = self.compact_context().await?; - self.channel.send_status_best_effort("").await; - } - Err(e) => return Err(e), - } - } - unreachable!("loop covers all attempts") - } } diff --git a/crates/zeph-core/src/agent/tool_execution/tests/parallel_and_handle_tests.rs b/crates/zeph-core/src/agent/tool_execution/tests/parallel_and_handle_tests.rs index 8fd7a44d5..a622654da 100644 --- a/crates/zeph-core/src/agent/tool_execution/tests/parallel_and_handle_tests.rs +++ b/crates/zeph-core/src/agent/tool_execution/tests/parallel_and_handle_tests.rs @@ -99,6 +99,14 @@ impl zeph_tools::executor::ToolExecutor for FailingNthExecutor { zeph_tools::tool_executor_no_inner_defaults!(); } +fn make_tool_use_request(id: &str, name: &str) -> zeph_llm::provider::ToolUseRequest { + zeph_llm::provider::ToolUseRequest { + id: id.into(), + name: name.into(), + input: serde_json::json!({}), + } +} + fn make_calls(n: usize) -> Vec { (0..n) .map(|i| ToolCall { @@ -297,11 +305,11 @@ fn last_user_query_no_user_messages_returns_empty() { } #[tokio::test] -async fn handle_tool_result_blocked_returns_false() { +async fn process_one_tool_result_blocked_is_error_with_policy_message() { use crate::agent::agent_tests::{ MockChannel, MockToolExecutor, create_test_registry, mock_provider, }; - use zeph_tools::executor::ToolError; + use zeph_llm::provider::MessagePart; let provider = mock_provider(vec![]); let channel = MockChannel::new(vec![]); @@ -309,31 +317,48 @@ async fn handle_tool_result_blocked_returns_false() { let executor = MockToolExecutor::no_tools(); let mut agent = crate::agent::Agent::new(provider, channel, registry, None, 5, executor); - let result = agent - .handle_tool_result( - "response", - Err(ToolError::Blocked { + let tc = make_tool_use_request("id-blocked", "bash"); + let mut result_parts = Vec::new(); + agent + .process_one_tool_result( + &tc, + "id-blocked", + &Instant::now(), + Err(zeph_tools::ToolError::Blocked { command: "rm -rf /".into(), }), + &mut result_parts, + &mut Vec::new(), + &mut false, + &mut None, + &mut Vec::new(), + &mut 0, + &mut zeph_sanitizer::ContentTrustLevel::Trusted, + &mut zeph_sanitizer::ContentSourceKind::ToolResult, ) .await .unwrap(); - assert!(!result); + + let is_error = result_parts + .iter() + .any(|p| matches!(p, MessagePart::ToolResult { is_error: true, .. })); + assert!(is_error, "a blocked command must be classified as an error"); assert!( agent .channel .sent_messages() .iter() - .any(|s| s.contains("blocked")) + .any(|s| s.contains("blocked")), + "the policy-blocked message must reach the channel" ); } #[tokio::test] -async fn handle_tool_result_cancelled_returns_false() { +async fn process_one_tool_result_cancelled_is_error() { use crate::agent::agent_tests::{ MockChannel, MockToolExecutor, create_test_registry, mock_provider, }; - use zeph_tools::executor::ToolError; + use zeph_llm::provider::MessagePart; let provider = mock_provider(vec![]); let channel = MockChannel::new(vec![]); @@ -341,19 +366,41 @@ async fn handle_tool_result_cancelled_returns_false() { let executor = MockToolExecutor::no_tools(); let mut agent = crate::agent::Agent::new(provider, channel, registry, None, 5, executor); - let result = agent - .handle_tool_result("response", Err(ToolError::Cancelled)) + let tc = make_tool_use_request("id-cancel", "bash"); + let mut result_parts = Vec::new(); + agent + .process_one_tool_result( + &tc, + "id-cancel", + &Instant::now(), + Err(zeph_tools::ToolError::Cancelled), + &mut result_parts, + &mut Vec::new(), + &mut false, + &mut None, + &mut Vec::new(), + &mut 0, + &mut zeph_sanitizer::ContentTrustLevel::Trusted, + &mut zeph_sanitizer::ContentSourceKind::ToolResult, + ) .await .unwrap(); - assert!(!result); + + let is_error = result_parts + .iter() + .any(|p| matches!(p, MessagePart::ToolResult { is_error: true, .. })); + assert!( + is_error, + "a cancelled tool call must be classified as an error" + ); } #[tokio::test] -async fn handle_tool_result_sandbox_violation_returns_false() { +async fn process_one_tool_result_sandbox_violation_is_error_with_sandbox_message() { use crate::agent::agent_tests::{ MockChannel, MockToolExecutor, create_test_registry, mock_provider, }; - use zeph_tools::executor::ToolError; + use zeph_llm::provider::MessagePart; let provider = mock_provider(vec![]); let channel = MockChannel::new(vec![]); @@ -361,30 +408,51 @@ async fn handle_tool_result_sandbox_violation_returns_false() { let executor = MockToolExecutor::no_tools(); let mut agent = crate::agent::Agent::new(provider, channel, registry, None, 5, executor); - let result = agent - .handle_tool_result( - "response", - Err(ToolError::SandboxViolation { + let tc = make_tool_use_request("id-sandbox", "bash"); + let mut result_parts = Vec::new(); + agent + .process_one_tool_result( + &tc, + "id-sandbox", + &Instant::now(), + Err(zeph_tools::ToolError::SandboxViolation { path: "/etc/passwd".into(), }), + &mut result_parts, + &mut Vec::new(), + &mut false, + &mut None, + &mut Vec::new(), + &mut 0, + &mut zeph_sanitizer::ContentTrustLevel::Trusted, + &mut zeph_sanitizer::ContentSourceKind::ToolResult, ) .await .unwrap(); - assert!(!result); + + let is_error = result_parts + .iter() + .any(|p| matches!(p, MessagePart::ToolResult { is_error: true, .. })); + assert!( + is_error, + "a sandbox violation must be classified as an error" + ); assert!( agent .channel .sent_messages() .iter() - .any(|s| s.contains("sandbox")) + .any(|s| s.contains("sandbox")), + "the sandbox-violation message must reach the channel" ); } #[tokio::test] -async fn handle_tool_result_none_returns_false() { +async fn process_one_tool_result_ok_none_is_success_with_no_output_marker() { use crate::agent::agent_tests::{ MockChannel, MockToolExecutor, create_test_registry, mock_provider, }; + use zeph_llm::provider::MessagePart; let provider = mock_provider(vec![]); let channel = MockChannel::new(vec![]); @@ -392,18 +460,48 @@ async fn handle_tool_result_none_returns_false() { let executor = MockToolExecutor::no_tools(); let mut agent = crate::agent::Agent::new(provider, channel, registry, None, 5, executor); - let result = agent - .handle_tool_result("response", Ok(None)) + let tc = make_tool_use_request("id-none", "bash"); + let mut result_parts = Vec::new(); + agent + .process_one_tool_result( + &tc, + "id-none", + &Instant::now(), + Ok(None), + &mut result_parts, + &mut Vec::new(), + &mut false, + &mut None, + &mut Vec::new(), + &mut 0, + &mut zeph_sanitizer::ContentTrustLevel::Trusted, + &mut zeph_sanitizer::ContentSourceKind::ToolResult, + ) .await .unwrap(); - assert!(!result); + + let (content, is_error) = result_parts + .iter() + .find_map(|p| match p { + MessagePart::ToolResult { + content, is_error, .. + } => Some((content.clone(), *is_error)), + _ => None, + }) + .expect("a ToolResult message part must be pushed"); + assert!(!is_error, "Ok(None) must not be classified as an error"); + assert!( + content.contains("no output"), + "Ok(None) must surface a no-output marker, got: {content}" + ); } #[tokio::test] -async fn handle_tool_result_with_output_returns_true() { +async fn process_one_tool_result_with_output_pushes_success_tool_result() { use crate::agent::agent_tests::{ MockChannel, MockToolExecutor, create_test_registry, mock_provider, }; + use zeph_llm::provider::MessagePart; let provider = mock_provider(vec![]); let channel = MockChannel::new(vec![]); @@ -411,6 +509,7 @@ async fn handle_tool_result_with_output_returns_true() { let executor = MockToolExecutor::no_tools(); let mut agent = crate::agent::Agent::new(provider, channel, registry, None, 5, executor); + let tc = make_tool_use_request("id-out", "bash"); let output = ToolOutput { tool_name: "bash".into(), summary: "hello from tool".into(), @@ -424,18 +523,48 @@ async fn handle_tool_result_with_output_returns_true() { claim_source: None, ..Default::default() }; - let result = agent - .handle_tool_result("response", Ok(Some(output))) + let mut result_parts = Vec::new(); + agent + .process_one_tool_result( + &tc, + "id-out", + &Instant::now(), + Ok(Some(output)), + &mut result_parts, + &mut Vec::new(), + &mut false, + &mut None, + &mut Vec::new(), + &mut 0, + &mut zeph_sanitizer::ContentTrustLevel::Trusted, + &mut zeph_sanitizer::ContentSourceKind::ToolResult, + ) .await .unwrap(); - assert!(result); + + let (content, is_error) = result_parts + .iter() + .find_map(|p| match p { + MessagePart::ToolResult { + content, is_error, .. + } => Some((content.clone(), *is_error)), + _ => None, + }) + .expect("a ToolResult message part must be pushed"); + assert!(!is_error); + assert!(content.contains("hello from tool")); } #[tokio::test] -async fn handle_tool_result_empty_output_returns_false() { +async fn process_one_tool_result_whitespace_only_output_is_still_success() { + // Unlike the removed legacy harness (which special-cased a trim()-empty summary as an + // early "no further action" skip), the production classify_tool_result path only treats + // output as an error when it contains "[error]"/"[stderr]" markers — whitespace-only + // output is not special-cased. This asserts the current, real behavior. use crate::agent::agent_tests::{ MockChannel, MockToolExecutor, create_test_registry, mock_provider, }; + use zeph_llm::provider::MessagePart; let provider = mock_provider(vec![]); let channel = MockChannel::new(vec![]); @@ -443,9 +572,10 @@ async fn handle_tool_result_empty_output_returns_false() { let executor = MockToolExecutor::no_tools(); let mut agent = crate::agent::Agent::new(provider, channel, registry, None, 5, executor); + let tc = make_tool_use_request("id-empty", "bash"); let output = ToolOutput { tool_name: "bash".into(), - summary: " ".into(), // whitespace only → considered empty + summary: " ".into(), blocks_executed: 0, diff: None, filter_stats: None, @@ -456,15 +586,36 @@ async fn handle_tool_result_empty_output_returns_false() { claim_source: None, ..Default::default() }; - let result = agent - .handle_tool_result("response", Ok(Some(output))) + let mut result_parts = Vec::new(); + agent + .process_one_tool_result( + &tc, + "id-empty", + &Instant::now(), + Ok(Some(output)), + &mut result_parts, + &mut Vec::new(), + &mut false, + &mut None, + &mut Vec::new(), + &mut 0, + &mut zeph_sanitizer::ContentTrustLevel::Trusted, + &mut zeph_sanitizer::ContentSourceKind::ToolResult, + ) .await .unwrap(); - assert!(!result); + + let is_error = result_parts + .iter() + .any(|p| matches!(p, MessagePart::ToolResult { is_error: true, .. })); + assert!( + !is_error, + "whitespace-only output must not be classified as an error" + ); } #[tokio::test] -async fn handle_tool_result_error_prefix_triggers_anomaly_error() { +async fn process_one_tool_result_error_prefix_records_tool_failure_outcome() { use crate::agent::agent_tests::{ MockChannel, MockToolExecutor, create_test_registry, mock_provider, }; @@ -474,7 +625,10 @@ async fn handle_tool_result_error_prefix_triggers_anomaly_error() { let registry = create_test_registry(); let executor = MockToolExecutor::no_tools(); let mut agent = crate::agent::Agent::new(provider, channel, registry, None, 5, executor); + // reflection_used = true so the self-reflection path is skipped + agent.services.learning_engine.mark_reflection_used(); + let tc = make_tool_use_request("id-error-prefix", "bash"); let output = ToolOutput { tool_name: "bash".into(), summary: "[error] spawn failed".into(), @@ -488,52 +642,36 @@ async fn handle_tool_result_error_prefix_triggers_anomaly_error() { claim_source: None, ..Default::default() }; - // reflection_used = true so reflection path is skipped - agent.services.learning_engine.mark_reflection_used(); - let result = agent - .handle_tool_result("response", Ok(Some(output))) + let mut pending_outcomes = Vec::new(); + agent + .process_one_tool_result( + &tc, + "id-error-prefix", + &Instant::now(), + Ok(Some(output)), + &mut Vec::new(), + &mut Vec::new(), + &mut false, + &mut None, + &mut pending_outcomes, + &mut 0, + &mut zeph_sanitizer::ContentTrustLevel::Trusted, + &mut zeph_sanitizer::ContentSourceKind::ToolResult, + ) .await .unwrap(); - // Returns true because the tool loop continues after recording failure - assert!(result); -} - -#[tokio::test] -async fn handle_tool_result_stderr_prefix_triggers_anomaly_error() { - use crate::agent::agent_tests::{ - MockChannel, MockToolExecutor, create_test_registry, mock_provider, - }; - let provider = mock_provider(vec![]); - let channel = MockChannel::new(vec![]); - let registry = create_test_registry(); - let executor = MockToolExecutor::no_tools(); - let mut agent = crate::agent::Agent::new(provider, channel, registry, None, 5, executor); - - // [stderr] prefix is produced by ShellExecutor when the child process writes to stderr. - // Prior to this fix, such output was silently classified as AnomalyOutcome::Success. - let output = ToolOutput { - tool_name: "bash".into(), - summary: "[stderr] warning: deprecated API used".into(), - blocks_executed: 1, - diff: None, - filter_stats: None, - streamed: false, - terminal_id: None, - locations: None, - raw_response: None, - claim_source: None, - ..Default::default() - }; - agent.services.learning_engine.mark_reflection_used(); - let result = agent - .handle_tool_result("response", Ok(Some(output))) - .await - .unwrap(); - // handle_tool_result returns true (tool loop continues) regardless of anomaly outcome - assert!(result); + assert!( + pending_outcomes.iter().any(|o| o.outcome == "tool_failure"), + "output containing [error] must be recorded as a tool_failure skill outcome" + ); } +// classify_tool_result's "[stderr]" branch (tool_result.rs:295) currently has no test that can +// distinguish Error from Success classification for it — a pre-existing gap, not introduced by +// this PR (the removed legacy-harness test was equally non-discriminating). Follow-up test- +// coverage issue to be filed separately. + #[tokio::test] async fn buffered_preserves_order() { use futures::StreamExt; @@ -695,189 +833,3 @@ fn inject_active_skill_env_clears_after_call() { assert!(calls[0].is_some(), "first call must set env"); assert!(calls[1].is_none(), "second call must clear env"); } - -#[tokio::test] -async fn call_llm_returns_cached_response_without_provider_call() { - use crate::agent::agent_tests::{ - MockChannel, MockToolExecutor, create_test_registry, mock_provider_streaming, - }; - use std::sync::Arc; - use zeph_llm::provider::{Message, MessageMetadata, Role}; - use zeph_memory::{ResponseCache, store::SqliteStore}; - - let channel = MockChannel::new(vec![]); - let registry = create_test_registry(); - let executor = MockToolExecutor::no_tools(); - // Streaming provider — cache must be consulted regardless of streaming support. - let provider = mock_provider_streaming(vec!["uncached response".into()]); - let mut agent = crate::agent::Agent::new(provider, channel, registry, None, 5, executor); - - // Set up a response cache with a pre-populated entry. - let store = SqliteStore::new(":memory:").await.unwrap(); - let cache = Arc::new(ResponseCache::new(store.pool().clone(), 3600)); - - // Pre-populate cache for the user message we're about to add. - let user_content = "what is 2+2?"; - let key = ResponseCache::compute_key(user_content, &agent.runtime.config.model_name); - cache - .put(&key, "cached response", "test-model") - .await - .unwrap(); - - agent.services.session.response_cache = Some(cache); - - agent.msg.messages.push(Message { - role: Role::User, - content: user_content.into(), - parts: vec![], - metadata: MessageMetadata::default(), - }); - - let result = agent.call_llm_with_timeout().await.unwrap(); - assert_eq!(result.as_deref(), Some("cached response")); - // Channel should have received the cached response - assert!( - agent - .channel - .sent_messages() - .iter() - .any(|s| s == "cached response") - ); -} - -#[tokio::test] -async fn store_response_in_cache_enables_second_call_to_return_cached() { - use crate::agent::agent_tests::{ - MockChannel, MockToolExecutor, create_test_registry, mock_provider, - }; - use std::sync::Arc; - use zeph_llm::provider::{Message, MessageMetadata, Role}; - use zeph_memory::{ResponseCache, store::SqliteStore}; - - // Non-streaming provider has one response; the second call must come from cache. - let provider = mock_provider(vec!["provider response".into()]); - let channel = MockChannel::new(vec![]); - let registry = create_test_registry(); - let executor = MockToolExecutor::no_tools(); - let mut agent = crate::agent::Agent::new(provider, channel, registry, None, 5, executor); - - let store = SqliteStore::new(":memory:").await.unwrap(); - let cache = Arc::new(ResponseCache::new(store.pool().clone(), 3600)); - agent.services.session.response_cache = Some(cache); - - agent.msg.messages.push(Message { - role: Role::User, - content: "what is 3+3?".into(), - parts: vec![], - metadata: MessageMetadata::default(), - }); - - // First call — hits provider, stores response in cache. - let first = agent.call_llm_with_timeout().await.unwrap(); - assert_eq!(first.as_deref(), Some("provider response")); - - // Second call with the same messages — must return cached value. - let second = agent.call_llm_with_timeout().await.unwrap(); - assert_eq!( - second.as_deref(), - Some("provider response"), - "second call must return cached response" - ); - - // Both first call (provider) and second call (cache hit) send via channel.send(). - let sent = agent.channel.sent_messages(); - assert!( - sent.iter().any(|s| s == "provider response"), - "provider response must have been sent via channel" - ); -} - -#[tokio::test] -async fn cache_key_stable_across_growing_history() { - use crate::agent::agent_tests::{ - MockChannel, MockToolExecutor, create_test_registry, mock_provider_streaming, - }; - use std::sync::Arc; - use zeph_llm::provider::{Message, MessageMetadata, Role}; - use zeph_memory::{ResponseCache, store::SqliteStore}; - - let provider = mock_provider_streaming(vec!["turn2 response".into()]); - let channel = MockChannel::new(vec![]); - let registry = create_test_registry(); - let executor = MockToolExecutor::no_tools(); - let mut agent = crate::agent::Agent::new(provider, channel, registry, None, 5, executor); - - let store = SqliteStore::new(":memory:").await.unwrap(); - let cache = Arc::new(ResponseCache::new(store.pool().clone(), 3600)); - - // Simulate turn 1: store a cached response for user message "hello". - let user_msg = "hello"; - let key = ResponseCache::compute_key(user_msg, &agent.runtime.config.model_name); - cache - .put(&key, "cached hello response", "test-model") - .await - .unwrap(); - agent.services.session.response_cache = Some(cache); - - // Add history from turn 1: system context + prior exchange. - agent.msg.messages.push(Message { - role: Role::Assistant, - content: "cached hello response".into(), - parts: vec![], - metadata: MessageMetadata::default(), - }); - - // Turn 2: same user message "hello" but history has grown. - agent.msg.messages.push(Message { - role: Role::User, - content: user_msg.into(), - parts: vec![], - metadata: MessageMetadata::default(), - }); - - // Must hit cache despite history growth — key is based on last user message only. - let result = agent.call_llm_with_timeout().await.unwrap(); - assert_eq!( - result.as_deref(), - Some("cached hello response"), - "cache must hit for same user message regardless of preceding history" - ); -} - -#[tokio::test] -async fn cache_skipped_when_no_user_message() { - use crate::agent::agent_tests::{ - MockChannel, MockToolExecutor, create_test_registry, mock_provider_streaming, - }; - use std::sync::Arc; - use zeph_llm::provider::{Message, MessageMetadata, Role}; - use zeph_memory::{ResponseCache, store::SqliteStore}; - - let provider = mock_provider_streaming(vec!["llm response".into()]); - let channel = MockChannel::new(vec![]); - let registry = create_test_registry(); - let executor = MockToolExecutor::no_tools(); - let mut agent = crate::agent::Agent::new(provider, channel, registry, None, 5, executor); - - let store = SqliteStore::new(":memory:").await.unwrap(); - let cache = Arc::new(ResponseCache::new(store.pool().clone(), 3600)); - agent.services.session.response_cache = Some(cache); - - // Only system/assistant messages, no user message. - agent.msg.messages.push(Message { - role: Role::System, - content: "you are helpful".into(), - parts: vec![], - metadata: MessageMetadata::default(), - }); - agent.msg.messages.push(Message { - role: Role::Assistant, - content: "hello".into(), - parts: vec![], - metadata: MessageMetadata::default(), - }); - - // Should skip cache (no user message) and call LLM. - let result = agent.call_llm_with_timeout().await.unwrap(); - assert_eq!(result.as_deref(), Some("llm response")); -} diff --git a/crates/zeph-core/src/agent/tool_execution/tests/retry_and_skill_env_tests.rs b/crates/zeph-core/src/agent/tool_execution/tests/retry_and_skill_env_tests.rs index f35ae3b6b..c4bddfdb8 100644 --- a/crates/zeph-core/src/agent/tool_execution/tests/retry_and_skill_env_tests.rs +++ b/crates/zeph-core/src/agent/tool_execution/tests/retry_and_skill_env_tests.rs @@ -1,84 +1,6 @@ // SPDX-FileCopyrightText: 2026 Andrei G // SPDX-License-Identifier: MIT OR Apache-2.0 -mod retry_tests { - use crate::agent::agent_tests::*; - use zeph_llm::LlmError; - use zeph_llm::any::AnyProvider; - use zeph_llm::mock::MockProvider; - use zeph_llm::provider::{Message, MessageMetadata, Role}; - - fn agent_with_provider(provider: AnyProvider) -> crate::agent::Agent { - let channel = MockChannel::new(vec![]); - let registry = create_test_registry(); - let executor = MockToolExecutor::no_tools(); - let mut agent = crate::agent::Agent::new(provider, channel, registry, None, 5, executor); - agent.msg.messages.push(Message { - role: Role::User, - content: "hello".into(), - parts: vec![], - metadata: MessageMetadata::default(), - }); - agent - } - - #[tokio::test] - async fn call_llm_with_retry_succeeds_on_first_attempt() { - let provider = AnyProvider::Mock(MockProvider::with_responses(vec!["ok".into()])); - let mut agent = agent_with_provider(provider); - let result = agent.call_llm_with_retry(2).await.unwrap(); - assert_eq!(result.as_deref(), Some("ok")); - } - - #[tokio::test] - async fn call_llm_with_retry_recovers_after_context_length_error() { - // First call returns ContextLengthExceeded, second succeeds. - // compact_context() is a no-op with only 1 non-system message + system prompt, - // but the retry logic itself must still re-call after compaction. - let provider = AnyProvider::Mock( - MockProvider::with_responses(vec!["recovered".into()]) - .with_errors(vec![LlmError::ContextLengthExceeded]), - ); - let mut agent = agent_with_provider(provider); - // Add context budget so compact_context can run - agent.context_manager.budget = Some(zeph_core_budget_for_test()); - let result = agent.call_llm_with_retry(2).await.unwrap(); - assert_eq!(result.as_deref(), Some("recovered")); - } - - fn zeph_core_budget_for_test() -> crate::context::ContextBudget { - crate::context::ContextBudget::new(200_000, 0.20) - } - - #[tokio::test] - async fn call_llm_with_retry_propagates_non_context_error() { - let provider = AnyProvider::Mock( - MockProvider::with_responses(vec![]) - .with_errors(vec![LlmError::Other("network error".into())]), - ); - let mut agent = agent_with_provider(provider); - let result: Result, _> = agent.call_llm_with_retry(2).await; - assert!(result.is_err()); - let err = result.unwrap_err(); - assert!(!err.is_context_length_error()); - } - - #[tokio::test] - async fn call_llm_with_retry_exhausts_all_attempts() { - // Two context length errors, max_attempts=2 — second attempt has no guard, - // so it returns the error directly. - let provider = AnyProvider::Mock(MockProvider::with_responses(vec![]).with_errors(vec![ - LlmError::ContextLengthExceeded, - LlmError::ContextLengthExceeded, - ])); - let mut agent = agent_with_provider(provider); - agent.context_manager.budget = Some(zeph_core_budget_for_test()); - let result: Result, _> = agent.call_llm_with_retry(2).await; - assert!(result.is_err()); - assert!(result.unwrap_err().is_context_length_error()); - } -} - mod retry_integration { use crate::agent::agent_tests::*; use zeph_llm::LlmError; @@ -148,12 +70,61 @@ mod retry_integration { assert!(result.is_err()); assert!(result.unwrap_err().is_context_length_error()); } + + #[tokio::test] + async fn call_chat_with_tools_retry_propagates_non_context_error() { + let provider = AnyProvider::Mock( + MockProvider::with_responses(vec![]) + .with_errors(vec![LlmError::Other("network error".into())]), + ); + let mut agent = agent_with_provider(provider); + let result: Result, _> = agent.call_chat_with_tools_retry(&no_tools(), 2).await; + assert!(result.is_err()); + assert!(!result.unwrap_err().is_context_length_error()); + } +} + +fn make_tool_use_request(id: &str, name: &str) -> zeph_llm::provider::ToolUseRequest { + zeph_llm::provider::ToolUseRequest { + id: id.into(), + name: name.into(), + input: serde_json::json!({}), + } +} + +#[allow(clippy::too_many_arguments)] +async fn call_process_one_tool_result( + agent: &mut crate::agent::Agent, + tc: &zeph_llm::provider::ToolUseRequest, + tool_call_id: &str, + started_at: &std::time::Instant, + tool_result: Result, zeph_tools::executor::ToolError>, +) -> Vec { + let mut result_parts = Vec::new(); + agent + .process_one_tool_result( + tc, + tool_call_id, + started_at, + tool_result, + &mut result_parts, + &mut Vec::new(), + &mut false, + &mut None, + &mut Vec::new(), + &mut 0, + &mut zeph_sanitizer::ContentTrustLevel::Trusted, + &mut zeph_sanitizer::ContentSourceKind::ToolResult, + ) + .await + .unwrap(); + result_parts } // Regression tests for issue #1003: tool output must reach all channel types // regardless of whether the tool streamed its output. #[tokio::test] -async fn handle_tool_result_sends_output_when_streamed_true() { +async fn process_one_tool_result_sends_output_when_streamed_true() { use crate::agent::agent_tests::{ MockChannel, MockToolExecutor, create_test_registry, mock_provider, }; @@ -165,6 +136,7 @@ async fn handle_tool_result_sends_output_when_streamed_true() { let executor = MockToolExecutor::no_tools(); let mut agent = crate::agent::Agent::new(provider, channel, registry, None, 5, executor); + let tc = make_tool_use_request("id-streamed", "bash"); let output = ToolOutput { tool_name: "bash".into(), summary: "streamed content".into(), @@ -178,10 +150,14 @@ async fn handle_tool_result_sends_output_when_streamed_true() { claim_source: None, ..Default::default() }; - agent - .handle_tool_result("response", Ok(Some(output))) - .await - .unwrap(); + call_process_one_tool_result( + &mut agent, + &tc, + "id-streamed", + &std::time::Instant::now(), + Ok(Some(output)), + ) + .await; let sent = agent.channel.sent_messages(); assert!( @@ -191,7 +167,7 @@ async fn handle_tool_result_sends_output_when_streamed_true() { } #[tokio::test] -async fn handle_tool_result_fenced_emits_tool_start_then_output_via_loopback() { +async fn process_one_tool_result_emits_tool_start_then_output_via_loopback() { use crate::agent::agent_tests::{MockToolExecutor, create_test_registry, mock_provider}; use crate::channel::{LoopbackChannel, LoopbackEvent}; use zeph_tools::executor::ToolOutput; @@ -202,6 +178,23 @@ async fn handle_tool_result_fenced_emits_tool_start_then_output_via_loopback() { let executor = MockToolExecutor::no_tools(); let mut agent = crate::agent::Agent::new(provider, loopback, registry, None, 5, executor); + // Mirrors the real per-tier dispatch sequence in tier_loop.rs: ToolStart is stamped and + // sent before the tool executes, then process_one_tool_result handles the result — the + // two are separate production call sites, not a single legacy convenience wrapper. + let tc = make_tool_use_request("call-grep", "grep"); + let tool_call_id = "call-grep".to_owned(); + let mut started_ats = [std::time::Instant::now()]; + agent + .stamp_and_send_tier_start( + &[0], + std::slice::from_ref(&tc), + std::slice::from_ref(&tool_call_id), + &mut started_ats, + &std::collections::HashSet::new(), + ) + .await + .unwrap(); + let output = ToolOutput { tool_name: "grep".into(), summary: "match found".into(), @@ -215,10 +208,14 @@ async fn handle_tool_result_fenced_emits_tool_start_then_output_via_loopback() { claim_source: None, ..Default::default() }; - agent - .handle_tool_result("response", Ok(Some(output))) - .await - .unwrap(); + call_process_one_tool_result( + &mut agent, + &tc, + &tool_call_id, + &started_ats[0], + Ok(Some(output)), + ) + .await; drop(agent); @@ -271,7 +268,7 @@ async fn handle_tool_result_fenced_emits_tool_start_then_output_via_loopback() { } #[tokio::test] -async fn handle_tool_result_locations_propagated_to_loopback_event() { +async fn process_one_tool_result_locations_propagated_to_loopback_event() { use crate::agent::agent_tests::{MockToolExecutor, create_test_registry, mock_provider}; use crate::channel::{LoopbackChannel, LoopbackEvent}; use zeph_tools::executor::ToolOutput; @@ -282,6 +279,7 @@ async fn handle_tool_result_locations_propagated_to_loopback_event() { let executor = MockToolExecutor::no_tools(); let mut agent = crate::agent::Agent::new(provider, loopback, registry, None, 5, executor); + let tc = make_tool_use_request("id-locations", "read_file"); let output = ToolOutput { tool_name: "read_file".into(), summary: "file content".into(), @@ -295,10 +293,14 @@ async fn handle_tool_result_locations_propagated_to_loopback_event() { claim_source: None, ..Default::default() }; - agent - .handle_tool_result("response", Ok(Some(output))) - .await - .unwrap(); + call_process_one_tool_result( + &mut agent, + &tc, + "id-locations", + &std::time::Instant::now(), + Ok(Some(output)), + ) + .await; drop(agent); let mut events = Vec::new(); @@ -325,7 +327,7 @@ async fn handle_tool_result_locations_propagated_to_loopback_event() { // `send_tool_output`, which caused newlines inside the output to be lost in ACP consumers // that read `terminal_output.data` or `raw_output` as plain text. #[tokio::test] -async fn handle_tool_result_display_is_raw_body_not_markdown_wrapped() { +async fn process_one_tool_result_display_is_raw_body_not_markdown_wrapped() { use crate::agent::agent_tests::{MockToolExecutor, create_test_registry, mock_provider}; use crate::channel::{LoopbackChannel, LoopbackEvent}; use zeph_tools::executor::ToolOutput; @@ -336,6 +338,7 @@ async fn handle_tool_result_display_is_raw_body_not_markdown_wrapped() { let executor = MockToolExecutor::no_tools(); let mut agent = crate::agent::Agent::new(provider, loopback, registry, None, 5, executor); + let tc = make_tool_use_request("id-raw-body", "bash"); let output = ToolOutput { tool_name: "bash".into(), summary: "line1\nline2\nline3".into(), @@ -349,10 +352,14 @@ async fn handle_tool_result_display_is_raw_body_not_markdown_wrapped() { claim_source: None, ..Default::default() }; - agent - .handle_tool_result("response", Ok(Some(output))) - .await - .unwrap(); + call_process_one_tool_result( + &mut agent, + &tc, + "id-raw-body", + &std::time::Instant::now(), + Ok(Some(output)), + ) + .await; drop(agent); let mut events = Vec::new(); diff --git a/crates/zeph-core/src/agent/tool_execution/tests/sanitize_and_native_tests.rs b/crates/zeph-core/src/agent/tool_execution/tests/sanitize_and_native_tests.rs index c4da5de0f..cd0c25188 100644 --- a/crates/zeph-core/src/agent/tool_execution/tests/sanitize_and_native_tests.rs +++ b/crates/zeph-core/src/agent/tool_execution/tests/sanitize_and_native_tests.rs @@ -1090,6 +1090,120 @@ async fn native_tool_use_cache_stores_only_text_responses() { ); } +#[tokio::test] +#[allow(clippy::large_futures)] +async fn native_tool_use_cache_key_stable_across_growing_history() { + use crate::agent::agent_tests::*; + use std::sync::Arc; + use zeph_llm::any::AnyProvider; + use zeph_llm::mock::MockProvider; + use zeph_llm::provider::{ChatResponse, Message, MessageMetadata, Role}; + use zeph_memory::{ResponseCache, store::SqliteStore}; + + let (mock, call_count) = MockProvider::with_responses(vec![]) + .with_tool_use(vec![ChatResponse::Text("turn2 response".into())]); + let provider = AnyProvider::Mock(mock); + + let channel = MockChannel::new(vec![]); + let registry = create_test_registry(); + let executor = MockToolExecutor::no_tools(); + let mut agent = crate::agent::Agent::new(provider, channel, registry, None, 5, executor); + + let store = SqliteStore::new(":memory:").await.unwrap(); + let cache = Arc::new(ResponseCache::new(store.pool().clone(), 3600)); + + // Simulate turn 1: store a cached response for user message "hello". + let user_msg = "hello"; + let key = ResponseCache::compute_key(user_msg, &agent.runtime.config.model_name); + cache + .put(&key, "cached hello response", "test-model") + .await + .unwrap(); + agent.services.session.response_cache = Some(cache); + + // Add history from turn 1: prior exchange. + agent.msg.messages.push(Message { + role: Role::Assistant, + content: "cached hello response".into(), + parts: vec![], + metadata: MessageMetadata::default(), + }); + + // Turn 2: same user message "hello" but history has grown. + agent.msg.messages.push(Message { + role: Role::User, + content: user_msg.into(), + parts: vec![], + metadata: MessageMetadata::default(), + }); + + // Must hit cache despite history growth — key is based on last user message only. + agent.process_response().await.unwrap(); + + assert_eq!( + *call_count.lock().unwrap(), + 0, + "provider must not be called: cache key is based on the last user message only" + ); + let sent = agent.channel.sent_messages(); + assert!( + sent.iter().any(|s| s == "cached hello response"), + "cached response must be sent despite grown history; got: {sent:?}" + ); +} + +#[tokio::test] +#[allow(clippy::large_futures)] +async fn native_tool_use_cache_skipped_when_no_user_message() { + use crate::agent::agent_tests::*; + use std::sync::Arc; + use zeph_llm::any::AnyProvider; + use zeph_llm::mock::MockProvider; + use zeph_llm::provider::{ChatResponse, Message, MessageMetadata, Role}; + use zeph_memory::{ResponseCache, store::SqliteStore}; + + let (mock, call_count) = MockProvider::with_responses(vec![]) + .with_tool_use(vec![ChatResponse::Text("llm response".into())]); + let provider = AnyProvider::Mock(mock); + + let channel = MockChannel::new(vec![]); + let registry = create_test_registry(); + let executor = MockToolExecutor::no_tools(); + let mut agent = crate::agent::Agent::new(provider, channel, registry, None, 5, executor); + + let store = SqliteStore::new(":memory:").await.unwrap(); + let cache = Arc::new(ResponseCache::new(store.pool().clone(), 3600)); + agent.services.session.response_cache = Some(cache); + + // Only system/assistant messages, no user message. + agent.msg.messages.push(Message { + role: Role::System, + content: "you are helpful".into(), + parts: vec![], + metadata: MessageMetadata::default(), + }); + agent.msg.messages.push(Message { + role: Role::Assistant, + content: "hello".into(), + parts: vec![], + metadata: MessageMetadata::default(), + }); + + // Should skip cache (no user message) and call the LLM. + agent.process_response().await.unwrap(); + + assert_eq!( + *call_count.lock().unwrap(), + 1, + "provider must be called when there is no user message to key the cache on" + ); + let sent = agent.channel.sent_messages(); + assert!( + sent.iter().any(|s| s == "llm response"), + "llm response must be sent when cache is skipped; got: {sent:?}" + ); +} + // ── handle_native_tool_calls retry (RF-2) ──────────────────────────────── /// Returns `Transient` io error for the first `fail_times` calls, then success. diff --git a/crates/zeph-core/src/agent/tool_execution/tool_result.rs b/crates/zeph-core/src/agent/tool_execution/tool_result.rs index 3aae938c9..4b2154eff 100644 --- a/crates/zeph-core/src/agent/tool_execution/tool_result.rs +++ b/crates/zeph-core/src/agent/tool_execution/tool_result.rs @@ -656,307 +656,4 @@ impl Agent { .last_tool_per_skill .insert(skill_name, tool_name.to_owned()); } - - // Legacy sequential-dispatch harness, superseded in production by - // `process_one_tool_result`/`handle_confirmation_phase` (tier_loop.rs), which processes - // every tool result — including confirmation-resolved ones — through the same batched path. - // Retained `#[cfg(test)]`-only to keep exercising `sanitize_tool_output`/ - // `maybe_summarize_tool_output`/`dump_tool_output` via a simpler single-call harness; not - // dead in the sense of untested, but unreachable from any production call site (#6364). - #[cfg(test)] - pub(super) async fn handle_tool_result( - &mut self, - response: &str, - result: Result, zeph_tools::executor::ToolError>, - ) -> Result { - use zeph_sanitizer::{ContentSource, ContentSourceKind}; - use zeph_tools::executor::ToolError; - match result { - Ok(Some(output)) => self.process_successful_tool_output(output).await, - Ok(None) => { - self.record_skill_outcomes("success", None, None).await; - self.record_anomaly_outcome(AnomalyOutcome::Success).await?; - Ok(false) - } - Err(ToolError::Blocked { command }) => { - tracing::warn!("blocked command: {command}"); - self.channel - .send("This command is blocked by security policy.") - .await?; - self.record_anomaly_outcome(AnomalyOutcome::Blocked).await?; - Ok(false) - } - Err(ToolError::ConfirmationRequired { command }) => { - self.handle_confirmation_required(response, &command).await - } - Err(ToolError::Cancelled) => { - tracing::info!("tool execution cancelled"); - self.update_metrics(|m| m.cancellations += 1); - self.channel.send("[Cancelled]").await?; - Ok(false) - } - Err(ToolError::SandboxViolation { path }) => { - tracing::warn!("sandbox violation: {path}"); - self.channel - .send("Command targets a path outside the sandbox.") - .await?; - self.record_anomaly_outcome(AnomalyOutcome::Error).await?; - Ok(false) - } - Err(e) => { - let category = e.category(); - let err_str = format!("{e:#}"); - tracing::error!("tool execution error: {err_str}"); - if let Some(ref d) = self.runtime.debug.debug_dumper { - d.dump_tool_error("legacy", &e); - } - let kind = FailureKind::from(category); - let sanitized_err = self - .services - .security - .sanitizer - .sanitize(&err_str, ContentSource::new(ContentSourceKind::McpResponse)) - .body; - self.record_skill_outcomes("tool_failure", Some(&err_str), Some(kind.as_str())) - .await; - self.record_anomaly_outcome(AnomalyOutcome::Error).await?; - - if !self.services.learning_engine.was_reflection_used() - && self.attempt_self_reflection(&sanitized_err, "").await? - { - return Ok(false); - } - - self.channel - .send("Tool execution failed. Please try a different approach.") - .await?; - Ok(false) - } - } - } - - /// Record skill learning outcomes for a tool output and optionally trigger self-reflection. - /// - /// Returns `Ok(true)` when the caller should return early (reflection consumed the turn), - /// `Ok(false)` to continue, or `Err` on a hard error. - #[cfg(test)] - async fn record_tool_output_outcome( - &mut self, - output: &zeph_tools::executor::ToolOutput, - ) -> Result { - if let Some(ref fs) = output.filter_stats { - self.record_filter_metrics(fs); - } - if output.summary.trim().is_empty() { - tracing::warn!("tool execution returned empty output"); - self.record_skill_outcomes("success", None, None).await; - return Ok(true); - } - if output.summary.contains("[error]") || output.summary.contains("[exit code") { - let kind = FailureKind::from_error(&output.summary); - self.record_skill_outcomes("tool_failure", Some(&output.summary), Some(kind.as_str())) - .await; - if !self.services.learning_engine.was_reflection_used() - && self - .attempt_self_reflection(&output.summary, &output.summary) - .await? - { - return Ok(true); - } - } else { - self.record_skill_outcomes("success", None, None).await; - } - Ok(false) - } - - // This function's `dump_tool_output` call below is one of the two legacy sites whose - // green `#[cfg(test)]` coverage gave false confidence that `dump_tool_output` was reachable - // in production — it wasn't (#6364). See the retention note above `handle_tool_result`. - #[cfg(test)] - async fn process_successful_tool_output( - &mut self, - output: zeph_tools::executor::ToolOutput, - ) -> Result { - use crate::agent::format_tool_output; - use crate::channel::{ToolOutputEvent, ToolStartEvent}; - use zeph_llm::provider::{Message, MessagePart, Role}; - - if self.record_tool_output_outcome(&output).await? { - return Ok(false); - } - - let tool_call_id = uuid::Uuid::new_v4().to_string(); - let tool_started_at = std::time::Instant::now(); - self.channel - .send_tool_start(ToolStartEvent { - tool_name: output.tool_name.clone(), - tool_call_id: tool_call_id.clone(), - params: None, - parent_tool_use_id: self.services.session.parent_tool_use_id.clone(), - started_at: std::time::Instant::now(), - speculative: false, - sandbox_profile: None, - is_mcp: false, - }) - .await?; - if let Some(ref d) = self.runtime.debug.debug_dumper { - let dump_content = if self.services.security.pii_filter.is_enabled() { - self.services - .security - .pii_filter - .scrub(&output.summary) - .into_owned() - } else { - output.summary.clone() - }; - d.dump_tool_output(output.tool_name.as_str(), &dump_content); - } - let processed = self - .maybe_summarize_tool_output(&output.summary, output.max_result_size_chars) - .await; - let body = if let Some(ref fs) = output.filter_stats - && fs.filtered_chars < fs.raw_chars - { - format!( - "{}\n{processed}", - fs.format_inline(output.tool_name.as_str()) - ) - } else { - processed.clone() - }; - let filter_stats_inline = output.filter_stats.as_ref().and_then(|fs| { - (fs.filtered_chars < fs.raw_chars).then(|| fs.format_inline(output.tool_name.as_str())) - }); - let formatted_output = format_tool_output(output.tool_name.as_str(), &body); - self.channel - .send_tool_output(ToolOutputEvent { - tool_name: output.tool_name.clone(), - display: self.maybe_redact(&body).to_string(), - diff: None, - filter_stats: filter_stats_inline, - kept_lines: None, - locations: output.locations, - tool_call_id: tool_call_id.clone(), - terminal_id: None, - is_error: false, - parent_tool_use_id: self.services.session.parent_tool_use_id.clone(), - raw_response: None, - started_at: Some(tool_started_at), - }) - .await?; - - let (llm_body, has_injection_flags, _kind, _trust_level) = self - .sanitize_tool_output(&processed, output.tool_name.as_str()) - .await; - let user_msg = Message::from_parts( - Role::User, - vec![MessagePart::ToolOutput { - tool_name: output.tool_name.clone(), - body: llm_body, - compacted_at: None, - }], - ); - self.persist_message( - Role::User, - &formatted_output, - &user_msg.parts, - has_injection_flags || !self.services.security.flagged_urls.is_empty(), - ) - .await; - self.push_message(user_msg); - let outcome = if output.summary.contains("[error]") || output.summary.contains("[stderr]") { - AnomalyOutcome::Error - } else { - AnomalyOutcome::Success - }; - self.record_anomaly_outcome(outcome).await?; - Ok(true) - } - - // This function's `dump_tool_output` call below is the other legacy site whose green - // `#[cfg(test)]` coverage gave false confidence that `dump_tool_output` was reachable in - // production — it wasn't (#6364). See the retention note above `handle_tool_result`. - #[cfg(test)] - async fn handle_confirmation_required( - &mut self, - response: &str, - command: &str, - ) -> Result { - use crate::agent::format_tool_output; - use crate::channel::{ToolOutputEvent, ToolStartEvent}; - use zeph_llm::provider::{Message, MessagePart, Role}; - let prompt = format!("Allow command: {command}?"); - if self.channel.confirm(&prompt).await? { - if let Ok(Some(out)) = self.tool_executor.execute_confirmed_erased(response).await { - let confirmed_tool_call_id = uuid::Uuid::new_v4().to_string(); - let confirmed_started_at = std::time::Instant::now(); - self.channel - .send_tool_start(ToolStartEvent { - tool_name: out.tool_name.clone(), - tool_call_id: confirmed_tool_call_id.clone(), - params: None, - parent_tool_use_id: self.services.session.parent_tool_use_id.clone(), - started_at: std::time::Instant::now(), - speculative: false, - sandbox_profile: None, - is_mcp: false, - }) - .await?; - if let Some(ref d) = self.runtime.debug.debug_dumper { - let dump_content = if self.services.security.pii_filter.is_enabled() { - self.services - .security - .pii_filter - .scrub(&out.summary) - .into_owned() - } else { - out.summary.clone() - }; - d.dump_tool_output(out.tool_name.as_str(), &dump_content); - } - let processed = self - .maybe_summarize_tool_output(&out.summary, out.max_result_size_chars) - .await; - let formatted = format_tool_output(out.tool_name.as_str(), &processed); - self.channel - .send_tool_output(ToolOutputEvent { - tool_name: out.tool_name.clone(), - display: self.maybe_redact(&processed).to_string(), - diff: None, - filter_stats: None, - kept_lines: None, - locations: out.locations, - tool_call_id: confirmed_tool_call_id.clone(), - terminal_id: None, - is_error: false, - parent_tool_use_id: self.services.session.parent_tool_use_id.clone(), - raw_response: None, - started_at: Some(confirmed_started_at), - }) - .await?; - let (llm_body, has_injection_flags, _kind, _trust_level) = self - .sanitize_tool_output(&processed, out.tool_name.as_str()) - .await; - let confirmed_msg = Message::from_parts( - Role::User, - vec![MessagePart::ToolOutput { - tool_name: out.tool_name.clone(), - body: llm_body, - compacted_at: None, - }], - ); - self.persist_message( - Role::User, - &formatted, - &confirmed_msg.parts, - has_injection_flags || !self.services.security.flagged_urls.is_empty(), - ) - .await; - self.push_message(confirmed_msg); - } - } else { - self.channel.send("Command cancelled.").await?; - } - Ok(false) - } }