feat(tiers): expand skill and tool management in tier evaluations

- Updated tier evaluation functions to include skills alongside tools, enhancing the logic for determining unused and promoted entities.
- Refactored `_tool_unused_at_t2` and `crystallize_successful_tool_promotions_at_epoch` to accommodate skills, ensuring consistent behavior across both entity types.
- Introduced new functions for handling shadow evaluations of skills, improving integration with the existing tier management system.
- Enhanced tests to cover the new skill functionalities and ensure correct tier transitions for both tools and skills.

This update improves the overall functionality of the tier management system by providing better support for skills, aligning with recent enhancements in skills management.

Author: qdrddr

src/cyt/tiers/adapters/skills.py

@@ -811,6 +811,35 @@ def resolve_tiered_skill_matches(
     partition = manager.partition_skills(resolved_entries, config)
     manager.record_skill_candidates(partition.search_entries, config)
 
+    if query.strip():
+        dormant_ids = [
+            entity_id
+            for entry in resolved_entries
+            if (entity_id := skill_entity_id(entry))
+            and partition.tier_by_skill.get(entity_id) == Tier.DORMANT
+        ]
+        if dormant_ids:
+            skills_by_id: dict[str, dict[str, Any]] = {}
+            for entry in resolved_entries:
+                entity_id = skill_entity_id(entry)
+                if not entity_id:
+                    continue
+                document = entry.document if isinstance(entry.document, dict) else {}
+                skills_by_id[entity_id] = {
+                    "name": document.get("name") or entry.doc_id,
+                    "description": document.get("description") or "",
+                    "doc_id": entry.doc_id,
+                }
+            from cyt.tiers.shadow import schedule_skill_shadow_evaluation
+
+            schedule_skill_shadow_evaluation(
+                config=config,
+                query=query,
+                dormant_entity_ids=dormant_ids,
+                skills_by_id=skills_by_id,
+                manager=manager,
+            )
+
     search_pool = prepare_skill_entries_for_tier_search(
         partition.search_entries,
         partition.tier_by_skill,

src/cyt/tiers/evaluator.py

@@ -81,11 +81,11 @@ def _min_exposure_met(state: EntityTierState, minimum: int) -> bool:
 
 
 def _tool_unused_at_t2(state: EntityTierState) -> bool:
-    """True when a tool is still default T2 but never injected or invoked.
+    """True when a tool or skill is still default T2 but never injected or invoked.
 
     BM25 candidacy alone does not preserve T2 across epochs; only injection/use signals do.
     """
-    if state.kind != "tool":
+    if state.kind not in ("tool", "skill"):
         return False
     stats = state.stats
     return (
@@ -278,10 +278,10 @@ def crystallize_successful_tool_promotions_at_epoch(
     *,
     epoch: EpochState,
 ) -> list[TierTransition]:
-    """Promote tools with successful use this epoch to stable HOT (recent signal wins)."""
+    """Promote tools/skills with successful use this epoch to stable HOT (recent signal wins)."""
     transitions: list[TierTransition] = []
     for state in states.values():
-        if state.kind != "tool" or not _epoch_had_success(state):
+        if state.kind not in ("tool", "skill") or not _epoch_had_success(state):
             continue
         state.temp_promotion_until_ms = None
         state.overlap_tier = None
@@ -317,7 +317,11 @@ def expire_temporary_promotions(
             continue
         if state.effective_tier != state.stable_tier:
             old = state.effective_tier
-            crystallize = state.kind == "tool" and old > state.stable_tier and state.stats.used > 0
+            crystallize = (
+                state.kind in ("tool", "skill")
+                and old > state.stable_tier
+                and state.stats.used > 0
+            )
             if crystallize:
                 old_stable = state.stable_tier
                 state.stable_tier = old

src/cyt/tiers/manager.py

@@ -795,20 +795,36 @@ def record_skill_used(self, entity_id: str, *, config: dict[str, Any]) -> None:
         )
         if not canonical_id:
             return
+        cfg = tier_section_config(config, kind="skill")
         epoch_ran = False
+        sync_flush = False
         with self._state_lock:
             state = self._ensure_state(EntityKind.SKILL, canonical_id)
             if state is None:
                 return
             state.stats.used += 1.0
+            state.stats.epoch_used += 1.0
             self._mark_cycle_used(EntityKind.SKILL, canonical_id)
             if canonical_id not in self._last_injected_skills:
                 state.stats.used_without_injection += 1.0
+            fast_promote_on_tool_use(state, cfg=cfg)
+            state.stats.last_seen_ms = int(time.time() * 1000)
             epoch_ran = self._touch_request(config)
+            self._touch_stats_request(state)
+            sync_flush = (
+                state.temp_promotion_until_ms is not None
+                or state.effective_tier > state.stable_tier
+            )
         if not epoch_ran:
-            self._finish_record(config)
+            self._finish_record(config, sync_flush=sync_flush)
 
-    def apply_shadow_tool_hits(self, hits: list[tuple[str, float]], config: dict[str, Any]) -> None:
+    def apply_shadow_tool_hits(
+        self,
+        hits: list[tuple[str, float]],
+        config: dict[str, Any],
+        *,
+        tools_by_id: dict[str, dict[str, Any]] | None = None,
+    ) -> None:
         from cyt.tiers.adapters.tools import tool_entity_tracked_for_config
         from cyt.tiers.shadow import record_shadow_hits
 
@@ -830,11 +846,34 @@ def apply_shadow_tool_hits(self, hits: list[tuple[str, float]], config: dict[str
                 hits=tracked_hits,
                 cfg=cfg,
                 wake_cycle_id=self._epoch.wake_cycle_id,
+                tools_by_id=tools_by_id,
             )
         if transitions and cfg.mode == TierMode.SHADOW:
             logger.debug("tool shadow transitions: %d", len(transitions))
         self._finish_record(config)
 
+    def apply_shadow_skill_hits(self, hits: list[tuple[str, float]], config: dict[str, Any]) -> None:
+        from cyt.tiers.shadow import record_shadow_hits
+
+        if not tiers_active(config, kind="skill"):
+            return
+        cfg = tier_section_config(config, kind="skill")
+        if not hits:
+            return
+        with self._state_lock:
+            for entity_id, _score in hits:
+                self._mark_cycle_shadow(EntityKind.SKILL, entity_id)
+            transitions = record_shadow_hits(
+                self._states,
+                kind=EntityKind.SKILL,
+                hits=hits,
+                cfg=cfg,
+                wake_cycle_id=self._epoch.wake_cycle_id,
+            )
+        if transitions and cfg.mode == TierMode.SHADOW:
+            logger.debug("skill shadow transitions: %d", len(transitions))
+        self._finish_record(config)
+
     def status(
         self,
         config: dict[str, Any] | None = None,

src/cyt/tiers/shadow.py

@@ -70,7 +70,7 @@ def _run() -> None:
 
             section = tier_section_config(config, kind="tool")
             hits = _lexical_shadow_hits(query, dormant_ids, tools_by_id)
-            manager.apply_shadow_tool_hits(hits, config)
+            manager.apply_shadow_tool_hits(hits, config, tools_by_id=tools_by_id)
             if section.mode == TierMode.SHADOW and hits:
                 logger.debug("tool shadow hits: %d dormant tools", len(hits))
         except Exception:
@@ -107,19 +107,98 @@ def _run() -> None:
         _executor.submit(_run)
 
 
+def _entity_mcp_server(entity_id: str, tools_by_id: dict[str, dict[str, Any]]) -> str | None:
+    tool = tools_by_id.get(entity_id)
+    if tool is not None:
+        from cyt.tool_examples.identity import resolve_mcp_server_and_tool
+
+        server, _bare = resolve_mcp_server_and_tool(tool)
+        return server or None
+    if ":" not in entity_id:
+        return None
+    wire_name = entity_id.split(":", 1)[1]
+    from cyt.tool_examples.identity import resolve_mcp_server_and_tool_from_wire_name
+
+    server, _bare = resolve_mcp_server_and_tool_from_wire_name(wire_name)
+    return server or None
+
+
+def _lexical_skill_shadow_hits(
+    query: str,
+    dormant_ids: list[str],
+    skills_by_id: dict[str, dict[str, Any]],
+) -> list[tuple[str, float]]:
+    q = query.lower()
+    hits: list[tuple[str, float]] = []
+    for entity_id in dormant_ids:
+        skill = skills_by_id.get(entity_id)
+        if skill is None:
+            continue
+        name = str(skill.get("name") or skill.get("doc_id") or "").lower()
+        desc = str(skill.get("description") or "").lower()
+        score = 0.0
+        if name and name in q:
+            score = 0.85
+        elif name:
+            parts = [part for part in name.replace("-", "_").split("_") if len(part) > 2]
+            if any(part in q for part in parts):
+                score = 0.55
+        if desc:
+            words = [word for word in desc.split() if len(word) > 4][:12]
+            if any(word.lower() in q for word in words):
+                score = max(score, 0.35)
+        if score > 0.0:
+            hits.append((entity_id, score))
+    return hits
+
+
+def schedule_skill_shadow_evaluation(
+    *,
+    config: dict[str, Any],
+    query: str,
+    dormant_entity_ids: list[str],
+    skills_by_id: dict[str, dict[str, Any]],
+    manager: TierManager | NoOpTierManager,
+) -> None:
+    if not tiers_active(config, kind="skill") or not query.strip():
+        return
+    if not dormant_entity_ids:
+        return
+
+    def _run() -> None:
+        try:
+            from cyt.tiers.config import tier_section_config
+
+            section = tier_section_config(config, kind="skill")
+            hits = _lexical_skill_shadow_hits(query, dormant_entity_ids, skills_by_id)
+            manager.apply_shadow_skill_hits(hits, config)
+            if section.mode == TierMode.SHADOW and hits:
+                logger.debug("skill shadow hits: %d dormant skills", len(hits))
+        except Exception:
+            logger.exception("skill shadow evaluation failed")
+
+    with _executor_lock:
+        _executor.submit(_run)
+
+
 def record_shadow_hits(
     states: dict[tuple[str, str], EntityTierState],
     *,
     kind: str,
     hits: list[tuple[str, float]],
     cfg: TierSectionConfig,
     wake_cycle_id: int,
+    tools_by_id: dict[str, dict[str, Any]] | None = None,
 ) -> list[Any]:
     from cyt.tiers.models import TierTransition
 
     transitions: list[TierTransition] = []
+    woken: set[str] = set()
     hit_set = {entity_id for entity_id, score in hits if score > 0}
-    for entity_id in hit_set:
+
+    def _wake_entity(entity_id: str, *, query_relevance: float, sibling_activity: float) -> None:
+        if entity_id in woken:
+            return
         key = (kind, entity_id)
         state = states.get(key)
         if state is None:
@@ -130,21 +209,64 @@ def record_shadow_hits(
                 effective_tier=Tier.DORMANT,
             )
             states[key] = state
-        state.stats.shadow_evaluations += 1.0
-        state.stats.shadow_hits += 1.0
-        top_score = max((score for eid, score in hits if eid == entity_id), default=0.0)
+        if entity_id in hit_set:
+            state.stats.shadow_hits += 1.0
         wake = evaluate_fast_wake(
             state,
             cfg=cfg,
             wake_cycle_id=wake_cycle_id,
-            query_relevance=top_score,
+            query_relevance=query_relevance,
+            sibling_activity=sibling_activity,
         )
         if wake is not None:
             transitions.append(wake)
-    for state_key in states:
-        if state_key[0] != kind:
-            continue
-        eid = state_key[1]
-        if eid not in hit_set and states[state_key].effective_tier == Tier.DORMANT:
-            states[state_key].stats.shadow_evaluations += 1.0
+            woken.add(entity_id)
+
+    for entity_id in hit_set:
+        key = (kind, entity_id)
+        state = states.get(key)
+        if state is None:
+            state = EntityTierState(
+                entity_id=entity_id,
+                kind=kind,
+                stable_tier=Tier.DORMANT,
+                effective_tier=Tier.DORMANT,
+            )
+            states[key] = state
+        state.stats.shadow_evaluations += 1.0
+        top_score = max((score for eid, score in hits if eid == entity_id), default=0.0)
+        _wake_entity(entity_id, query_relevance=top_score, sibling_activity=0.0)
+
+    server_max_score: dict[str, float] = {}
+    if tools_by_id is not None:
+        for entity_id in hit_set:
+            server = _entity_mcp_server(entity_id, tools_by_id)
+            if not server:
+                continue
+            top_score = max((score for eid, score in hits if eid == entity_id), default=0.0)
+            server_max_score[server] = max(server_max_score.get(server, 0.0), top_score)
+
+    if tools_by_id is not None and server_max_score:
+        for state_key, state in states.items():
+            if state_key[0] != kind or state.effective_tier != Tier.DORMANT:
+                continue
+            entity_id = state_key[1]
+            server = _entity_mcp_server(entity_id, tools_by_id)
+            if not server or server not in server_max_score:
+                continue
+            if entity_id not in hit_set:
+                state.stats.shadow_evaluations += 1.0
+            _wake_entity(
+                entity_id,
+                query_relevance=server_max_score[server],
+                sibling_activity=0.0 if entity_id in hit_set else 1.0,
+            )
+    else:
+        for state_key in states:
+            if state_key[0] != kind:
+                continue
+            eid = state_key[1]
+            if eid not in hit_set and states[state_key].effective_tier == Tier.DORMANT:
+                states[state_key].stats.shadow_evaluations += 1.0
+
     return transitions

src/cyt/tiers/wake.py

@@ -59,7 +59,11 @@ def evaluate_fast_wake(
         query_relevance=query_relevance,
         sibling_activity=sibling_activity,
     )
-    if pressure < cfg.wake_threshold and state.stats.used_without_injection <= 0:
+    if (
+        pressure < cfg.wake_threshold
+        and state.stats.used_without_injection <= 0
+        and query_relevance < cfg.wake_threshold
+    ):
         return None
     old = state.effective_tier
     state.effective_tier = Tier.COLD
@@ -116,8 +120,8 @@ def fast_promote_on_tool_use(
     *,
     cfg: TierSectionConfig | None = None,
 ) -> TierTransition | None:
-    """Tools-only fast signal when the agent invoked the tool (any args)."""
-    if state.kind != "tool":
+    """Fast signal when the agent invoked a tool or skill."""
+    if state.kind not in ("tool", "skill"):
         return None
     now_ms = int(time.time() * 1000)
     if state.stable_tier >= Tier.HOT:
@@ -144,7 +148,7 @@ def fast_promote_on_tool_use(
         entity_id=state.entity_id,
         from_tier=old,
         to_tier=target,
-        reason="tool_used",
+        reason="skill_used" if state.kind == "skill" else "tool_used",
         temporary=True,
     )