diff --git a/CHANGELOG.md b/CHANGELOG.md index 340310436..a33217b18 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -28,6 +28,22 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). `IndexMcpServer` always walked the full process working directory instead (#6129). Extracted the existing resolution logic into a shared `resolve_workspace_root` helper so both call sites resolve `workspace_root` identically. +- `zeph-core`: `apply_tier_results` processed each tool result's `RuntimeLayer::after_tool` + chain and `PostToolUse` hook firing sequentially, one index at a time, even though the + tier's tool execution itself already runs bounded-parallel (#6128). A tier with many + `PostToolUse`-matching calls paid N sequential subprocess spawns after already paying for + parallel tool execution. The layer/hook phase now runs concurrently across a tier's + indices via `futures::future::join_all`, bounded by the same `max_parallel` semaphore the + tier's tool execution uses. +- `zeph-core`: `MetricsBridge::WATCHED_SPANS` (profiling feature) named three spans + (`agent.prepare_context`, `agent.tool_loop`, `agent.persist_message`) that never matched any + real `tracing` span, silently making their `on_close` handling dead code — only `llm.chat` + was ever observed (#6111). Renamed the first two to their real span names + (`core.context.prepare_context`, `core.tool.native_loop`). `agent.persist_message` is + intentionally left unwatched: its real span (`core.persist.persist_message`) fires 7+ times + per turn, not once, so bridging it would report the wrong (last) call's duration instead of + the first user-message persist that `TurnTimings::persist_message_ms` is meant to measure; + that field stays manually-timed only. - `zeph-tui`: closed three independent dispatch/state gaps (#6061, #5984, #5983). - The task-registry overlay's supervisor-unavailable fallback (`render_subagents_slot`) drew its "supervisor not available" message directly into the shared subagents-slot diff --git a/crates/zeph-core/src/agent/tool_execution/tests/apply_tier_results_tests.rs b/crates/zeph-core/src/agent/tool_execution/tests/apply_tier_results_tests.rs new file mode 100644 index 000000000..fd99cc7b4 --- /dev/null +++ b/crates/zeph-core/src/agent/tool_execution/tests/apply_tier_results_tests.rs @@ -0,0 +1,135 @@ +// SPDX-FileCopyrightText: 2026 Andrei G +// SPDX-License-Identifier: MIT OR Apache-2.0 + +//! Tests for #6128: `apply_tier_results` must run `RuntimeLayer::after_tool` and `PostToolUse` +//! hook firing concurrently across a tier's tool-result indices, not serially. + +use std::future::Future; +use std::pin::Pin; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; + +use zeph_llm::provider::ToolUseRequest; +use zeph_tools::ToolError; +use zeph_tools::executor::{ToolCall, ToolOutput}; + +use crate::agent::agent_tests::{ + MockChannel, MockToolExecutor, create_test_registry, mock_provider, +}; +use crate::runtime_layer::{LayerContext, RuntimeLayer}; + +/// `RuntimeLayer` whose `after_tool` sleeps and records every `tool_call_id` it observed, so +/// tests can assert both concurrency (elapsed time) and per-index correctness (every index seen +/// exactly once). +struct DelayRecordingLayer { + delay: Duration, + seen: Arc>>, +} + +impl RuntimeLayer for DelayRecordingLayer { + fn after_tool<'a>( + &'a self, + _ctx: &'a LayerContext<'_>, + call: &'a ToolCall, + _result: &'a Result, ToolError>, + ) -> Pin + Send + 'a>> { + Box::pin(async move { + tokio::time::sleep(self.delay).await; + self.seen.lock().unwrap().push(call.tool_call_id.clone()); + }) + } +} + +fn make_tool_use_request(id: &str, name: &str) -> ToolUseRequest { + ToolUseRequest { + id: id.into(), + name: name.into(), + input: serde_json::json!({}), + } +} + +/// N tool results land in a single tier, each with a `RuntimeLayer::after_tool` that sleeps. +/// Serial processing (the pre-#6128 behavior) would take N * delay; concurrent processing +/// should stay close to a single delay regardless of N. +#[tokio::test] +async fn after_tool_hooks_run_concurrently_across_tier_indices() { + let n = 5; + let delay = Duration::from_millis(60); + let seen: Arc>> = Arc::new(Mutex::new(Vec::new())); + + let provider = mock_provider(vec![]); + let channel = MockChannel::new(vec![]); + let registry = create_test_registry(); + let executor = MockToolExecutor::new((0..n).map(|_| Ok(None)).collect()); + let mut agent = crate::agent::Agent::new(provider, channel, registry, None, 5, executor); + agent.runtime.config.timeouts.max_parallel_tools = n; + agent + .runtime + .config + .layers + .push(Arc::new(DelayRecordingLayer { + delay, + seen: Arc::clone(&seen), + })); + + let tool_calls: Vec = (0..n) + .map(|i| make_tool_use_request(&format!("id-{i}"), "noop")) + .collect(); + + let start = Instant::now(); + agent + .handle_native_tool_calls(None, &tool_calls) + .await + .unwrap(); + let elapsed = start.elapsed(); + + let seen = seen.lock().unwrap(); + assert_eq!( + seen.len(), + n, + "after_tool must run exactly once for every tool result in the tier" + ); + let unique: std::collections::HashSet<&String> = seen.iter().collect(); + assert_eq!( + unique.len(), + n, + "each tool result's after_tool call must carry its own tool_call_id, not a shared one" + ); + assert!( + elapsed < delay * u32::try_from(n).unwrap(), + "after_tool hooks appear to have run serially: took {elapsed:?} for {n} x {delay:?}" + ); +} + +/// A tier with no `RuntimeLayer`s and no `PostToolUse` hooks configured must not pay any +/// concurrency-machinery overhead — regression guard for the early-return path added in #6128. +#[tokio::test] +async fn apply_tier_results_no_layers_no_hooks_still_writes_all_results() { + let n = 3; + let provider = mock_provider(vec![]); + let channel = MockChannel::new(vec![]); + let registry = create_test_registry(); + let executor = MockToolExecutor::new((0..n).map(|_| Ok(None)).collect()); + let mut agent = crate::agent::Agent::new(provider, channel, registry, None, 5, executor); + + let tool_calls: Vec = (0..n) + .map(|i| make_tool_use_request(&format!("id-{i}"), "noop")) + .collect(); + + agent + .handle_native_tool_calls(None, &tool_calls) + .await + .unwrap(); + + let tool_result_count = agent + .msg + .messages + .iter() + .flat_map(|m| m.parts.iter()) + .filter(|p| matches!(p, zeph_llm::provider::MessagePart::ToolResult { .. })) + .count(); + assert_eq!( + tool_result_count, n, + "every tool call must still get a persisted ToolResult when no layers/hooks are configured" + ); +} diff --git a/crates/zeph-core/src/agent/tool_execution/tests/mod.rs b/crates/zeph-core/src/agent/tool_execution/tests/mod.rs index 9ea519a67..88e9c5228 100644 --- a/crates/zeph-core/src/agent/tool_execution/tests/mod.rs +++ b/crates/zeph-core/src/agent/tool_execution/tests/mod.rs @@ -1,6 +1,7 @@ // SPDX-FileCopyrightText: 2026 Andrei G // SPDX-License-Identifier: MIT OR Apache-2.0 +mod apply_tier_results_tests; mod boundary_and_classifier_tests; mod focus_tests; mod hook_block_cap_tests; diff --git a/crates/zeph-core/src/agent/tool_execution/tier_loop.rs b/crates/zeph-core/src/agent/tool_execution/tier_loop.rs index 2407bd23b..a440d31a4 100644 --- a/crates/zeph-core/src/agent/tool_execution/tier_loop.rs +++ b/crates/zeph-core/src/agent/tool_execution/tier_loop.rs @@ -1410,6 +1410,7 @@ impl Agent { mcp_tool_ids, args_hashes, tool_started_ats, + max_parallel, &mut failed_ids, &mut tool_results, ) @@ -2215,10 +2216,17 @@ impl Agent { mcp_tool_ids: &std::collections::HashSet, args_hashes: &[u64], tool_started_ats: &[std::time::Instant], + max_parallel: usize, failed_ids: &mut std::collections::HashSet, tool_results: &mut [Result, zeph_tools::ToolError>], ) { - for (idx, mut result) in indices.into_iter().zip(tier_results) { + // Phase 1: sequential bookkeeping (dependency-graph, result cache, failed_ids). Cheap, + // synchronous, and needs `&mut self` — kept sequential rather than folded into phase 2. + let mut pending: Vec<( + usize, + Result, zeph_tools::ToolError>, + )> = Vec::with_capacity(indices.len()); + for (idx, result) in indices.into_iter().zip(tier_results) { // IMP-02: Err(_) covers all error variants including ConfirmationRequired. // Ok(Some(out)) with "[error]" prefix covers synthetic/blocked results. let is_failed = match &result { @@ -2258,55 +2266,75 @@ impl Agent { .insert(tool_calls[idx].name.to_string()); } - // RuntimeLayer after_tool hooks. - if !self.runtime.config.layers.is_empty() { - let conv_id_str = self - .services - .memory - .persistence - .conversation_id - .map(|id| id.0.to_string()); - let ctx = crate::runtime_layer::LayerContext { - conversation_id: conv_id_str.as_deref(), - turn_number: u32::try_from(self.services.sidequest.turn_counter) - .unwrap_or(u32::MAX), - }; - for layer in &self.runtime.config.layers { + pending.push((idx, result)); + } + + let layers_empty = self.runtime.config.layers.is_empty(); + let post_hooks = self.services.session.hooks_config.post_tool_use.clone(); + if layers_empty && post_hooks.is_empty() { + for (idx, result) in pending { + tool_results[idx] = result; + } + return; + } + + // Phase 2: RuntimeLayer::after_tool and PostToolUse hook firing per index. Unlike + // process_one_tool_result (which must stay strictly sequential to preserve + // OpenAI/Claude message-ordering — see the comment above its call site), there is no + // ordering constraint between different tool-result indices here: each index's layer + // chain and hook firing only observes/mutates that index's own result. Run them + // concurrently, bounded by the same `max_parallel` semaphore the tier's tool execution + // itself already uses, so a tier with many hook-matching calls doesn't serialize N + // subprocess spawns after already paying for bounded-parallel tool execution (#6128). + let conv_id_str = self + .services + .memory + .persistence + .conversation_id + .map(|id| id.0.to_string()); + let ctx = crate::runtime_layer::LayerContext { + conversation_id: conv_id_str.as_deref(), + turn_number: u32::try_from(self.services.sidequest.turn_counter).unwrap_or(u32::MAX), + }; + let dispatch = self.mcp_dispatch(); + let mcp: Option<&dyn zeph_subagent::McpDispatch> = dispatch + .as_ref() + .map(|d| d as &dyn zeph_subagent::McpDispatch); + let semaphore = std::sync::Arc::new(tokio::sync::Semaphore::new(max_parallel.max(1))); + let layers = &self.runtime.config.layers; + // Reborrow as plain references (`&T` is `Copy`) so the `FnMut` closure below can + // capture them by value on every invocation instead of moving the owned originals. + let ctx_ref = &ctx; + let conv_id = conv_id_str.as_deref(); + + let futs = pending.into_iter().map(|(idx, result)| { + let sem = std::sync::Arc::clone(&semaphore); + let tc = &tool_calls[idx]; + let call = &calls[idx]; + let matched: Vec<&zeph_config::HookDef> = + zeph_subagent::matching_hooks(&post_hooks, tc.name.as_str()); + async move { + let _permit = sem.acquire().await; + let mut result = result; + + // RuntimeLayer after_tool hooks. + for layer in layers { let hook_result = - std::panic::AssertUnwindSafe(layer.after_tool(&ctx, &calls[idx], &result)) + std::panic::AssertUnwindSafe(layer.after_tool(ctx_ref, call, &result)) .catch_unwind() .await; if hook_result.is_err() { tracing::warn!("RuntimeLayer::after_tool panicked, continuing"); } } - } - // Fire PostToolUse hooks after the tool result is available (fail-open). - let post_hooks = self.services.session.hooks_config.post_tool_use.clone(); - if !post_hooks.is_empty() { - let matched: Vec<&zeph_config::HookDef> = - zeph_subagent::matching_hooks(&post_hooks, tool_calls[idx].name.as_str()); + // Fire PostToolUse hooks after the tool result is available (fail-open). if !matched.is_empty() { - let conv_id_str = self - .services - .memory - .persistence - .conversation_id - .map(|id| id.0.to_string()); - let mut env = make_tool_hook_env( - tool_calls[idx].name.as_str(), - &tool_calls[idx].input, - conv_id_str.as_deref(), - ); + let mut env = make_tool_hook_env(tc.name.as_str(), &tc.input, conv_id); let duration_ms = u64::try_from(tool_started_ats[idx].elapsed().as_millis()) .unwrap_or(u64::MAX); env.insert("ZEPH_TOOL_DURATION_MS".to_owned(), duration_ms.to_string()); let owned: Vec = matched.into_iter().cloned().collect(); - let dispatch = self.mcp_dispatch(); - let mcp: Option<&dyn zeph_subagent::McpDispatch> = dispatch - .as_ref() - .map(|d| d as &dyn zeph_subagent::McpDispatch); // Build stdin JSON context for the hook process. let tool_output_text = match &result { @@ -2318,13 +2346,13 @@ impl Agent { _ => None, }; let hook_input = zeph_subagent::PostToolUseHookInput { - tool_name: tool_calls[idx].name.as_str(), - tool_args: &tool_calls[idx].input, - session_id: conv_id_str.as_deref(), + tool_name: tc.name.as_str(), + tool_args: &tc.input, + session_id: conv_id, duration_ms, tool_output: tool_output_text, tool_error: tool_error_text.as_deref(), - agent_id: conv_id_str.as_deref(), + agent_id: conv_id, agent_type: "main", }; let stdin_bytes = serde_json::to_vec(&hook_input).ok(); @@ -2337,7 +2365,7 @@ impl Agent { ) .instrument(tracing::info_span!( "core.hooks.post_tool_use", - tool = %tool_calls[idx].name + tool = %tc.name )) .await { @@ -2346,7 +2374,7 @@ impl Agent { // Apply hook-requested output substitution. if let Ok(Some(ref mut out)) = result { tracing::debug!( - tool = %tool_calls[idx].name, + tool = %tc.name, original_len = out.summary.len(), replacement_len = replacement.len(), "PostToolUse hook replaced tool output" @@ -2358,14 +2386,18 @@ impl Agent { Err(e) => { tracing::warn!( error = %e, - tool = %tool_calls[idx].name, + tool = %tc.name, "PostToolUse hook failed" ); } } } + + (idx, result) } + }); + for (idx, result) in futures::future::join_all(futs).await { tool_results[idx] = result; } } diff --git a/crates/zeph-core/src/agent/utils.rs b/crates/zeph-core/src/agent/utils.rs index 025ff3fb3..4036a4f7a 100644 --- a/crates/zeph-core/src/agent/utils.rs +++ b/crates/zeph-core/src/agent/utils.rs @@ -197,9 +197,9 @@ impl Agent { if mask & crate::metrics_bridge::TimingField::ToolExec.bridge_bit() != 0 { timings.tool_exec_ms = m.last_turn_timings.tool_exec_ms; } - if mask & crate::metrics_bridge::TimingField::PersistMessage.bridge_bit() != 0 { - timings.persist_message_ms = m.last_turn_timings.persist_message_ms; - } + // persist_message_ms is intentionally never bridged (#6111) — its real span fires + // 7+ times per turn, not once, so `timings.persist_message_ms` always keeps the + // manual `Instant::now()` value computed in `agent/mod.rs`. m.bridge_timings_written = 0; }); diff --git a/crates/zeph-core/src/metrics_bridge.rs b/crates/zeph-core/src/metrics_bridge.rs index ecc137dce..2c0eb1028 100644 --- a/crates/zeph-core/src/metrics_bridge.rs +++ b/crates/zeph-core/src/metrics_bridge.rs @@ -1,7 +1,7 @@ // SPDX-FileCopyrightText: 2026 Andrei G // SPDX-License-Identifier: MIT OR Apache-2.0 -//! Tracing layer that derives [`TurnTimings`] from span durations. +//! Tracing layer that derives [`crate::metrics::TurnTimings`] from span durations. //! //! [`MetricsBridge`] implements [`tracing_subscriber::Layer`] and observes //! the close event of a fixed set of known spans. When a watched span closes, @@ -29,11 +29,19 @@ use tracing_subscriber::registry::LookupSpan; use crate::metrics::MetricsCollector; /// Span names watched by the bridge, mapped to the [`TimingField`] they update. +/// +/// `persist_message`'s real span (`core.persist.persist_message`) fires on every +/// `persist_message()` call — 7+ times per turn (user message, assistant response(s), tool +/// results) — not once per turn like the other three. `TurnTimings::persist_message_ms` is +/// meant to time only the first user-message persist of the turn, which only the pre-existing +/// manual `Instant::now()` timing in `agent/mod.rs` captures; watching the real span here would +/// make the bridge overwrite that value with whichever persist call happens to close last, which +/// is wrong. So `persist_message` is intentionally left out of `WATCHED_SPANS` — that field stays +/// manually-timed only (#6111). const WATCHED_SPANS: &[(&str, TimingField)] = &[ - ("agent.prepare_context", TimingField::PrepareContext), + ("core.context.prepare_context", TimingField::PrepareContext), ("llm.chat", TimingField::LlmChat), - ("agent.tool_loop", TimingField::ToolExec), - ("agent.persist_message", TimingField::PersistMessage), + ("core.tool.native_loop", TimingField::ToolExec), ]; /// Identifies which [`crate::metrics::TurnTimings`] field a watched span maps to. @@ -46,7 +54,6 @@ pub(crate) enum TimingField { PrepareContext, LlmChat, ToolExec, - PersistMessage, } impl TimingField { @@ -59,7 +66,6 @@ impl TimingField { Self::PrepareContext => 1 << 0, Self::LlmChat => 1 << 1, Self::ToolExec => 1 << 2, - Self::PersistMessage => 1 << 3, } } } @@ -84,8 +90,8 @@ struct SpanTiming(u64); /// Custom tracing layer that derives [`crate::metrics::TurnTimings`] from span durations. /// -/// Watches a fixed set of span names (`agent.prepare_context`, `llm.chat`, -/// `agent.tool_loop`, `agent.persist_message`) and records their close-time +/// Watches a fixed set of span names (`core.context.prepare_context`, `llm.chat`, +/// `core.tool.native_loop` — see `WATCHED_SPANS`) and records their close-time /// durations into a shared [`MetricsCollector`]. /// /// Timing is captured on `on_enter` (not `on_new_span`) so that async spans @@ -187,9 +193,6 @@ where TimingField::ToolExec => { m.last_turn_timings.tool_exec_ms = duration_ms; } - TimingField::PersistMessage => { - m.last_turn_timings.persist_message_ms = duration_ms; - } } m.bridge_timings_written |= field.bridge_bit(); }); @@ -263,14 +266,14 @@ mod tests { assert_eq!(snapshot.last_turn_timings.persist_message_ms, 0); } - /// All four watched span names must be present in `WATCHED_SPANS`. + /// The three watched span names must be present in `WATCHED_SPANS` (#6111: + /// `agent.persist_message` was deliberately dropped, see the `WATCHED_SPANS` doc comment). #[test] fn all_watched_span_names_registered() { let expected = [ - "agent.prepare_context", + "core.context.prepare_context", "llm.chat", - "agent.tool_loop", - "agent.persist_message", + "core.tool.native_loop", ]; for span_name in expected { @@ -285,4 +288,29 @@ mod tests { "unexpected extra spans in WATCHED_SPANS" ); } + + /// Regression guard for #6111: `WATCHED_SPANS` entries whose span is defined in this crate + /// must match a real `#[tracing::instrument(name = "...")]` name, not a name that looks + /// plausible but was never wired up. `llm.chat` lives in `zeph-llm` and is exercised by + /// that crate's own span-name literals instead. + #[test] + fn watched_spans_match_real_instrument_names_in_this_crate() { + let assembly_src = include_str!(concat!( + env!("CARGO_MANIFEST_DIR"), + "/src/agent/context/assembly.rs" + )); + let tier_loop_src = include_str!(concat!( + env!("CARGO_MANIFEST_DIR"), + "/src/agent/tool_execution/tier_loop.rs" + )); + + assert!( + assembly_src.contains(r#"name = "core.context.prepare_context""#), + "core.context.prepare_context span not found in assembly.rs — WATCHED_SPANS has drifted" + ); + assert!( + tier_loop_src.contains(r#"name = "core.tool.native_loop""#), + "core.tool.native_loop span not found in tier_loop.rs — WATCHED_SPANS has drifted" + ); + } }