feat(skills): add runtime skill activation

src/openharness/commands/registry.py

@@ -49,6 +49,7 @@
 )
 from openharness.services.session_storage import get_project_session_dir, load_session_snapshot
 from openharness.skills import load_skill_registry
+from openharness.skills.runtime import activate_skill, apply_skill_overrides
 from openharness.tasks import get_task_manager
 
 if TYPE_CHECKING:
@@ -79,6 +80,19 @@ class CommandContext:
     app_state: AppStateStore | None = None
 
 
+def _refresh_engine_prompt(context: CommandContext, latest_user_prompt: str | None = None) -> Settings:
+    settings = apply_skill_overrides(load_settings(), context.engine.active_skill)
+    context.engine.set_model(settings.model)
+    context.engine.set_system_prompt(
+        build_runtime_system_prompt(
+            settings,
+            cwd=context.cwd,
+            latest_user_prompt=latest_user_prompt,
+        )
+    )
+    return settings
+
+
 CommandHandler = Callable[[str, CommandContext], Awaitable[CommandResult]]
 
 
@@ -666,9 +680,32 @@ async def _skills_handler(args: str, context: CommandContext) -> CommandResult:
         lines = ["Available skills:"]
         for skill in skills:
             source = f" [{skill.source}]"
-            lines.append(f"- {skill.name}{source}: {skill.description}")
+            flags: list[str] = []
+            if skill.user_invocable:
+                flags.append("slash")
+            if skill.model_invocable:
+                flags.append("tool")
+            suffix = f" ({', '.join(flags)})" if flags else ""
+            lines.append(f"- {skill.name}{source}{suffix}: {skill.description}")
         return CommandResult(message="\n".join(lines))
 
+    def _make_skill_command(skill_name: str, skill_description: str):
+        async def _handler(args: str, context: CommandContext) -> CommandResult:
+            del args
+            active_skill = activate_skill(skill_name, context.cwd)
+            if active_skill is None:
+                return CommandResult(message=f"Skill not found: {skill_name}")
+            context.engine.set_active_skill(active_skill)
+            settings = _refresh_engine_prompt(context)
+            return CommandResult(
+                message=(
+                    f"Activated skill /{skill_name} using model {settings.model}.\n\n"
+                    f"{active_skill.definition.instructions}"
+                )
+            )
+
+        return SlashCommand(skill_name, skill_description, _handler)
+
     async def _config_handler(args: str, context: CommandContext) -> CommandResult:
         del context
         settings = load_settings()
@@ -1317,4 +1354,12 @@ async def _tasks_handler(args: str, context: CommandContext) -> CommandResult:
     registry.register(SlashCommand("upgrade", "Show upgrade instructions", _upgrade_handler))
     registry.register(SlashCommand("agents", "List or inspect agent and teammate tasks", _agents_handler))
     registry.register(SlashCommand("tasks", "Manage background tasks", _tasks_handler))
+
+    skill_registry = load_skill_registry(Path.cwd())
+    built_in_names = {command.name for command in registry._commands.values()}
+    for skill in skill_registry.list_user_invocable():
+        normalized = skill.name.strip().lower()
+        if normalized in built_in_names or normalized in {alias.strip().lower() for alias in skill.aliases}:
+            continue
+        registry.register(_make_skill_command(skill.name, skill.description))
     return registry

src/openharness/engine/query.py

@@ -24,6 +24,7 @@
 )
 from openharness.hooks import HookEvent, HookExecutor
 from openharness.permissions.checker import PermissionChecker
+from openharness.skills.runtime import build_effective_system_prompt, filter_tool_registry
 from openharness.tools.base import ToolExecutionContext
 from openharness.tools.base import ToolRegistry
 
@@ -50,6 +51,25 @@ class QueryContext:
     tool_metadata: dict[str, object] | None = None
 
 
