"""SkillRunner — orchestrates skill activation, chaining, config scoping, and deactivation.""" from __future__ import annotations import logging from dataclasses import dataclass, field from app.agent.context import SessionContext from app.models.config import AppConfig from app.models.skill import SkillManifest from app.services.skills import Skill, SkillsManager from app.tools.registry import ToolRegistry logger = logging.getLogger(__name__) class SkillChainError(Exception): """Raised when skill chain resolution fails (e.g., cycle detected).""" @dataclass class _SkillSnapshot: """Captured state before skill activation, for restoration on deactivate.""" temperature: float max_tokens: int disabled_tools: set[str] = field(default_factory=set) class SkillRunner: """Manages skill lifecycle: activation, chaining, config overrides, deactivation. Only one skill can be active at a time. Activating a new skill while one is active will first deactivate the current skill. """ def __init__( self, skills_manager: SkillsManager, config: AppConfig, ctx: SessionContext, registry: ToolRegistry, ) -> None: self._skills = skills_manager self._config = config self._ctx = ctx self._registry = registry self._active_skill: Skill | None = None self._snapshot: _SkillSnapshot | None = None @property def is_active(self) -> bool: """Whether a skill is currently active.""" return self._active_skill is not None @property def active_skill_name(self) -> str | None: """Name of the currently active skill, or None.""" return self._active_skill.name if self._active_skill else None @property def active_skill(self) -> Skill | None: """The currently active skill, or None.""" return self._active_skill def activate(self, skill_name: str) -> str | None: """Activate a skill by name. Resolves chain dependencies (depth-first), applies config overrides, injects prompt content into conversation context. Args: skill_name: Name of the skill to activate. Returns: The concatenated prompt content injected, or None on failure. Raises: SkillChainError: If chain resolution detects a cycle. """ skill = self._skills.get_skill(skill_name) if skill is None: logger.warning("Cannot activate unknown skill: %s", skill_name) return None # Deactivate current skill if one is active if self._active_skill is not None: self.deactivate() # Resolve chain dependencies chain = self._resolve_chain(skill, set()) # Snapshot current config for restoration self._snapshot = _SkillSnapshot( temperature=self._config.llm.temperature, max_tokens=self._config.llm.max_tokens, ) # Collect and inject chain skill prompts first all_prompts: list[str] = [] for chained_skill in chain: content = self._skills.load_skill(chained_skill.name) if content: all_prompts.append(f"[Chained skill: {chained_skill.name}]\n{content}") # Load the target skill's prompts content = self._skills.load_skill(skill.name) if content: all_prompts.append(content) # Apply config overrides from the target skill if skill.manifest: self._apply_overrides(skill.manifest) # Inject prompts into context full_prompt = "\n\n".join(all_prompts) if all_prompts else None if full_prompt: self._ctx.add_message( "system", f"[Skill activated: {skill.name}]\n{full_prompt}", ) self._active_skill = skill logger.info("Skill activated: %s", skill.name) return full_prompt def activate_by_trigger(self, trigger: str) -> str | None: """Activate a skill by its /command trigger. Args: trigger: The trigger string (with or without leading /). Returns: The concatenated prompt content, or None if no skill matches. """ skill = self._skills.get_skill_by_trigger(trigger) if skill is None: return None return self.activate(skill.name) def deactivate(self, summary: str | None = None) -> None: """Deactivate the current skill, restoring config and tool state. Args: summary: Optional summary message to inject into context. """ if self._active_skill is None: return skill_name = self._active_skill.name # Restore config if self._snapshot is not None: self._config.llm.temperature = self._snapshot.temperature self._config.llm.max_tokens = self._snapshot.max_tokens self._registry.restore_filter(self._snapshot.disabled_tools) self._snapshot = None if summary: self._ctx.add_message( "system", f"[Skill completed: {skill_name}] {summary}", ) self._active_skill = None logger.info("Skill deactivated: %s", skill_name) def _resolve_chain( self, skill: Skill, in_progress: set[str], completed: set[str] | None = None, ) -> list[Skill]: """Depth-first resolution of skill chain dependencies. Uses separate in_progress (current path) and completed sets to correctly handle diamond dependencies without false cycle detection. Args: skill: The skill whose chain to resolve. in_progress: Skills on the current recursion path (for cycle detection). completed: Skills already fully resolved (skip duplicates). Returns: Ordered list of chained skills to activate before the target. Raises: SkillChainError: If a cycle is detected. """ if completed is None: completed = set() if skill.manifest is None or not skill.manifest.chain: return [] result: list[Skill] = [] for dep_name in skill.manifest.chain: if dep_name in completed: continue # Already resolved via another branch (diamond dep) if dep_name in in_progress: raise SkillChainError( f"Cycle detected in skill chain: {dep_name} already in progress " f"(path: {' -> '.join(in_progress)} -> {dep_name})" ) dep_skill = self._skills.get_skill(dep_name) if dep_skill is None: logger.warning("Chained skill not found: %s (required by %s)", dep_name, skill.name) continue in_progress.add(dep_name) result.extend(self._resolve_chain(dep_skill, in_progress, completed)) in_progress.discard(dep_name) completed.add(dep_name) result.append(dep_skill) return result def _apply_overrides(self, manifest: SkillManifest) -> None: """Apply config overrides from a skill manifest.""" overrides = manifest.config_overrides if overrides.temperature is not None: self._config.llm.temperature = overrides.temperature if overrides.max_tokens is not None: self._config.llm.max_tokens = overrides.max_tokens if overrides.tools_enable is not None or overrides.tools_disable is not None: previous = self._registry.apply_filter( enable=overrides.tools_enable, disable=overrides.tools_disable, ) # Store for restoration if self._snapshot: self._snapshot.disabled_tools = previous