fix(core): parallelize apply_tier_results hooks and align WATCHED_SPANS names

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

crates/zeph-core/src/agent/tool_execution/tests/apply_tier_results_tests.rs

@@ -0,0 +1,135 @@
+// SPDX-FileCopyrightText: 2026 Andrei G <bug-ops>
+// 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<Mutex<Vec<String>>>,
+}
+
+impl RuntimeLayer for DelayRecordingLayer {
+    fn after_tool<'a>(
+        &'a self,
+        _ctx: &'a LayerContext<'_>,
+        call: &'a ToolCall,
+        _result: &'a Result<Option<ToolOutput>, ToolError>,
+    ) -> Pin<Box<dyn Future<Output = ()> + 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<Mutex<Vec<String>>> = 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<ToolUseRequest> = (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<ToolUseRequest> = (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"
+    );
+}

crates/zeph-core/src/agent/tool_execution/tests/mod.rs

@@ -1,6 +1,7 @@
 // SPDX-FileCopyrightText: 2026 Andrei G <bug-ops>
 // 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;

crates/zeph-core/src/agent/tool_execution/tier_loop.rs

@@ -1410,6 +1410,7 @@ impl<C: Channel> Agent<C> {
                 mcp_tool_ids,
                 args_hashes,
                 tool_started_ats,
+                max_parallel,
                 &mut failed_ids,
                 &mut tool_results,
             )
@@ -2215,10 +2216,17 @@ impl<C: Channel> Agent<C> {
         mcp_tool_ids: &std::collections::HashSet<String>,
         args_hashes: &[u64],
         tool_started_ats: &[std::time::Instant],
+        max_parallel: usize,
         failed_ids: &mut std::collections::HashSet<String>,
         tool_results: &mut [Result<Option<zeph_tools::ToolOutput>, 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<Option<zeph_tools::ToolOutput>, 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<C: Channel> Agent<C> {
                     .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<zeph_config::HookDef> = 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<C: Channel> Agent<C> {
                         _ => 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<C: Channel> Agent<C> {
                     )
                     .instrument(tracing::info_span!(
                         "core.hooks.post_tool_use",
-                        tool = %tool_calls[idx].name
+                        tool = %tc.name
                     ))
                     .await
                     {
@@ -2346,7 +2374,7 @@ impl<C: Channel> Agent<C> {
                                 // 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<C: Channel> Agent<C> {
                         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;
         }
     }

crates/zeph-core/src/agent/utils.rs

@@ -197,9 +197,9 @@ impl<C: Channel> Agent<C> {
             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;
         });