fix(skills): stop a proactively-matched Quarantined skill from locking a turn's tools

CHANGELOG.md

@@ -175,6 +175,39 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
   call site shared by both the startup catalog emit and every hot-reload — so the warning no
   longer depends on `SemanticMemory` being configured and fires at startup, not only after the
   first post-startup file change.
+- `zeph-core`, `zeph-tools`, `zeph-common`, `zeph-subagent`: a skill that merely scored above
+  the matcher's similarity threshold this turn — never explicitly invoked by the model — could
+  drop the entire turn's `effective_trust` to `Quarantined` via the weakest-link fold in
+  `apply_skill_trust_and_gating`, denying every `bash`/`write`/MCP tool call for the rest of the
+  turn and every foreground-spawned subagent (issue #6701, security P1). Root causes and fixes:
+  - **RC-1**: `active_skill_names` was assigned directly from the matcher's output with no trust
+    filter. Added a trust-aware activation filter (`Agent::filter_active_skills_by_trust`) that
+    drops `Quarantined`/`Blocked` matches from the active set before the fold runs, while keeping
+    them visible in the `<other_skills>` catalog (`format_skills_catalog` now takes a trust-level
+    map and annotates entries with `trust="quarantined"`/`trust="blocked"`) so the model/operator
+    can still name them and promote with `zeph skill trust <name> trusted`. A `tracing::warn!`
+    names the dropped skill and the promotion command.
+  - **RC-2**: retrieval-fallback mode (matcher unavailable or unconfigured) previously folded
+    trust over the full, unscored skill set even though only description-only catalog text is
+    injected in that mode. `apply_skill_trust_and_gating`'s effective-trust computation
+    (`compute_effective_trust`) now forces `Trusted` whenever `skill_fallback_mode` is set,
+    regardless of how many Quarantined/Blocked skills exist in the registry.
+  - **RC-3**: `SkillTrustGate::resolve_body` refused only `Blocked` skills — an explicit
+    `invoke_skill`/`load_skill` of a `Quarantined` skill returned its sanitized body but never
+    degraded the turn's trust for subsequent tool calls, contradicting the documented invariant.
+    It now folds the new `zeph_common::TurnTrustFloor` (a monotonic-downgrade-only shared cell,
+    `set`/`fold`/`get`) to `Quarantined` whenever it returns a Quarantined body.
+  - **RC-5**: subagent spawn/resume applied an inherited trust cap via a plain
+    `set_effective_trust(cap)`, which could restore trust above a floor already folded lower
+    earlier in the same task. `SpawnContext` gained a `turn_trust_floor` handle
+    (`Option<TurnTrustFloor>`), and spawn/resume now calls `TurnTrustFloor::fold(cap)` on it when
+    present, falling back to the old `set_effective_trust` behavior otherwise.
+  - The weakest-link fold itself is unchanged as defense-in-depth for skills actually
+    injected/invoked this turn, and is now a single shared helper (`fold_weakest_trust`) used by
+    both `apply_skill_trust_and_gating` and the subagent trust-cap computation
+    (`parent_effective_trust_level`), which previously duplicated the same logic.
+  - RC-4 (AutoSkill draft-name collisions with native tool IDs) is tracked separately in #6702
+    and intentionally out of scope here.
 
 - `zeph-channels`: `MAX_RETRY_SECS` (the upper bound `send_with_retry` clamps a `Retry-After`
   delay to) had no compile-time invariant guard (issue #6517). #6516 (closing #6496) filtered

crates/zeph-common/src/lib.rs

@@ -44,6 +44,7 @@ pub mod text;
 pub mod timestamp;
 pub mod tool_classification;
 pub mod trust_level;
+pub mod turn_trust_floor;
 pub mod types;
 
 /// Prefix embedded in tool output bodies when the full output was stored externally.
@@ -65,6 +66,7 @@ pub use task_supervisor::{
 };
 pub use text::format_tokens;
 pub use trust_level::SkillTrustLevel;
+pub use turn_trust_floor::TurnTrustFloor;
 pub use types::{
     ProviderName, SessionId, SessionIdError, SkillName, StopHint, ToolDefinition, ToolName,
 };

crates/zeph-common/src/trust_level.rs

@@ -103,6 +103,30 @@ impl SkillTrustLevel {
         }
     }
 
+    /// Inverse of [`severity`](Self::severity): reconstructs a level from its ordinal.
+    ///
+    /// Any value `>= 3` maps to [`Blocked`](Self::Blocked) — the most restrictive level —
+    /// so a corrupted or out-of-range stored ordinal fails closed rather than open.
+    ///
+    /// # Examples
+    ///
+    /// ```rust
+    /// use zeph_common::SkillTrustLevel;
+    ///
+    /// assert_eq!(SkillTrustLevel::from_severity(0), SkillTrustLevel::Trusted);
+    /// assert_eq!(SkillTrustLevel::from_severity(3), SkillTrustLevel::Blocked);
+    /// assert_eq!(SkillTrustLevel::from_severity(255), SkillTrustLevel::Blocked);
+    /// ```
+    #[must_use]
+    pub const fn from_severity(v: u8) -> Self {
+        match v {
+            0 => Self::Trusted,
+            1 => Self::Verified,
+            2 => Self::Quarantined,
+            _ => Self::Blocked,
+        }
+    }
+
     /// Returns the string representation used for database storage.
     #[must_use]
     pub const fn as_str(self) -> &'static str {
@@ -217,4 +241,25 @@ mod tests {
             SkillTrustLevel::Verified
         );
     }
