fix(core): parallelize PreToolUse hooks, backfill extraction, dedupe MAGE mapping

CHANGELOG.md

@@ -210,6 +210,23 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
 ### Fixed
 
+- **core**: `build_tier_call_futures` fired each tier's `PreToolUse` hooks sequentially,
+  adding `N × hook_latency` of purely serial blocking on the agent turn loop before the
+  tier's already-parallelized tool execution even began — the same defect class already
+  fixed for the `PostToolUse` side (#6128) but never mirrored to `PreToolUse`. Hooks now
+  fire concurrently, bounded by the tier semaphore, with the per-call invariant preserved
+  (each call's own hook still fires before that call's own gate check) (#6259).
+- **core**: `AgentAccess::graph_backfill` extracted entities/edges from each unprocessed
+  message strictly sequentially — one LLM call plus SQLite/Qdrant write at a time — despite
+  the store's `UNIQUE(canonical_name, entity_type)` upsert already making concurrent
+  extraction across messages safe. Now uses `futures::stream::iter(...).buffer_unordered(4)`,
+  matching the existing `semantic_scan_plugin_add` pattern, cutting backfill wall time
+  roughly 4x with no correctness change (#6261).
+- **core**: `Agent::begin_turn` re-derived the MAGE `(AuditSignalType, Severity)` pair from
+  the raw trajectory-signal `u8` code via an independent hand-rolled match, duplicating the
+  code-to-meaning table already authoritative in `RiskSignal::from_code` — the two tables
+  were not compiler-coupled and could silently drift. Now matches on the already-computed
+  `RiskSignal` enum value instead; zero behavior change (#6272).
 - **Security (`ShadowSentinel`)**: `check_tool_call` awaited its two pre-tool-dispatch DB reads
   (`get_trajectory`, `get_tool_history`) with no timeout, so a stalled DB connection (e.g. a
   slow/unresponsive Postgres backend) could block dispatch of every `Shell`/`FileWrite`/

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

@@ -942,24 +942,34 @@ impl<C: Channel> Agent<C> {
         // Spec 050 §2: drain pending risk signals from executor layers before advancing.
         // Also advance MAGE accumulator (spec 004-16 FR-009) and ingest mapped signals.
         {
+            use crate::agent::trajectory::{RiskSignal, VigilRiskLevel};
             use zeph_memory::shadow::{AuditSignalType as MageSignal, Severity as MageSev};
             let pending: Vec<u8> = {
                 let mut q = self.services.security.trajectory_signal_queue.lock();
                 std::mem::take(&mut *q)
             };
             self.services.security.mage_accumulator.advance_turn();
             for code in pending {
-                self.services
-                    .security
-                    .trajectory
-                    .record(crate::agent::trajectory::RiskSignal::from_code(code));
-                // Map signal codes to MAGE AuditSignalType + Severity (spec 004-16 FR-002, FR-007).
-                // Code 1=PolicyDeny, 6=VigilMedium, 7=VigilHigh, 2=ExfiltrationRedaction.
-                let mage_signal: Option<(MageSignal, MageSev)> = match code {
-                    1 => Some((MageSignal::PolicyViolation, MageSev::Medium)),
-                    2 => Some((MageSignal::ToolChainAnomaly, MageSev::Medium)),
-                    6 => Some((MageSignal::PromptInjectionPattern, MageSev::Medium)),
-                    7 => Some((MageSignal::PromptInjectionPattern, MageSev::High)),
+                let signal = RiskSignal::from_code(code);
+                self.services.security.trajectory.record(signal);
+                // Map RiskSignal to MAGE AuditSignalType + Severity (spec 004-16 FR-002, FR-007).
+                // Matching on the already-decoded `RiskSignal` (rather than the raw `code`)
+                // keeps this in sync with `RiskSignal::from_code`, the single source of truth
+                // for the code-to-meaning table. Only the four spec-004-16 signal classes have a
+                // MAGE equivalent; the remaining RiskSignal variants (OutOfScope, PiiRedaction,
+                // ToolFailure, HighCallRate, UnusualReadVolume, ToolPairTransition, and
+                // VigilFlagged(Low)) are trajectory-only and intentionally not surfaced to MAGE.
+                let mage_signal: Option<(MageSignal, MageSev)> = match signal {
+                    RiskSignal::PolicyDeny => Some((MageSignal::PolicyViolation, MageSev::Medium)),
+                    RiskSignal::ExfiltrationRedaction => {
+                        Some((MageSignal::ToolChainAnomaly, MageSev::Medium))
+                    }
+                    RiskSignal::VigilFlagged(VigilRiskLevel::Medium) => {
+                        Some((MageSignal::PromptInjectionPattern, MageSev::Medium))
+                    }
+                    RiskSignal::VigilFlagged(VigilRiskLevel::High) => {
+                        Some((MageSignal::PromptInjectionPattern, MageSev::High))
+                    }
                     _ => None,
                 };
                 if let Some((sig, sev)) = mage_signal {

crates/zeph-core/src/agent/tests/mage_signal_mapping_tests.rs

@@ -0,0 +1,136 @@
+// SPDX-FileCopyrightText: 2026 Andrei G <bug-ops>
+// SPDX-License-Identifier: MIT OR Apache-2.0
+
+//! Tests for #6272: `Agent::begin_turn` maps drained `RiskSignal`s to MAGE
+//! `(AuditSignalType, Severity)` pairs by matching on the already-decoded `RiskSignal` enum
+//! rather than re-deriving the mapping from the raw `u8` signal code. These tests pin the
+//! resulting mapping table (spec 004-16 FR-002/FR-007) so a future refactor of either
+//! `RiskSignal::from_code` or the MAGE match arm cannot silently desync the two.
+
+use zeph_config::TrajectoryRiskAccumulatorConfig;
+use zeph_memory::shadow::{AuditSignalType, Severity};
+
+use crate::agent::agent_tests::{
+    MockChannel, MockToolExecutor, create_test_registry, mock_provider,
+};
+use crate::agent::turn::TurnInput;
+use crate::agent::{Agent, trajectory::RiskSignal};
+
+fn make_agent_with_mage() -> Agent<MockChannel> {
+    let agent = Agent::new(
+        mock_provider(vec![]),
+        MockChannel::new(vec![]),
+        create_test_registry(),
+        None,
+        5,
+        MockToolExecutor::no_tools(),
+    );
+    agent.with_mage_accumulator_config(TrajectoryRiskAccumulatorConfig {
+        enabled: true,
+        ..Default::default()
+    })
+}
+
+/// Push a raw signal code into the trajectory queue the same way `RiskSignalSink` callbacks
+/// do, then drive one turn so `begin_turn` drains and maps it.
+fn drain_one_code(agent: &mut Agent<MockChannel>, code: u8) {
+    agent
+        .services
+        .security
+        .trajectory_signal_queue
+        .lock()
+        .push(code);
+    let _turn = agent.begin_turn(TurnInput::new("hi".to_owned(), vec![]));
+}
+
+/// Codes 1, 2, 6, 7 (`PolicyDeny`, `ExfiltrationRedaction`, `VigilFlagged(Medium)`,
+/// `VigilFlagged(High)`) are the only `RiskSignal` variants with a MAGE equivalent
+/// (spec 004-16 FR-002). Each must ingest into `mage_accumulator` with the exact
+/// `AuditSignalType`/`Severity` pair documented at the match site in `begin_turn`.
+#[test]
+fn begin_turn_maps_known_risk_codes_to_mage_signals() {
+    let cases: [(u8, AuditSignalType, Severity); 4] = [
+        (1, AuditSignalType::PolicyViolation, Severity::Medium),
+        (2, AuditSignalType::ToolChainAnomaly, Severity::Medium),
+        (6, AuditSignalType::PromptInjectionPattern, Severity::Medium),
+        (7, AuditSignalType::PromptInjectionPattern, Severity::High),
+    ];
+
+    for (code, expected_type, expected_severity) in cases {
+        let mut agent = make_agent_with_mage();
+        drain_one_code(&mut agent, code);
+
+        assert!(
+            agent.services.security.mage_accumulator.current_risk() > 0.0,
+            "code {code} must ingest a non-zero-weight MAGE signal"
+        );
+        let top = agent.services.security.mage_accumulator.top_signals(1);
+        assert_eq!(
+            top.len(),
+            1,
+            "code {code} must record exactly one MAGE signal event"
+        );
+        assert_eq!(
+            top[0].signal_type, expected_type,
+            "code {code} mapped to the wrong AuditSignalType"
+        );
+        assert_eq!(
+            top[0].severity, expected_severity,
+            "code {code} mapped to the wrong Severity"
+        );
+    }
+}
+
+/// The remaining `RiskSignal` variants — `OutOfScope` (3), `PiiRedaction` (4),
+/// `ToolFailure` (5), and the `VigilFlagged(Low)` fallback (any unmapped code, e.g. 99) —
+/// are trajectory-only per the doc comment above the match in `begin_turn` and must NOT
+/// surface to MAGE.
+#[test]
+fn begin_turn_no_mage_signal_for_trajectory_only_codes() {
+    for code in [3u8, 4, 5, 99] {
+        let mut agent = make_agent_with_mage();
+        drain_one_code(&mut agent, code);
+
+        // trajectory_risk only ever accumulates non-negative contributions, so `<= 0.0` is
+        // equivalent to `== 0.0` here without tripping clippy::float_cmp on exact equality.
+        assert!(
+            agent.services.security.mage_accumulator.current_risk() <= 0.0,
+            "code {code} must not ingest any MAGE signal (trajectory-only)"
+        );
+        assert!(
+            agent
+                .services
+                .security
+                .mage_accumulator
+                .top_signals(1)
+                .is_empty(),
+            "code {code} must leave MAGE signal history empty"
+        );
+    }
+}
+
+/// Sanity guard: `RiskSignal::from_code` itself must still decode these codes to the
+/// variants this test file assumes — if this fails, the MAGE-mapping tests above are
+/// exercising the wrong `RiskSignal`, not the mapping logic.
+#[test]
+fn risk_signal_from_code_matches_assumed_variants() {
+    use crate::agent::trajectory::VigilRiskLevel;
+
+    assert_eq!(RiskSignal::from_code(1), RiskSignal::PolicyDeny);
+    assert_eq!(RiskSignal::from_code(2), RiskSignal::ExfiltrationRedaction);
+    assert_eq!(
+        RiskSignal::from_code(6),
+        RiskSignal::VigilFlagged(VigilRiskLevel::Medium)
+    );
+    assert_eq!(
+        RiskSignal::from_code(7),
+        RiskSignal::VigilFlagged(VigilRiskLevel::High)
+    );
+    assert_eq!(RiskSignal::from_code(3), RiskSignal::OutOfScope);
+    assert_eq!(RiskSignal::from_code(4), RiskSignal::PiiRedaction);
+    assert_eq!(RiskSignal::from_code(5), RiskSignal::ToolFailure);
+    assert_eq!(
+        RiskSignal::from_code(99),
+        RiskSignal::VigilFlagged(VigilRiskLevel::Low)
+    );
+}

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

@@ -20,6 +20,8 @@ mod flush_orphaned_tests;
 #[cfg(test)]
 mod inline_tool_loop_tests;
 #[cfg(test)]
+mod mage_signal_mapping_tests;
+#[cfg(test)]
 mod pre_execution_audit_tests;
 #[cfg(test)]
 mod provider_override_masking_tests;

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

@@ -8,6 +8,7 @@ mod hook_block_cap_tests;
 mod mage_escalation_tests;
 mod native_tests;
 mod parallel_and_handle_tests;
+mod pre_tool_use_concurrency_tests;
 mod pure_helpers_tests;
 mod retry_and_skill_env_tests;
 mod sanitize_and_native_tests;

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

@@ -0,0 +1,173 @@
+// SPDX-FileCopyrightText: 2026 Andrei G <bug-ops>
+// SPDX-License-Identifier: MIT OR Apache-2.0
+
+//! Tests for #6259: `build_tier_call_futures` must fire `PreToolUse` hooks for every tool
+//! call in a tier concurrently (Phase 1), not serially, before running the sequential
+//! per-index gate-check loop (Phase 2). Mirrors `apply_tier_results_tests.rs`'s coverage of
+//! the already-fixed `PostToolUse`/`RuntimeLayer::after_tool` twin (#6128).
+
+use std::time::{Duration, Instant};
+
+use zeph_config::{HookAction, HookDef, HookMatcher};
+use zeph_llm::provider::{Message, MessagePart, Role, ToolUseRequest};
+
+use crate::agent::agent_tests::{
+    MockChannel, MockToolExecutor, create_test_registry, mock_provider,
+};
+
+fn make_tool_use_request(id: &str, name: &str) -> ToolUseRequest {
+    ToolUseRequest {
+        id: id.into(),
+        name: name.into(),
+        input: serde_json::json!({}),
+    }
+}
+
+fn sleep_hook(secs: f64) -> HookDef {
+    HookDef {
+        action: HookAction::Command {
+            command: format!("sleep {secs}"),
+        },
+        timeout_secs: 5,
+        fail_closed: false,
+        r#if: None,
+    }
+}
+
+/// N tool calls land in a single tier, each matching a `PreToolUse` hook that sleeps.
+/// Serial hook dispatch (the pre-#6259 behavior) would take N * delay; concurrent dispatch
+/// (Phase 1, bounded by the tier semaphore) should stay close to a single delay regardless
+/// of N.
+#[tokio::test]
+async fn pre_tool_use_hooks_fire_concurrently_across_tier_indices() {
+    let n = 4;
+    let delay_secs = 0.06;
+    let delay = Duration::from_millis(60);
+
+    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.services.session.hooks_config.pre_tool_use = vec![HookMatcher {
+        matcher: "noop".to_owned(),
+        hooks: vec![sleep_hook(delay_secs)],
+    }];
+    agent
+        .msg
+        .messages
+        .push(Message::from_legacy(Role::System, "system"));
+
+    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();
+
+    assert!(
+        elapsed < delay * u32::try_from(n).unwrap(),
+        "PreToolUse hooks appear to have fired serially: took {elapsed:?} for {n} x {delay:?}"
+    );
+
+    // Every call must still have proceeded to execution (hook is fail_open and succeeds).
+    let tool_result_count = agent
+        .msg
+        .messages
+        .iter()
+        .flat_map(|m| m.parts.iter())
+        .filter(|p| matches!(p, MessagePart::ToolResult { .. }))
+        .count();
+    assert_eq!(
+        tool_result_count, n,
+        "every tool call must get a persisted ToolResult after its PreToolUse hook fires"
+    );
+}
+
+/// Order invariant: a `fail_closed` `PreToolUse` hook block on one tier index must not affect
+/// sibling indices in the same tier. Regression guard for the Phase 1 / Phase 2 split — Phase
+/// 1 collects all blocked indices into a `HashMap<usize, String>` up front, and Phase 2 must
+/// only consult the entry for its own idx.
+#[tokio::test]
+async fn pre_tool_use_hook_block_on_one_index_does_not_affect_siblings() {
+    let provider = mock_provider(vec![]);
+    let channel = MockChannel::new(vec![]);
+    let registry = create_test_registry();
+    // Only one call ("read") will actually reach the executor; "shell" is blocked before
+    // dispatch by its fail_closed PreToolUse hook.
+    let executor = MockToolExecutor::new(vec![Ok(None)]);
+    let mut agent = crate::agent::Agent::new(provider, channel, registry, None, 5, executor);
+    agent.runtime.config.timeouts.max_parallel_tools = 2;
+    agent.services.session.hooks_config.pre_tool_use = vec![HookMatcher {
+        matcher: "shell".to_owned(),
+        hooks: vec![HookDef {
+            action: HookAction::Command {
+                command: "exit 1".to_owned(),
+            },
+            timeout_secs: 5,
+            fail_closed: true,
+            r#if: None,
+        }],
+    }];
+    agent
+        .msg
+        .messages
+        .push(Message::from_legacy(Role::System, "system"));
+
+    let tool_calls = vec![
+        make_tool_use_request("id-shell", "shell"),
+        make_tool_use_request("id-read", "read"),
+    ];
+
+    agent
+        .handle_native_tool_calls(None, &tool_calls)
+        .await
+        .unwrap();
+
+    assert_eq!(
+        agent.tool_orchestrator.hook_block_count, 1,
+        "exactly one call (shell) must be blocked by its own fail_closed hook"
+    );
+
+    let tool_results: Vec<(&str, &str)> = agent
+        .msg
+        .messages
+        .iter()
+        .flat_map(|m| m.parts.iter())
+        .filter_map(|p| {
+            if let MessagePart::ToolResult {
+                tool_use_id,
+                content,
+                ..
+            } = p
+            {
+                Some((tool_use_id.as_str(), content.as_str()))
+            } else {
+                None
+            }
+        })
+        .collect();
+
+    let shell_result = tool_results
+        .iter()
+        .find(|(id, _)| *id == "id-shell")
+        .expect("shell ToolResult must be present");
+    assert!(
+        shell_result.1.contains("[blocked]"),
+        "shell call must be blocked by its own hook: {shell_result:?}"
+    );
+
+    let read_result = tool_results
+        .iter()
+        .find(|(id, _)| *id == "id-read")
+        .expect("read ToolResult must be present");
+    assert!(
+        !read_result.1.contains("[blocked]"),
+        "read call has no matching hook and must NOT be blocked by shell's hook: {read_result:?}"
+    );
+}