diff --git a/docs/gns3-copilot/implemented/skills-repository.md b/docs/gns3-copilot/implemented/skills-repository.md
index aa9c8f2db..167941465 100644
--- a/docs/gns3-copilot/implemented/skills-repository.md
+++ b/docs/gns3-copilot/implemented/skills-repository.md
@@ -14,7 +14,7 @@ GNS3 Copilot loads all skills, prompts, and security configurations from an exte
The repository provides:
- **Injection skills** (39 categories): Network fault scenarios for troubleshooting practice
-- **Device skills**: Device-specific command knowledge (VPCS, etc.)
+- **Device skills**: Device-specific command knowledge (VPCS, etc.) — large devices split into per-protocol **topics**
- **Feature skills**: Topology planning, network design
- **System prompts**: Agent personality and behavior definitions
- **Forbidden commands**: Security rules for command filtering
@@ -24,7 +24,7 @@ The repository provides:
```mermaid
graph TD
subgraph "GNS3-Skills Repository"
- YAML[injection/*.yaml
device/*.yaml
feature/*.yaml]
+ YAML[injection/*.yaml
device/*.yaml + device/*/*.yaml
feature/*.yaml]
MD[prompts/*.md]
CFG[config/forbidden_commands.txt]
end
@@ -56,7 +56,11 @@ GNS3-Skills/
│ ├── vlan_issues.yaml
│ └── ...
├── device/ # Device-specific skills
-│ └── vpcs.yaml
+│ ├── vpcs.yaml # small devices: one single file
+│ └── frr/ # large devices: split per protocol topic
+│ ├── _base.yaml # device-level skill (console model, notes, aliases)
+│ ├── ospf.yaml # topic file (merged under "topics" at load time)
+│ └── bgp.yaml
├── feature/ # Feature skills
│ └── topology_planner.yaml
├── prompts/ # System prompts (Markdown)
@@ -68,6 +72,26 @@ GNS3-Skills/
└── forbidden_commands.txt
```
+## Device Topics
+
+A device with knowledge for many protocols would grow one YAML file indefinitely. Such devices use a split layout instead: `device//_base.yaml` holds the device-level skill, and one file per protocol topic (`ospf.yaml`, `bgp.yaml`, ...) holds its commands and troubleshooting entries. The loader merges them into a single `SKILLS_REGISTRY` entry:
+
+```
+SKILLS_REGISTRY["frr_vtysh"] = { ..._base.yaml..., "topics": { "ospf": {...}, "bgp": {...} } }
+```
+
+Topic files must declare `device_type` (matching their `_base.yaml`), `topic` and `name`; the CI validator in the skills repository enforces this.
+
+The `device_skills` tool exposes a three-step drill-down (mirroring `injection_skills`'s list → index → issue pattern):
+
+```json
+{"action": "list"}
+{"device_type": "frr_vtysh", "detail": "index"}
+{"device_type": "frr_vtysh", "topic": "bgp"}
+```
+
+Topic bodies are only served on an explicit `topic` request — every other detail level returns a topic index — so adding topics to a device does not grow the token cost of device-level lookups.
+
## Configuration
Skills repository settings are configured in `gns3_server.conf` under the `[Server]` section:
diff --git a/gns3server/agent/gns3_copilot/skills/loader.py b/gns3server/agent/gns3_copilot/skills/loader.py
index 4d585cac5..fee183ef6 100644
--- a/gns3server/agent/gns3_copilot/skills/loader.py
+++ b/gns3server/agent/gns3_copilot/skills/loader.py
@@ -100,6 +100,12 @@ class SkillsLoader:
"""
Load all device skills from YAML files.
+ Supports two layouts:
+ - Single file: device/.yaml
+ - Split directory: device//_base.yaml + device//.yaml
+ (topic files are merged into the base skill under "topics",
+ keyed by their "topic" field)
+
Returns:
Dictionary mapping skill keys to skill definitions
"""
@@ -107,33 +113,110 @@ class SkillsLoader:
logger.error("PyYAML is not installed. Cannot load skills from YAML.")
return {}
- skills = {}
+ skills: Dict[str, Dict[str, Any]] = {}
device_dir = self.skills_dir / "device"
if not device_dir.exists():
logger.warning(f"Device skills directory not found: {device_dir}")
return {}
- for yaml_file in device_dir.glob("*.yaml"):
- try:
- skill_data = self._load_yaml(yaml_file)
- if not skill_data:
- logger.warning(f"Skipping empty YAML file: {yaml_file}")
- continue
- # Use device_type from YAML content as the key
- # Fallback to filename stem if device_type not present
- skill_key = skill_data.get("device_type") if isinstance(skill_data, dict) else None
- if not skill_key:
- skill_key = yaml_file.stem
- logger.warning(f"No device_type in {yaml_file}, using filename '{skill_key}' as key")
- skills[skill_key] = skill_data
- logger.debug(f"Loaded device skill: {skill_key} from {yaml_file}")
- except Exception as e:
- logger.error(f"Failed to load skill from {yaml_file}: {e}")
+ for entry in sorted(device_dir.iterdir()):
+ if entry.is_file() and entry.suffix == ".yaml":
+ self._load_single_device_skill(skills, entry)
+ elif entry.is_dir():
+ self._load_split_device_skill(skills, entry)
logger.debug(f"Loaded {len(skills)} device skills from device directory")
return skills
+ def _load_single_device_skill(self, skills: Dict[str, Dict[str, Any]], yaml_file: Path) -> None:
+ """
+ Load a single-file device skill into the skills dictionary.
+ """
+ try:
+ skill_data = self._load_yaml(yaml_file)
+ if not skill_data:
+ logger.warning(f"Skipping empty YAML file: {yaml_file}")
+ return
+ # Use device_type from YAML content as the key
+ # Fallback to filename stem if device_type not present
+ skill_key = skill_data.get("device_type")
+ if not skill_key:
+ skill_key = yaml_file.stem
+ logger.warning(f"No device_type in {yaml_file}, using filename '{skill_key}' as key")
+ skills[skill_key] = skill_data
+ logger.debug(f"Loaded device skill: {skill_key} from {yaml_file}")
+ except Exception as e:
+ logger.error(f"Failed to load skill from {yaml_file}: {e}")
+
+ def _load_split_device_skill(self, skills: Dict[str, Dict[str, Any]], device_path: Path) -> None:
+ """
+ Load a split device skill (directory with _base.yaml + topic files).
+
+ The base file provides the device-level skill; every other YAML file
+ in the directory is a protocol topic merged under "topics".
+ """
+ base_file = device_path / "_base.yaml"
+ if not base_file.exists():
+ logger.error(f"No _base.yaml in device directory: {device_path}, skipping")
+ return
+
+ try:
+ base_data = self._load_yaml(base_file)
+ except Exception as e:
+ logger.error(f"Failed to load skill from {base_file}: {e}")
+ return
+ if not base_data:
+ logger.warning(f"Skipping empty YAML file: {base_file}")
+ return
+
+ skill_key = base_data.get("device_type")
+ if not skill_key:
+ skill_key = device_path.name
+ logger.warning(f"No device_type in {base_file}, using directory name '{skill_key}' as key")
+
+ # Seed topics from the base file (if any), then merge topic files
+ base_topics = base_data.get("topics")
+ topics: Dict[str, Any] = dict(base_topics) if isinstance(base_topics, dict) else {}
+
+ for yaml_file in sorted(device_path.glob("*.yaml")):
+ if yaml_file.name == "_base.yaml":
+ continue
+ try:
+ topic_data = self._load_yaml(yaml_file)
+ if not topic_data:
+ logger.warning(f"Skipping empty YAML file: {yaml_file}")
+ continue
+
+ topic_device_type = topic_data.pop("device_type", None)
+ if topic_device_type is not None and topic_device_type != skill_key:
+ logger.error(
+ f"device_type mismatch in {yaml_file}: '{topic_device_type}' "
+ f"!= base '{skill_key}', skipping topic"
+ )
+ continue
+
+ topic_key = topic_data.pop("topic", None)
+ if not topic_key:
+ topic_key = yaml_file.stem
+ logger.warning(f"No 'topic' field in {yaml_file}, using filename '{topic_key}'")
+ if topic_key in topics:
+ logger.warning(f"Duplicate topic '{topic_key}' in {device_path.name} (from {yaml_file}), overwriting")
+
+ # category/topics belong to the base skill only
+ topic_data.pop("category", None)
+ topic_data.pop("topics", None)
+
+ topics[topic_key] = topic_data
+ logger.debug(f"Loaded device topic: {skill_key}/{topic_key} from {yaml_file}")
+ except Exception as e:
+ logger.error(f"Failed to load topic from {yaml_file}: {e}")
+
+ if topics:
+ base_data["topics"] = topics
+ skills[skill_key] = base_data
+ logger.debug(f"Loaded device skill: {skill_key} from {device_path} ({len(topics)} topics)")
+
def load_feature_skills(self) -> Dict[str, Dict[str, Any]]:
"""
Load all feature skills from YAML files.
diff --git a/gns3server/agent/gns3_copilot/skills/manager.py b/gns3server/agent/gns3_copilot/skills/manager.py
index 6a663e0e8..e14b8a8dc 100644
--- a/gns3server/agent/gns3_copilot/skills/manager.py
+++ b/gns3server/agent/gns3_copilot/skills/manager.py
@@ -221,11 +221,18 @@ class SkillsManager:
logger.warning("No injection skills loaded, keeping existing skills")
return False
- # Validate injection skills
+ # Validate injection skills (drop invalid ones before merging)
+ valid_injection_skills = {}
for skill_key, skill_data in new_injection_skills.items():
- if not self.loader.validate_skill_format(skill_data):
+ if self.loader.validate_skill_format(skill_data):
+ valid_injection_skills[skill_key] = skill_data
+ else:
logger.error(f"Invalid skill format for {skill_key}, skipping")
- continue
+
+ if not valid_injection_skills:
+ logger.warning("No valid injection skills loaded, keeping existing skills")
+ return False
+ new_injection_skills = valid_injection_skills
# Load new device skills from YAML files
new_device_skills = self.loader.load_device_skills()
diff --git a/gns3server/agent/gns3_copilot/skills/registry.py b/gns3server/agent/gns3_copilot/skills/registry.py
index 4d0cf1a53..a003413be 100644
--- a/gns3server/agent/gns3_copilot/skills/registry.py
+++ b/gns3server/agent/gns3_copilot/skills/registry.py
@@ -373,6 +373,7 @@ def get_skill(
category: str | None = None,
detail: str = "full",
issue: str | None = None,
+ topic: str | None = None,
) -> dict[str, Any]:
"""
Get skill by device_type, with configurable detail level.
@@ -382,6 +383,9 @@ def get_skill(
category: Optional category filter
detail: Detail level - "index" (names only), "summary" (+desc/sev/diff), "full" (all)
issue: Optional specific issue key to retrieve
+ topic: Optional protocol topic to retrieve (split devices only).
+ Topic bodies are NEVER included without an explicit topic
+ request - all other detail levels return a topic index.
Returns:
Skill dictionary (detail varies by level), or error dict
@@ -412,6 +416,28 @@ def get_skill(
],
}
+ topics = skill.get("topics", {})
+
+ # Single topic lookup (topic bodies stay out of every other response)
+ if topic:
+ topic_data = topics.get(topic)
+ if not topic_data:
+ for key, data in topics.items():
+ if key.lower() == topic.lower():
+ topic_data = data
+ topic = key
+ break
+ if not topic_data:
+ return {
+ "error": f"Unknown topic '{topic}' in {device_type}",
+ "available_topics": list(topics.keys()),
+ }
+ return {
+ "device_type": device_type,
+ "skill_name": skill.get("name"),
+ "topic": {topic: topic_data},
+ }
+
issues = skill.get("issues", {})
# Single issue lookup (most token-efficient)
@@ -429,17 +455,20 @@ def get_skill(
}
if detail == "index":
- # Minimal: only issue keys and names (90%+ token savings)
- return {
+ # Minimal: only issue/topic keys and names (90%+ token savings)
+ result = {
"device_type": device_type,
"name": skill.get("name"),
"description": skill.get("description"),
"issues": {k: v["name"] for k, v in issues.items()},
}
+ if topics:
+ result["topics"] = {k: v.get("name", k) for k, v in topics.items()}
+ return result
if detail == "summary":
# Moderate: names + description + severity + difficulty
- return {
+ result = {
"device_type": device_type,
"name": skill.get("name"),
"description": skill.get("description"),
@@ -453,14 +482,22 @@ def get_skill(
for k, v in issues.items()
},
}
+ if topics:
+ result["topics"] = {
+ k: {"name": v.get("name", k), "description": v.get("description", "")}
+ for k, v in topics.items()
+ }
+ return result
- # Full detail (original behavior)
- result = dict(skill)
+ # Full detail: topic bodies are replaced by the topic index
+ result = {k: v for k, v in skill.items() if k != "topics"}
result["device_type"] = device_type
+ if topics:
+ result["topics"] = {k: v.get("name", k) for k, v in topics.items()}
return result
-def list_available_skills(category: str | None = None) -> list[dict[str, str]]:
+def list_available_skills(category: str | None = None) -> list[dict[str, Any]]:
"""List all available device/feature skills, optionally filtered by category."""
skills = []
for did, skill in SKILLS_REGISTRY.items():
@@ -470,12 +507,14 @@ def list_available_skills(category: str | None = None) -> list[dict[str, str]]:
"device_type": did,
"name": skill.get("name", did),
"category": skill.get("category"),
+ "topic_count": len(skill.get("topics", {})),
})
else:
skills.append({
"device_type": did,
"name": skill.get("name", did),
"category": skill.get("category"),
+ "topic_count": len(skill.get("topics", {})),
})
return skills
@@ -642,15 +681,21 @@ class DeviceSkillsTool(BaseTool):
Provides access to device command knowledge (VPCS), topology planning, etc.
For fault injection skills, use the injection_skills tool.
- INPUT FORMAT (JSON string):
- {
- "action": "get", # "get" (default) or "list"
- "device_type": "gns3_vpcs_telnet", # Required for action="get"
- "detail": "full" # "full" (default) for complete skill information
- }
+ TOKEN-EFFICIENT USAGE:
+ 1. List devices: {"action": "list"}
+ 2. List topics of a device: {"device_type": "frr_vtysh", "detail": "index"}
+ 3. Get ONE protocol topic (devices with topics): {"device_type": "frr_vtysh", "topic": "bgp"}
+ 4. Devices without topics: {"device_type": "gns3_vpcs_telnet"}
- For action="list":
- {"action": "list"} # Lists all available device/feature skills
+ Topic bodies are NEVER returned without an explicit "topic" - fetching a
+ device without one only returns its base skill plus the topic index, so
+ always request the specific protocol topic before configuring it.
+
+ PARAMETERS:
+ - action: "list" or "get" (default "get")
+ - device_type: Required for action="get" (e.g., "frr_vtysh")
+ - topic: Protocol topic key from the topic index (e.g., "ospf", "bgp")
+ - detail: "index" | "summary" | "full" (default "full")
"""
def _run(
@@ -693,8 +738,9 @@ class DeviceSkillsTool(BaseTool):
category = params.get("category")
detail = params.get("detail", "full")
issue = params.get("issue")
+ topic = params.get("topic")
- skill = get_skill(device_type, category, detail=detail, issue=issue)
+ skill = get_skill(device_type, category, detail=detail, issue=issue, topic=topic)
return json.dumps(skill, ensure_ascii=False, indent=2)
diff --git a/tests/agent/test_skills_device_topics.py b/tests/agent/test_skills_device_topics.py
new file mode 100644
index 000000000..68d8371c5
--- /dev/null
+++ b/tests/agent/test_skills_device_topics.py
@@ -0,0 +1,348 @@
+#!/usr/bin/env python
+#
+# Copyright (C) 2026 GNS3 Technologies Inc.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program. If not, see .
+
+"""
+Tests for device skill topic splitting: directory layout loading
+(_base.yaml + topic files) and topic-level retrieval via get_skill /
+DeviceSkillsTool.
+"""
+
+import pytest
+
+from gns3server.agent.gns3_copilot.skills.loader import SkillsLoader
+from gns3server.agent.gns3_copilot.skills.registry import (
+ DeviceSkillsTool,
+ get_skill,
+)
+
+
+@pytest.fixture
+def skills_dir(tmp_path):
+ """
+ Build a skills directory with both device layouts:
+
+ - vpcs.yaml: single-file device (no topics)
+ - frr/: split device (_base.yaml + ospf/bgp topic files,
+ one mismatched topic file, one file without a topic field)
+ - orphan/: directory without _base.yaml (skipped)
+ """
+ device_dir = tmp_path / "device"
+ device_dir.mkdir()
+
+ (device_dir / "vpcs.yaml").write_text(
+ """
+name: "VPCS"
+description: "VPCS test device"
+device_type: "gns3_vpcs_telnet"
+category: "device"
+config_commands:
+ ip_config:
+ syntax: "ip / "
+ description: "Set PC address"
+""",
+ encoding="utf-8",
+ )
+
+ frr_dir = device_dir / "frr"
+ frr_dir.mkdir()
+ (frr_dir / "_base.yaml").write_text(
+ """
+name: "FRR (Free Range Routing)"
+description: "FRR test device"
+device_type: "frr_vtysh"
+category: "device"
+config_commands:
+ write_memory:
+ syntax: "write memory"
+ description: "Save config"
+""",
+ encoding="utf-8",
+ )
+ (frr_dir / "ospf.yaml").write_text(
+ """
+device_type: "frr_vtysh"
+topic: ospf
+name: "OSPF (FRR 10.x)"
+description: "OSPF topic"
+config_commands:
+ ospfv2:
+ syntax: "router ospf"
+ description: "OSPFv2 process"
+""",
+ encoding="utf-8",
+ )
+ (frr_dir / "bgp.yaml").write_text(
+ """
+device_type: "frr_vtysh"
+topic: bgp
+name: "BGP (FRR 10.x)"
+description: "BGP topic"
+config_commands:
+ bgp_base:
+ syntax: "router bgp "
+ description: "BGP base"
+""",
+ encoding="utf-8",
+ )
+ # device_type mismatch with the base -> topic must be skipped
+ (frr_dir / "mpls.yaml").write_text(
+ """
+device_type: "other_device_type"
+topic: mpls
+name: "MPLS"
+config_commands:
+ mpls_base:
+ syntax: "router mpls"
+ description: "..."
+""",
+ encoding="utf-8",
+ )
+ # no topic field -> falls back to the filename stem
+ (frr_dir / "static.yaml").write_text(
+ """
+device_type: "frr_vtysh"
+name: "Static routing"
+config_commands:
+ static_routes:
+ syntax: "ip route / "
+ description: "..."
+""",
+ encoding="utf-8",
+ )
+
+ # directory without _base.yaml -> whole device skipped
+ orphan_dir = device_dir / "orphan"
+ orphan_dir.mkdir()
+ (orphan_dir / "some_topic.yaml").write_text(
+ """
+device_type: "orphan_device"
+topic: anything
+name: "Orphan"
+""",
+ encoding="utf-8",
+ )
+
+ return tmp_path
+
+
+class TestDeviceSkillsLoading:
+ """
+ SkillsLoader.load_device_skills() with single-file and split layouts.
+ """
+
+ def test_both_layouts_are_loaded(self, skills_dir):
+ skills = SkillsLoader(str(skills_dir)).load_device_skills()
+
+ assert "gns3_vpcs_telnet" in skills
+ assert "frr_vtysh" in skills
+ # the orphan directory (no _base.yaml) must not produce an entry
+ assert "orphan_device" not in skills
+ assert len(skills) == 2
+
+ def test_topics_are_merged_under_topics(self, skills_dir):
+ skills = SkillsLoader(str(skills_dir)).load_device_skills()
+ frr = skills["frr_vtysh"]
+
+ # base-level content stays at the top level
+ assert frr["config_commands"]["write_memory"]["syntax"] == "write memory"
+ assert frr["category"] == "device"
+
+ topics = frr["topics"]
+ assert topics["ospf"]["name"] == "OSPF (FRR 10.x)"
+ assert topics["bgp"]["config_commands"]["bgp_base"]["syntax"] == "router bgp "
+ # file without a topic field falls back to its filename stem
+ assert "static" in topics
+ # mismatched device_type is skipped
+ assert "mpls" not in topics
+
+ def test_topic_metadata_is_stripped(self, skills_dir):
+ skills = SkillsLoader(str(skills_dir)).load_device_skills()
+ for topic_data in skills["frr_vtysh"]["topics"].values():
+ assert "device_type" not in topic_data
+ assert "topic" not in topic_data
+ assert "category" not in topic_data
+ assert "topics" not in topic_data
+
+ def test_single_file_device_has_no_topics_key(self, skills_dir):
+ skills = SkillsLoader(str(skills_dir)).load_device_skills()
+ assert "topics" not in skills["gns3_vpcs_telnet"]
+
+
+@pytest.fixture
+def device_registry():
+ """
+ Populate SKILLS_REGISTRY with a split device and a single-file device,
+ restoring the previous content afterwards.
+ """
+ from gns3server.agent.gns3_copilot.skills import registry
+
+ saved = dict(registry.SKILLS_REGISTRY)
+ registry.SKILLS_REGISTRY.clear()
+ registry.SKILLS_REGISTRY.update(
+ {
+ "frr_vtysh": {
+ "name": "FRR (Free Range Routing)",
+ "description": "FRR test device",
+ "category": "device",
+ "config_commands": {"write_memory": {"syntax": "write memory"}},
+ "topics": {
+ "ospf": {
+ "name": "OSPF (FRR 10.x)",
+ "description": "OSPF topic",
+ "config_commands": {"ospfv2": {"syntax": "router ospf"}},
+ },
+ "bgp": {
+ "name": "BGP (FRR 10.x)",
+ "description": "BGP topic",
+ "config_commands": {"bgp_base": {"syntax": "router bgp "}},
+ },
+ },
+ },
+ "gns3_vpcs_telnet": {
+ "name": "VPCS",
+ "description": "VPCS test device",
+ "category": "device",
+ "config_commands": {"ip_config": {"syntax": "ip /"}},
+ },
+ }
+ )
+ yield registry
+ registry.SKILLS_REGISTRY.clear()
+ registry.SKILLS_REGISTRY.update(saved)
+
+
+class TestGetSkillTopics:
+ """
+ Topic-level retrieval and topic index behavior in get_skill().
+ """
+
+ def test_topic_lookup_returns_topic_body(self, device_registry):
+ result = get_skill("frr_vtysh", topic="bgp")
+ assert result["device_type"] == "frr_vtysh"
+ assert result["skill_name"] == "FRR (Free Range Routing)"
+ assert result["topic"]["bgp"]["config_commands"]["bgp_base"]["syntax"] == "router bgp "
+
+ def test_topic_lookup_is_case_insensitive(self, device_registry):
+ result = get_skill("frr_vtysh", topic="BGP")
+ assert "bgp" in result["topic"]
+
+ def test_unknown_topic_lists_available_topics(self, device_registry):
+ result = get_skill("frr_vtysh", topic="mpls")
+ assert "error" in result
+ assert sorted(result["available_topics"]) == ["bgp", "ospf"]
+
+ def test_full_without_topic_returns_index_not_bodies(self, device_registry):
+ result = get_skill("frr_vtysh", detail="full")
+ # base-level content is included...
+ assert result["config_commands"]["write_memory"]["syntax"] == "write memory"
+ # ...but topic bodies are never included without an explicit topic
+ assert result["topics"] == {
+ "ospf": "OSPF (FRR 10.x)",
+ "bgp": "BGP (FRR 10.x)",
+ }
+ assert "config_commands" not in result["topics"]["ospf"]
+
+ def test_index_includes_topic_index(self, device_registry):
+ result = get_skill("frr_vtysh", detail="index")
+ assert result["topics"]["bgp"] == "BGP (FRR 10.x)"
+
+ def test_summary_includes_topic_descriptions(self, device_registry):
+ result = get_skill("frr_vtysh", detail="summary")
+ assert result["topics"]["ospf"] == {
+ "name": "OSPF (FRR 10.x)",
+ "description": "OSPF topic",
+ }
+
+ def test_single_file_device_still_works(self, device_registry):
+ result = get_skill("gns3_vpcs_telnet")
+ assert result["config_commands"]["ip_config"]["syntax"] == "ip /"
+ assert "topics" not in result
+
+
+class TestDeviceSkillsToolTopics:
+ """
+ DeviceSkillsTool passes the topic parameter through to get_skill().
+ """
+
+ def test_tool_topic_request(self, device_registry):
+ import json
+
+ tool = DeviceSkillsTool()
+ result = json.loads(tool._run('{"device_type": "frr_vtysh", "topic": "ospf"}'))
+ assert result["topic"]["ospf"]["config_commands"]["ospfv2"]["syntax"] == "router ospf"
+
+ def test_tool_list_shows_topic_counts(self, device_registry):
+ import json
+
+ tool = DeviceSkillsTool()
+ result = json.loads(tool._run('{"action": "list"}'))
+ by_type = {s["device_type"]: s for s in result["skills"]}
+ assert by_type["frr_vtysh"]["topic_count"] == 2
+ assert by_type["gns3_vpcs_telnet"]["topic_count"] == 0
+
+
+class TestReloadSkillsValidation:
+ """
+ Invalid injection skills are dropped instead of merged into the registry.
+ """
+
+ def test_invalid_injection_skill_is_dropped(self, tmp_path, monkeypatch):
+ from gns3server.config import Config
+ from gns3server.agent.gns3_copilot.skills import registry
+ from gns3server.agent.gns3_copilot.skills.manager import SkillsManager
+
+ # SkillsManager derives its local path from /skills
+ injection_dir = tmp_path / "skills" / "injection"
+ injection_dir.mkdir(parents=True)
+ (injection_dir / "valid.yaml").write_text(
+ """
+name: "OSPF Issues Injection"
+description: "OSPF faults"
+category: "injection"
+issues:
+ ospf_area_mismatch:
+ name: "OSPF Area Mismatch"
+ description: "Areas differ"
+""",
+ encoding="utf-8",
+ )
+ # missing the required "issues" field -> invalid, must be dropped
+ (injection_dir / "broken.yaml").write_text(
+ """
+name: "Broken Injection"
+description: "No issues field"
+""",
+ encoding="utf-8",
+ )
+
+ monkeypatch.setattr(Config, "config_dir", property(lambda self: str(tmp_path)))
+
+ saved_injection = dict(registry.INJECTION_SKILLS_REGISTRY)
+ saved_skills = dict(registry.SKILLS_REGISTRY)
+ registry.INJECTION_SKILLS_REGISTRY.clear()
+ registry.SKILLS_REGISTRY.clear()
+ try:
+ manager = SkillsManager(repo_url="https://example.invalid/gns3-skills.git")
+ manager._repo = None
+ assert manager.reload_skills() is True
+ assert "injection_valid" in registry.INJECTION_SKILLS_REGISTRY
+ assert "injection_broken" not in registry.INJECTION_SKILLS_REGISTRY
+ finally:
+ registry.INJECTION_SKILLS_REGISTRY.clear()
+ registry.INJECTION_SKILLS_REGISTRY.update(saved_injection)
+ registry.SKILLS_REGISTRY.clear()
+ registry.SKILLS_REGISTRY.update(saved_skills)