+
+    #[test]
+    fn from_severity_round_trips_through_severity() {
+        for level in [
+            SkillTrustLevel::Trusted,
+            SkillTrustLevel::Verified,
+            SkillTrustLevel::Quarantined,
+            SkillTrustLevel::Blocked,
+        ] {
+            assert_eq!(SkillTrustLevel::from_severity(level.severity()), level);
+        }
+    }
+
+    #[test]
+    fn from_severity_out_of_range_fails_closed_to_blocked() {
+        assert_eq!(SkillTrustLevel::from_severity(4), SkillTrustLevel::Blocked);
+        assert_eq!(
+            SkillTrustLevel::from_severity(255),
+            SkillTrustLevel::Blocked
+        );
+    }
 }

crates/zeph-common/src/turn_trust_floor.rs

@@ -0,0 +1,150 @@
+// SPDX-FileCopyrightText: 2026 Andrei G <bug-ops>
+// SPDX-License-Identifier: MIT OR Apache-2.0
+
+//! Shared, monotonically-downgradable trust floor for a single agent turn (#6701).
+
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU8, Ordering};
+
+use crate::SkillTrustLevel;
+
+/// Per-turn trust floor shared between a `ToolExecutor`'s trust gate and any skill-body
+/// resolution path that can downgrade it mid-turn (e.g. an explicit `invoke_skill` of a
+/// Quarantined skill).
+///
+/// Wraps an `Arc<AtomicU8>` — cloning shares the same underlying cell, so every holder
+/// observes the same value. Two operations mutate it:
+///
+/// - [`set`](Self::set): turn-start assignment. Replaces the floor unconditionally — trust
+///   may go up or down relative to the previous turn. Call once per turn, before any tool
+///   dispatch.
+/// - [`fold`](Self::fold): monotonic downgrade. The floor becomes `min(current, level)` —
+///   it can never raise trust. Models "this turn's trust degraded because
+///   prompt-injected/quarantined content was actually read" — the only way back up is a
+///   fresh [`set`](Self::set) at the next turn boundary.
+///
+/// # Examples
+///
+/// ```rust
+/// use zeph_common::{SkillTrustLevel, TurnTrustFloor};
+///
+/// let floor = TurnTrustFloor::new(SkillTrustLevel::Trusted);
+/// floor.fold(SkillTrustLevel::Quarantined);
+/// assert_eq!(floor.get(), SkillTrustLevel::Quarantined);
+///
+/// // fold never raises trust, even toward a higher-trust argument.
+/// floor.fold(SkillTrustLevel::Trusted);
+/// assert_eq!(floor.get(), SkillTrustLevel::Quarantined);
+///
+/// // set is a full turn-start reset — it can raise trust again.
+/// floor.set(SkillTrustLevel::Trusted);
+/// assert_eq!(floor.get(), SkillTrustLevel::Trusted);
+/// ```
+#[derive(Clone, Debug)]
+pub struct TurnTrustFloor(Arc<AtomicU8>);
+
+impl TurnTrustFloor {
+    /// Creates a new floor initialized to `initial`.
+    #[must_use]
+    pub fn new(initial: SkillTrustLevel) -> Self {
+        Self(Arc::new(AtomicU8::new(initial.severity())))
+    }
+
+    /// Turn-start assignment: replaces the floor unconditionally.
+    ///
+    /// Call once per turn, before any tool dispatch — never mid-turn, or a genuine
+    /// mid-turn downgrade (see [`fold`](Self::fold)) could be silently undone.
+    pub fn set(&self, level: SkillTrustLevel) {
+        self.0.store(level.severity(), Ordering::Relaxed);
+    }
+
+    /// Monotonic downgrade: the floor becomes `min(current, level)`.
+    ///
+    /// Can never raise trust. The `Result` from the underlying CAS is intentionally
+    /// discarded — the closure always returns `Some`, so the update always succeeds.
+    pub fn fold(&self, level: SkillTrustLevel) {
+        let _ = self
+            .0
+            .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |cur| {
+                let current = SkillTrustLevel::from_severity(cur);
+                Some(current.min_trust(level).severity())
+            });
+    }
+
+    /// Returns the current floor value.
+    #[must_use]
+    pub fn get(&self) -> SkillTrustLevel {
+        SkillTrustLevel::from_severity(self.0.load(Ordering::Relaxed))
+    }
+}
+
+impl Default for TurnTrustFloor {
+    /// Defaults to [`SkillTrustLevel::Trusted`] — the same starting point
+    /// `TrustGateExecutor::new` used before this type existed.
+    fn default() -> Self {
+        Self::new(SkillTrustLevel::Trusted)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn set_replaces_unconditionally_in_either_direction() {
+        let floor = TurnTrustFloor::new(SkillTrustLevel::Quarantined);
+        floor.set(SkillTrustLevel::Trusted);
+        assert_eq!(floor.get(), SkillTrustLevel::Trusted);
+        floor.set(SkillTrustLevel::Blocked);
+        assert_eq!(floor.get(), SkillTrustLevel::Blocked);
+    }
+
+    #[test]
+    fn fold_lowers_trust() {
+        let floor = TurnTrustFloor::new(SkillTrustLevel::Trusted);
+        floor.fold(SkillTrustLevel::Quarantined);
+        assert_eq!(floor.get(), SkillTrustLevel::Quarantined);
+    }
+
+    #[test]
+    fn fold_never_raises_trust() {
+        let floor = TurnTrustFloor::new(SkillTrustLevel::Quarantined);
+        floor.fold(SkillTrustLevel::Trusted);
+        assert_eq!(
+            floor.get(),
+            SkillTrustLevel::Quarantined,
+            "fold must never raise trust above the current floor"
+        );
+    }
+
+    #[test]
+    fn fold_to_blocked_is_sticky_until_a_fresh_set() {
+        let floor = TurnTrustFloor::new(SkillTrustLevel::Verified);
+        floor.fold(SkillTrustLevel::Blocked);
+        floor.fold(SkillTrustLevel::Trusted);
+        assert_eq!(floor.get(), SkillTrustLevel::Blocked);
+        floor.set(SkillTrustLevel::Trusted);
+        assert_eq!(
+            floor.get(),
+            SkillTrustLevel::Trusted,
+            "only a fresh turn-start set recovers trust, never fold"
+        );
+    }
+
+    #[test]
+    fn clone_shares_the_same_underlying_cell() {
+        let floor = TurnTrustFloor::new(SkillTrustLevel::Trusted);
+        let clone = floor.clone();
+        clone.fold(SkillTrustLevel::Quarantined);
+        assert_eq!(
+            floor.get(),
+            SkillTrustLevel::Quarantined,
+            "clones must observe writes through any other clone"
+        );
+    }
+
+    #[test]
+    fn default_starts_trusted() {
+        assert_eq!(TurnTrustFloor::default().get(), SkillTrustLevel::Trusted);
+    }
+}

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

@@ -501,6 +501,21 @@ impl<C: Channel> Agent<C> {
         self
     }
 
+    /// Wire the shared per-turn trust floor (#6701) — the same handle
+    /// `zeph_tools::TrustGateExecutor::trust_floor()` returns for the executor built into this
+    /// agent's `tool_executor` chain.
+    ///
+    /// Enables two things: `SkillTrustGate::resolve_body` folding the turn's trust down on a
+    /// Quarantined body read (wired separately via `SkillLoaderExecutor`/`SkillInvokeExecutor`'s
+    /// own `with_turn_trust_floor`, sharing this same `Arc`), and subagent spawn folding an
+    /// inherited trust cap onto this exact cell (`SpawnContext::turn_trust_floor`) instead of
+    /// overwriting it via `set_effective_trust`.
+    #[must_use]
+    pub fn with_turn_trust_floor(mut self, floor: zeph_common::TurnTrustFloor) -> Self {
+        self.services.skill.turn_trust_floor = Some(floor);
+        self
+    }
+
     /// Configure skill matching parameters (disambiguation, two-stage, confusability).
     #[must_use]
     pub fn with_skill_matching_config(