+def _build_turn_request(context: QueryContext) -> ApiMessageRequest:
+    """Build the next model request from the latest runtime state."""
+    engine = context.tool_metadata.get("query_engine") if context.tool_metadata else None
+    active_skill = getattr(engine, "active_skill", None)
+    model = active_skill.model_override if active_skill is not None and active_skill.model_override else context.model
+    system_prompt = build_effective_system_prompt(context.system_prompt, active_skill)
+    tool_registry = filter_tool_registry(
+        context.tool_registry,
+        active_skill.allowed_tools if active_skill is not None else None,
+    )
+    return ApiMessageRequest(
+        model=model,
+        messages=[],
+        system_prompt=system_prompt,
+        max_tokens=context.max_tokens,
+        tools=tool_registry.to_api_schema(),
+    )
+
+
 async def run_query(
     context: QueryContext,
     messages: list[ConversationMessage],
@@ -70,28 +90,29 @@ async def run_query(
     compact_state = AutoCompactState()
 
     for _ in range(context.max_turns):
+        request = _build_turn_request(context)
+
         # --- auto-compact check before calling the model ---------------
         messages, was_compacted = await auto_compact_if_needed(
             messages,
             api_client=context.api_client,
-            model=context.model,
-            system_prompt=context.system_prompt,
+            model=request.model,
+            system_prompt=request.system_prompt,
             state=compact_state,
         )
         # ---------------------------------------------------------------
 
         final_message: ConversationMessage | None = None
         usage = UsageSnapshot()
+        request = ApiMessageRequest(
+            model=request.model,
+            messages=messages,
+            system_prompt=request.system_prompt,
+            max_tokens=request.max_tokens,
+            tools=request.tools,
+        )
 
-        async for event in context.api_client.stream_message(
-            ApiMessageRequest(
-                model=context.model,
-                messages=messages,
-                system_prompt=context.system_prompt,
-                max_tokens=context.max_tokens,
-                tools=context.tool_registry.to_api_schema(),
-            )
-        ):
+        async for event in context.api_client.stream_message(request):
             if isinstance(event, ApiTextDeltaEvent):
                 yield AssistantTextDelta(text=event.text), None
                 continue

src/openharness/engine/query_engine.py

@@ -12,6 +12,7 @@
 from openharness.engine.stream_events import StreamEvent
 from openharness.hooks import HookExecutor
 from openharness.permissions.checker import PermissionChecker
+from openharness.skills.runtime import ActiveSkillContext
 from openharness.tools.base import ToolRegistry
 
 
@@ -46,6 +47,7 @@ def __init__(
         self._tool_metadata = tool_metadata or {}
         self._messages: list[ConversationMessage] = []
         self._cost_tracker = CostTracker()
+        self._active_skill: ActiveSkillContext | None = None
 
     @property
     def messages(self) -> list[ConversationMessage]:
@@ -74,6 +76,15 @@ def set_permission_checker(self, checker: PermissionChecker) -> None:
         """Update the active permission checker for future turns."""
         self._permission_checker = checker
 
+    def set_active_skill(self, active_skill: ActiveSkillContext | None) -> None:
+        """Update the active skill scope for future turns."""
+        self._active_skill = active_skill
+
+    @property
+    def active_skill(self) -> ActiveSkillContext | None:
+        """Return the currently active skill context."""
+        return self._active_skill
+
     def load_messages(self, messages: list[ConversationMessage]) -> None:
         """Replace the in-memory conversation history."""
         self._messages = list(messages)

src/openharness/plugins/loader.py

@@ -8,7 +8,7 @@
 from openharness.config.paths import get_config_dir
 from openharness.plugins.schemas import PluginManifest
 from openharness.plugins.types import LoadedPlugin
-from openharness.skills.loader import _parse_skill_markdown
+from openharness.skills.markdown import load_skills_from_directory
 from openharness.skills.types import SkillDefinition
 
 
@@ -107,22 +107,7 @@ def load_plugin(path: Path, enabled_plugins: dict[str, bool]) -> LoadedPlugin |
 
 
 def _load_plugin_skills(path: Path) -> list[SkillDefinition]:
-    if not path.exists():
-        return []
-    skills: list[SkillDefinition] = []
-    for skill_path in sorted(path.glob("*.md")):
-        content = skill_path.read_text(encoding="utf-8")
-        name, description = _parse_skill_markdown(skill_path.stem, content)
-        skills.append(
-            SkillDefinition(
-                name=name,
-                description=description,
-                content=content,
-                source="plugin",
-                path=str(skill_path),
-            )
-        )
-    return skills
+    return load_skills_from_directory(path, source="plugin")
 
 
 def _load_plugin_hooks(path: Path) -> dict[str, list]:

src/openharness/prompts/context.py

@@ -10,12 +10,13 @@
 from openharness.prompts.claudemd import load_claude_md_prompt
 from openharness.prompts.system_prompt import build_system_prompt
 from openharness.skills.loader import load_skill_registry
+from openharness.skills.runtime import ActiveSkillContext, build_active_skill_section
 
 
 def _build_skills_section(cwd: str | Path) -> str | None:
     """Build a system prompt section listing available skills."""
     registry = load_skill_registry(cwd)
-    skills = registry.list_skills()
+    skills = registry.list_model_invocable()
     if not skills:
         return None
     lines = [
@@ -36,6 +37,7 @@ def build_runtime_system_prompt(
     *,
     cwd: str | Path,
     latest_user_prompt: str | None = None,
+    active_skill: ActiveSkillContext | None = None,
 ) -> str:
     """Build the runtime system prompt with project instructions and memory."""
     sections = [build_system_prompt(custom_prompt=settings.system_prompt, cwd=str(cwd))]
@@ -56,6 +58,9 @@ def build_runtime_system_prompt(
     if skills_section:
         sections.append(skills_section)
 
+    if active_skill is not None:
+        sections.append(build_active_skill_section(active_skill))
+
     claude_md = load_claude_md_prompt(cwd)
     if claude_md:
         sections.append(claude_md)

src/openharness/skills/bundled/__init__.py

@@ -4,41 +4,12 @@
 
 from pathlib import Path
 
+from openharness.skills.markdown import load_skills_from_directory
 from openharness.skills.types import SkillDefinition
 
 _CONTENT_DIR = Path(__file__).parent / "content"
 
 
 def get_bundled_skills() -> list[SkillDefinition]:
     """Load all bundled skills from the content/ directory."""
-    skills: list[SkillDefinition] = []
-    if not _CONTENT_DIR.exists():
-        return skills
-    for path in sorted(_CONTENT_DIR.glob("*.md")):
-        content = path.read_text(encoding="utf-8")
-        name, description = _parse_frontmatter(path.stem, content)
-        skills.append(
-            SkillDefinition(
-                name=name,
-                description=description,
-                content=content,
-                source="bundled",
-                path=str(path),
-            )
-        )
-    return skills
-
-
-def _parse_frontmatter(default_name: str, content: str) -> tuple[str, str]:
-    """Extract name and description from a skill markdown file."""
-    name = default_name
-    description = ""
-    for line in content.splitlines():
-        stripped = line.strip()
-        if stripped.startswith("# "):
-            name = stripped[2:].strip() or default_name
-            continue
-        if stripped and not stripped.startswith("#"):
-            description = stripped
-            break
-    return name, description or f"Bundled skill: {name}"
+    return load_skills_from_directory(_CONTENT_DIR, source="bundled")

src/openharness/skills/loader.py

@@ -7,6 +7,7 @@
 from openharness.config.paths import get_config_dir
 from openharness.config.settings import load_settings
 from openharness.skills.bundled import get_bundled_skills
+from openharness.skills.markdown import load_skills_from_directory, parse_skill_markdown
 from openharness.skills.registry import SkillRegistry
 from openharness.skills.types import SkillDefinition
 
@@ -39,58 +40,12 @@ def load_skill_registry(cwd: str | Path | None = None) -> SkillRegistry:
 
 def load_user_skills() -> list[SkillDefinition]:
     """Load markdown skills from the user config directory."""
-    skills: list[SkillDefinition] = []
-    for path in sorted(get_user_skills_dir().glob("*.md")):
-        content = path.read_text(encoding="utf-8")
-        name, description = _parse_skill_markdown(path.stem, content)
-        skills.append(
-            SkillDefinition(
-                name=name,
-                description=description,
-                content=content,
-                source="user",
-                path=str(path),
-            )
-        )
-    return skills
+    return load_skills_from_directory(get_user_skills_dir(), source="user")
 
 
-def _parse_skill_markdown(default_name: str, content: str) -> tuple[str, str]:
-    """Parse name and description from a skill markdown file with YAML frontmatter support."""
-    name = default_name
-    description = ""
-
-    lines = content.splitlines()
-
-    # Try YAML frontmatter first (--- ... ---)
-    if lines and lines[0].strip() == "---":
-        for i, line in enumerate(lines[1:], 1):
-            if line.strip() == "---":
-                # Parse frontmatter fields
-                for fm_line in lines[1:i]:
-                    fm_stripped = fm_line.strip()
-                    if fm_stripped.startswith("name:"):
-                        val = fm_stripped[5:].strip().strip("'\"")
-                        if val:
-                            name = val
-                    elif fm_stripped.startswith("description:"):
-                        val = fm_stripped[12:].strip().strip("'\"")
-                        if val:
-                            description = val
-                break
-
-    # Fallback: extract from headings and first paragraph
-    if not description:
-        for line in lines:
-            stripped = line.strip()
-            if stripped.startswith("# "):
-                if not name or name == default_name:
-                    name = stripped[2:].strip() or default_name
-                continue
-            if stripped and not stripped.startswith("---") and not stripped.startswith("#"):
-                description = stripped[:200]
-                break
-
-    if not description:
-        description = f"Skill: {name}"
-    return name, description
+__all__ = [
+    "get_user_skills_dir",
+    "load_skill_registry",
+    "load_user_skills",
+    "parse_skill_markdown",
+]