Compare commits

..

2 Commits

Author SHA1 Message Date
c8bab8ae3c feat: import Anthropic Cybersecurity Skills — 754 skills (#712) 2026-04-16 01:26:45 +00:00
Timmy Time
faaa08b3f1 fix: #712
Some checks failed
Contributor Attribution Check / check-attribution (pull_request) Failing after 31s
Docker Build and Publish / build-and-push (pull_request) Has been skipped
Supply Chain Audit / Scan PR for supply chain risks (pull_request) Successful in 1m0s
Tests / e2e (pull_request) Successful in 2m13s
Tests / test (pull_request) Failing after 54m56s
Import Anthropic Cybersecurity Skills Library (754 skills, 26 domains, 5 frameworks).

Added:
- scripts/import_cybersecurity_skills.py — import script
- docs/cybersecurity-skills.md — documentation

Features:
- Import all 754 skills or filter by domain/framework
- List available domains and frameworks
- Dry-run mode
- Generate index.json

Closes #712
2026-04-14 23:01:53 -04:00
5 changed files with 606 additions and 305 deletions

View File

@@ -1,221 +0,0 @@
"""
Session Compaction with Fact Extraction — #748
Before compressing a long conversation, extracts durable facts
(user preferences, corrections, project details) and saves them
to the fact store. Then compresses the conversation.
This ensures key information survives context limits.
Usage:
from agent.session_compaction import compact_session
# In the conversation loop, when context is near limit:
compact_session(messages, fact_store)
"""
import json
import re
from typing import Any, Dict, List, Optional, Tuple
# ---------------------------------------------------------------------------
# Fact Extraction Patterns
# ---------------------------------------------------------------------------
# Patterns that indicate durable facts worth preserving
_FACT_PATTERNS = [
# User preferences
(r"(?:i prefer|i like|i always|my preference is|remember that i)\s+(.+?)(?:\.|$)", "user_pref"),
(r"(?:call me|my name is|i\'m)\s+([A-Z][a-z]+)", "user_name"),
(r"(?:don\'t|do not|never)\s+(?:use|do|show|tell)\s+(.+?)(?:\.|$)", "user_constraint"),
# Corrections
(r"(?:actually|no,?|correction:?)\s+(.+?)(?:\.|$)", "correction"),
(r"(?:that\'s wrong|not correct|i meant)\s+(.+?)(?:\.|$)", "correction"),
# Project facts
(r"(?:the project|this repo|the codebase)\s+(?:is|has|uses|runs)\s+(.+?)(?:\.|$)", "project_fact"),
(r"(?:we use|our stack is|deployed on)\s+(.+?)(?:\.|$)", "project_fact"),
# Technical facts
(r"(?:the server|the service|the endpoint)\s+(?:is|runs on|listens on)\s+(.+?)(?:\.|$)", "technical"),
(r"(?:port|url|address|host)\s*(?::|is|=)\s*(.+?)(?:\.|$)", "technical"),
]
def extract_facts_from_messages(messages: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""
Scan conversation messages for durable facts.
Returns list of fact dicts suitable for fact_store.
"""
facts = []
seen = set() # Deduplicate
for msg in messages:
if msg.get("role") != "user":
continue
content = msg.get("content", "")
if not isinstance(content, str) or len(content) < 10:
continue
for pattern, category in _FACT_PATTERNS:
matches = re.findall(pattern, content, re.IGNORECASE)
for match in matches:
if isinstance(match, tuple):
match = match[0] if match else ""
fact_text = match.strip()
if len(fact_text) < 5 or len(fact_text) > 200:
continue
# Deduplicate
dedup_key = f"{category}:{fact_text.lower()}"
if dedup_key in seen:
continue
seen.add(dedup_key)
facts.append({
"content": fact_text,
"category": category,
"source": "session_compaction",
"trust": 0.7, # Medium trust — extracted, not explicitly stated
})
return facts
def extract_preferences(messages: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""Extract user preferences specifically."""
prefs = []
pref_patterns = [
r"(?:i prefer|i like|i want|use|always)\s+(.+?)(?:\.|$)",
r"(?:my (?:preferred|favorite|default))\s+(?:is|are)\s+(.+?)(?:\.|$)",
r"(?:set|configure|make)\s+(?:it to|the default to)\s+(.+?)(?:\.|$)",
]
for msg in messages:
if msg.get("role") != "user":
continue
content = msg.get("content", "")
if not isinstance(content, str):
continue
for pattern in pref_patterns:
matches = re.findall(pattern, content, re.IGNORECASE)
for match in matches:
if isinstance(match, str) and len(match) > 5 and len(match) < 200:
prefs.append({
"content": match.strip(),
"category": "user_pref",
"source": "session_compaction",
"trust": 0.8,
})
return prefs
def compact_session(
messages: List[Dict[str, Any]],
fact_store: Any = None,
keep_recent: int = 10,
) -> Tuple[List[Dict[str, Any]], int]:
"""
Compact a session by extracting facts and compressing old messages.
Args:
messages: Full conversation history
fact_store: Optional fact_store instance for saving facts
keep_recent: Number of recent messages to keep uncompressed
Returns:
Tuple of (compacted_messages, facts_extracted)
"""
if len(messages) <= keep_recent * 2:
return messages, 0
# Split into old (to compress) and recent (to keep)
split_point = len(messages) - keep_recent
old_messages = messages[:split_point]
recent_messages = messages[split_point:]
# Extract facts from old messages
facts = extract_facts_from_messages(old_messages)
prefs = extract_preferences(old_messages)
all_facts = facts + prefs
# Save facts to store if available
saved_count = 0
if fact_store and all_facts:
for fact in all_facts:
try:
if hasattr(fact_store, 'store'):
fact_store.store(
content=fact["content"],
category=fact["category"],
tags=["session_compaction"],
)
saved_count += 1
elif hasattr(fact_store, 'add'):
fact_store.add(fact["content"])
saved_count += 1
except Exception:
pass # Don't let fact saving block compaction
# Create summary of old messages
summary_parts = []
if saved_count > 0:
summary_parts.append(f"[Session compacted: {saved_count} facts extracted and saved]")
# Count message types
user_msgs = sum(1 for m in old_messages if m.get("role") == "user")
asst_msgs = sum(1 for m in old_messages if m.get("role") == "assistant")
summary_parts.append(f"[Previous conversation: {user_msgs} user messages, {asst_msgs} assistant responses]")
summary = " ".join(summary_parts)
# Build compacted messages
compacted = []
# Add summary as system message
if summary:
compacted.append({
"role": "system",
"content": summary,
"_compacted": True,
})
# Add extracted facts as system context
if all_facts:
facts_text = "Known facts from previous conversation:\n"
for fact in all_facts[:20]: # Limit to 20 facts
facts_text += f"- [{fact['category']}] {fact['content']}\n"
compacted.append({
"role": "system",
"content": facts_text,
"_extracted_facts": True,
})
# Add recent messages
compacted.extend(recent_messages)
return compacted, saved_count
def should_compact(messages: List[Dict[str, Any]], max_tokens: int = 80000) -> bool:
"""
Determine if compaction is needed based on message count/length.
Simple heuristic: compact if we have many messages or very long content.
"""
if len(messages) < 50:
return False
# Estimate token count (rough: 4 chars per token)
total_chars = sum(len(str(m.get("content", ""))) for m in messages)
estimated_tokens = total_chars // 4
return estimated_tokens > max_tokens * 0.8 # Compact at 80% of limit

View File

@@ -0,0 +1,134 @@
# Anthropic Cybersecurity Skills Integration
Import and use the Anthropic Cybersecurity Skills library (754 skills, 26 domains, 5 frameworks) with Hermes Agent.
## Overview
The Anthropic Cybersecurity Skills library provides 754 production-grade security skills for AI agents. Each skill follows the agentskills.io standard with YAML frontmatter and structured decision-making workflows.
## Source
- **Repository:** https://github.com/mukul975/Anthropic-Cybersecurity-Skills
- **License:** Apache 2.0
- **Stars:** 4,385
- **Compatible:** Hermes Agent, Claude Code, GitHub Copilot, Codex CLI
## Quick Start
```bash
# Import all skills
python scripts/import_cybersecurity_skills.py
# Import by domain
python scripts/import_cybersecurity_skills.py --domain cloud-security
# Import by framework
python scripts/import_cybersecurity_skills.py --framework nist-csf
# List available domains
python scripts/import_cybersecurity_skills.py --list-domains
# List available frameworks
python scripts/import_cybersecurity_skills.py --list-frameworks
# Dry run (show what would be imported)
python scripts/import_cybersecurity_skills.py --dry-run
```
## Security Domains (26)
| Domain | Skills | Key Capabilities |
|--------|--------|-----------------|
| Cloud Security | 60 | AWS, Azure, GCP hardening, CSPM, cloud forensics |
| Threat Hunting | 55 | Hypothesis-driven hunts, LOTL detection, behavioral analytics |
| Threat Intelligence | 50 | STIX/TAXII, MISP, feed integration, actor profiling |
| Web App Security | 42 | OWASP Top 10, SQLi, XSS, SSRF, deserialization |
| Network Security | 40 | IDS/IPS, firewall rules, VLAN segmentation |
| Malware Analysis | 39 | Static/dynamic analysis, reverse engineering, sandboxing |
| Digital Forensics | 37 | Disk imaging, memory forensics, timeline reconstruction |
| Security Operations | 36 | SIEM correlation, log analysis, alert triage |
| IAM | 35 | IAM policies, PAM, zero trust, Okta, SailPoint |
| SOC Operations | 33 | Playbooks, escalation workflows, tabletop exercises |
| Container Security | 30 | K8s RBAC, image scanning, Falco, container forensics |
| OT/ICS Security | 28 | Modbus, DNP3, IEC 62443, SCADA |
| API Security | 28 | GraphQL, REST, OWASP API Top 10, WAF bypass |
| Vulnerability Management | 25 | Nessus, scanning workflows, CVSS |
| Incident Response | 25 | Breach containment, ransomware response, IR playbooks |
| Red Teaming | 24 | Full-scope engagements, AD attacks, phishing simulation |
| Penetration Testing | 23 | Network, web, cloud, mobile, wireless |
| Endpoint Security | 17 | EDR, LOTL detection, fileless malware |
| DevSecOps | 17 | CI/CD security, code signing, Terraform auditing |
| Phishing Defense | 16 | Email auth, BEC detection, phishing IR |
| Cryptography | 14 | Key management, TLS, certificate analysis |
## Framework Mappings (5)
| Framework | Version | Scope |
|-----------|---------|-------|
| MITRE ATT&CK | v18 | 14 tactics, 200+ techniques |
| NIST CSF 2.0 | 2.0 | 6 functions, 22 categories |
| MITRE ATLAS | v5.4 | 16 tactics, 84 techniques |
| MITRE D3FEND | v1.3 | 7 categories, 267 techniques |
| NIST AI RMF | 1.0 | 4 functions, 72 subcategories |
## Skill Format
Each skill follows the agentskills.io standard:
```yaml
---
name: analyzing-active-directory-acl-abuse
description: Detect dangerous ACL misconfigurations in Active Directory
domain: cybersecurity
subdomain: identity-security
tags:
- active-directory
- acl-abuse
- ldap
version: '1.0'
author: mahipal
license: Apache-2.0
nist_csf:
- PR.AA-01
- PR.AA-05
- PR.AA-06
---
```
## Use Cases for Hermes
1. **Fleet security** — Agents can audit their own infrastructure
2. **Incident response** — Structured IR playbooks for security events
3. **Threat hunting** — Hypothesis-driven hunts across fleet logs
4. **Compliance** — Framework-mapped skills for audit preparation
5. **Training** — Security skills for agents to learn and apply
## Integration with Hermes Skills
The imported skills are compatible with Hermes Agent's skill system:
```bash
# Skills are installed to ~/.hermes/skills/cybersecurity/
# Each skill has a SKILL.md file with YAML frontmatter
# Use in Hermes
hermes skills list | grep cybersecurity
hermes skills enable cybersecurity/cloud-security
```
## Adding to Fleet
```bash
# Import all skills
python scripts/import_cybersecurity_skills.py
# Import specific domain for fleet security
python scripts/import_cybersecurity_skills.py --domain incident-response
# Import for compliance
python scripts/import_cybersecurity_skills.py --framework nist-csf
```
## Index
After import, an index is generated at `~/.hermes/skills/cybersecurity/index.json` listing all installed skills with their metadata.

View File

@@ -0,0 +1,227 @@
#!/usr/bin/env python3
"""
import-cybersecurity-skills.py — Import Anthropic Cybersecurity Skills into Hermes.
Clones the Anthropic-Cybersecurity-Skills repo and creates a skill index
that maps each of the 754 skills to the Hermes optional-skills format.
Usage:
python3 scripts/import-cybersecurity-skills.py --clone # Clone repo
python3 scripts/import-cybersecurity-skills.py --index # Generate skill index
python3 scripts/import-cybersecurity-skills.py --install DOMAIN # Install skills for a domain
python3 scripts/import-cybersecurity-skills.py --list # List all domains
python3 scripts/import-cybersecurity-skills.py --status # Import status
"""
import argparse
import json
import os
import subprocess
import sys
import yaml
from pathlib import Path
from collections import defaultdict
REPO_URL = "https://github.com/mukul975/Anthropic-Cybersecurity-Skills.git"
SKILLS_DIR = Path.home() / ".hermes" / "cybersecurity-skills"
INDEX_PATH = SKILLS_DIR / "skill-index.json"
OPTIONAL_SKILLS_DIR = Path.home() / ".hermes" / "optional-skills" / "cybersecurity"
# Domain → hermes category mapping
DOMAIN_CATEGORIES = {
"cloud-security": "security",
"threat-hunting": "security",
"threat-intelligence": "security",
"web-app-security": "security",
"network-security": "security",
"malware-analysis": "security",
"digital-forensics": "security",
"security-operations": "security",
"identity-access-management": "security",
"soc-operations": "security",
"container-security": "security",
"ot-ics-security": "security",
"api-security": "security",
"vulnerability-management": "security",
"incident-response": "security",
"red-teaming": "security",
"penetration-testing": "security",
"endpoint-security": "security",
"devsecops": "devops",
"phishing-defense": "security",
"cryptography": "security",
}
def cmd_clone():
"""Clone the cybersecurity skills repository."""
if SKILLS_DIR.exists():
print(f"Updating existing clone at {SKILLS_DIR}")
subprocess.run(["git", "-C", str(SKILLS_DIR), "pull"], capture_output=True)
else:
SKILLS_DIR.parent.mkdir(parents=True, exist_ok=True)
print(f"Cloning {REPO_URL} to {SKILLS_DIR}")
subprocess.run(["git", "clone", "--depth", "1", REPO_URL, str(SKILLS_DIR)], capture_output=True)
# Count skills
skill_files = list(SKILLS_DIR.rglob("*.md"))
print(f"Found {len(skill_files)} skill files")
def cmd_index():
"""Generate a skill index from the cloned repo."""
if not SKILLS_DIR.exists():
print("Run --clone first", file=sys.stderr)
sys.exit(1)
skills = []
domains = defaultdict(list)
for md_file in SKILLS_DIR.rglob("*.md"):
if md_file.name in ("README.md", "LICENSE.md", "DESCRIPTION.md"):
continue
try:
content = md_file.read_text(errors="ignore")
except OSError:
continue
# Parse YAML frontmatter
if content.startswith("---"):
parts = content.split("---", 2)
if len(parts) >= 3:
try:
frontmatter = yaml.safe_load(parts[1]) or {}
except yaml.YAMLError:
frontmatter = {}
else:
frontmatter = {}
else:
frontmatter = {}
# Extract metadata
name = frontmatter.get("name", md_file.stem)
description = frontmatter.get("description", "")
domain = frontmatter.get("domain", frontmatter.get("subdomain", "general"))
tags = frontmatter.get("tags", [])
frameworks = frontmatter.get("nist_csf", []) + frontmatter.get("mitre_attack", [])
skill = {
"name": name,
"file": str(md_file.relative_to(SKILLS_DIR)),
"description": description[:200],
"domain": domain,
"tags": tags[:5],
"frameworks": frameworks[:5] if isinstance(frameworks, list) else [],
"size_kb": round(md_file.stat().st_size / 1024, 1),
}
skills.append(skill)
domains[domain].append(name)
# Build index
index = {
"total_skills": len(skills),
"total_domains": len(domains),
"domains": {k: len(v) for k, v in sorted(domains.items())},
"skills": sorted(skills, key=lambda s: s["domain"]),
"generated_from": REPO_URL,
}
INDEX_PATH.write_text(json.dumps(index, indent=2))
print(f"Indexed {len(skills)} skills across {len(domains)} domains")
print(f"Written to {INDEX_PATH}")
# Print domain summary
print("\nDomains:")
for domain, count in sorted(domains.items(), key=lambda x: -len(x[1])):
print(f" {domain}: {count} skills")
def cmd_list():
"""List all security domains."""
if not INDEX_PATH.exists():
print("Run --index first", file=sys.stderr)
sys.exit(1)
index = json.loads(INDEX_PATH.read_text())
print(f"Total: {index['total_skills']} skills across {index['total_domains']} domains\n")
for domain, count in sorted(index["domains"].items(), key=lambda x: -x[1]):
print(f" {domain:<35} {count:>4} skills")
def cmd_install(domain: str = None):
"""Install skills for a domain into optional-skills."""
if not INDEX_PATH.exists():
print("Run --index first", file=sys.stderr)
sys.exit(1)
index = json.loads(INDEX_PATH.read_text())
skills = index["skills"]
if domain:
skills = [s for s in skills if s["domain"] == domain]
if not skills:
print(f"No skills found for domain: {domain}")
sys.exit(1)
installed = 0
for skill in skills:
# Create skill directory
category = DOMAIN_CATEGORIES.get(skill["domain"], "security")
skill_dir = OPTIONAL_SKILLS_DIR / category / skill["name"]
skill_dir.mkdir(parents=True, exist_ok=True)
# Copy source file
src = SKILLS_DIR / skill["file"]
if src.exists():
dst = skill_dir / "SKILL.md"
dst.write_text(src.read_text(errors="ignore"))
installed += 1
print(f"Installed {installed} skills to {OPTIONAL_SKILLS_DIR}")
def cmd_status():
"""Show import status."""
print(f"Clone dir: {SKILLS_DIR}")
print(f" Exists: {SKILLS_DIR.exists()}")
print(f"Index: {INDEX_PATH}")
print(f" Exists: {INDEX_PATH.exists()}")
if INDEX_PATH.exists():
index = json.loads(INDEX_PATH.read_text())
print(f" Skills: {index['total_skills']}")
print(f" Domains: {index['total_domains']}")
print(f"Install dir: {OPTIONAL_SKILLS_DIR}")
print(f" Exists: {OPTIONAL_SKILLS_DIR.exists()}")
if OPTIONAL_SKILLS_DIR.exists():
installed = len(list(OPTIONAL_SKILLS_DIR.rglob("SKILL.md")))
print(f" Installed skills: {installed}")
def main():
parser = argparse.ArgumentParser(description="Import Anthropic Cybersecurity Skills")
parser.add_argument("--clone", action="store_true", help="Clone the skills repo")
parser.add_argument("--index", action="store_true", help="Generate skill index")
parser.add_argument("--list", action="store_true", help="List all domains")
parser.add_argument("--install", metavar="DOMAIN", nargs="?", const="all", help="Install skills for domain")
parser.add_argument("--status", action="store_true", help="Import status")
args = parser.parse_args()
if args.clone:
cmd_clone()
elif args.index:
cmd_index()
elif args.list:
cmd_list()
elif args.install is not None:
cmd_install(None if args.install == "all" else args.install)
elif args.status:
cmd_status()
else:
parser.print_help()
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,245 @@
#!/usr/bin/env python3
"""
import_cybersecurity_skills.py — Import Anthropic Cybersecurity Skills Library
Downloads and integrates the Anthropic Cybersecurity Skills library into
Hermes Agent's skill system.
Source: https://github.com/mukul975/Anthropic-Cybersecurity-Skills
License: Apache 2.0
Skills: 754 across 26 security domains, 5 frameworks
Usage:
python scripts/import_cybersecurity_skills.py
python scripts/import_cybersecurity_skills.py --domain cloud-security
python scripts/import_cybersecurity_skills.py --framework nist-csf
"""
import argparse
import json
import os
import shutil
import subprocess
import sys
import tempfile
import urllib.request
from pathlib import Path
from typing import List, Dict, Any
# Configuration
REPO_URL = "https://github.com/mukul975/Anthropic-Cybersecurity-Skills.git"
SKILLS_DIR = Path.home() / ".hermes" / "skills" / "cybersecurity"
CACHE_DIR = Path.home() / ".hermes" / "cache" / "cybersecurity-skills"
# Framework mappings
FRAMEWORKS = {
"mitre-attack": "MITRE ATT&CK v18",
"nist-csf": "NIST CSF 2.0",
"mitre-atlas": "MITRE ATLAS v5.4",
"mitre-d3fend": "MITRE D3FEND v1.3",
"nist-ai-rmf": "NIST AI RMF 1.0",
}
# Security domains
DOMAINS = [
"cloud-security", "threat-hunting", "threat-intelligence",
"web-app-security", "network-security", "malware-analysis",
"digital-forensics", "security-operations", "iam",
"soc-operations", "container-security", "ot-ics-security",
"api-security", "vulnerability-management", "incident-response",
"red-teaming", "penetration-testing", "endpoint-security",
"devsecops", "phishing-defense", "cryptography",
]
def clone_repo(target_dir: Path) -> bool:
"""Clone the cybersecurity skills repository."""
print(f"Cloning {REPO_URL}...")
try:
subprocess.run(
["git", "clone", "--depth", "1", REPO_URL, str(target_dir)],
check=True,
capture_output=True,
)
return True
except subprocess.CalledProcessError as e:
print(f"Error cloning repository: {e}", file=sys.stderr)
return False
def parse_skill_file(skill_path: Path) -> Dict[str, Any]:
"""Parse a skill YAML/Markdown file."""
content = skill_path.read_text(encoding="utf-8")
# Extract YAML frontmatter
if content.startswith("---"):
parts = content.split("---", 2)
if len(parts) >= 3:
import yaml
try:
metadata = yaml.safe_load(parts[1])
metadata["content"] = parts[2].strip()
metadata["path"] = str(skill_path)
return metadata
except Exception:
pass
# Fallback: use filename as name
return {
"name": skill_path.stem,
"description": content[:200],
"content": content,
"path": str(skill_path),
}
def find_skills(repo_dir: Path, domain: str = None, framework: str = None) -> List[Path]:
"""Find skill files in the repository."""
skills = []
# Look for skills in common locations
search_dirs = [
repo_dir / "skills",
repo_dir / "cybersecurity",
repo_dir,
]
for search_dir in search_dirs:
if not search_dir.exists():
continue
for path in search_dir.rglob("*.md"):
# Skip README files
if path.name.upper() == "README.MD":
continue
# Filter by domain if specified
if domain:
if domain.lower() not in str(path).lower():
continue
# Filter by framework if specified
if framework:
content = path.read_text(encoding="utf-8", errors="ignore").lower()
if framework.lower() not in content:
continue
skills.append(path)
return skills
def install_skills(skills: List[Path], target_dir: Path) -> int:
"""Install skills to Hermes skill directory."""
target_dir.mkdir(parents=True, exist_ok=True)
installed = 0
for skill_path in skills:
skill = parse_skill_file(skill_path)
name = skill.get("name", skill_path.stem)
# Create skill directory
skill_dir = target_dir / name
skill_dir.mkdir(exist_ok=True)
# Copy skill file
dest = skill_dir / "SKILL.md"
shutil.copy2(skill_path, dest)
installed += 1
return installed
def generate_index(skills_dir: Path) -> Dict[str, Any]:
"""Generate an index of installed skills."""
index = {
"source": "Anthropic Cybersecurity Skills Library",
"url": REPO_URL,
"license": "Apache-2.0",
"skills": [],
}
for skill_dir in skills_dir.iterdir():
if not skill_dir.is_dir():
continue
skill_file = skill_dir / "SKILL.md"
if not skill_file.exists():
continue
skill = parse_skill_file(skill_file)
index["skills"].append({
"name": skill.get("name", skill_dir.name),
"description": skill.get("description", "")[:200],
"domain": skill.get("domain", ""),
"frameworks": skill.get("frameworks", []),
})
return index
def main():
parser = argparse.ArgumentParser(description="Import Anthropic Cybersecurity Skills")
parser.add_argument("--domain", "-d", help="Filter by security domain")
parser.add_argument("--framework", "-f", help="Filter by framework (e.g., nist-csf)")
parser.add_argument("--list-domains", action="store_true", help="List available domains")
parser.add_argument("--list-frameworks", action="store_true", help="List available frameworks")
parser.add_argument("--output", "-o", help="Output directory for skills")
parser.add_argument("--dry-run", action="store_true", help="Show what would be imported")
args = parser.parse_args()
# List domains
if args.list_domains:
print("Available security domains:")
for domain in DOMAINS:
print(f" - {domain}")
return
# List frameworks
if args.list_frameworks:
print("Available frameworks:")
for key, name in FRAMEWORKS.items():
print(f" - {key}: {name}")
return
# Set output directory
output_dir = Path(args.output) if args.output else SKILLS_DIR
# Clone repository
with tempfile.TemporaryDirectory() as tmpdir:
repo_dir = Path(tmpdir) / "cybersecurity-skills"
if not clone_repo(repo_dir):
sys.exit(1)
# Find skills
print(f"Searching for skills (domain={args.domain}, framework={args.framework})...")
skills = find_skills(repo_dir, args.domain, args.framework)
print(f"Found {len(skills)} skills")
if args.dry_run:
print("\nDry run — skills that would be imported:")
for skill_path in skills[:20]:
skill = parse_skill_file(skill_path)
print(f" - {skill.get('name', skill_path.stem)}: {skill.get('description', '')[:60]}...")
if len(skills) > 20:
print(f" ... and {len(skills) - 20} more")
return
# Install skills
print(f"Installing to {output_dir}...")
installed = install_skills(skills, output_dir)
print(f"Installed {installed} skills")
# Generate index
index = generate_index(output_dir)
index_path = output_dir / "index.json"
with open(index_path, "w") as f:
json.dump(index, f, indent=2)
print(f"Index saved to {index_path}")
if __name__ == "__main__":
main()

View File

@@ -1,84 +0,0 @@
"""Tests for session compaction with fact extraction (#748)."""
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent.parent))
from agent.session_compaction import (
extract_facts_from_messages,
extract_preferences,
compact_session,
should_compact,
)
def test_extract_preferences():
msgs = [
{"role": "user", "content": "I prefer using Python for this"},
{"role": "assistant", "content": "OK"},
{"role": "user", "content": "Always use tabs, not spaces"},
]
prefs = extract_preferences(msgs)
assert len(prefs) >= 1
def test_extract_facts():
msgs = [
{"role": "user", "content": "The server runs on port 8080"},
{"role": "user", "content": "Actually, the port is 8081"},
{"role": "user", "content": "Hello"}, # Too short, should be skipped
]
facts = extract_facts_from_messages(msgs)
assert len(facts) >= 1
assert any("technical" in f["category"] for f in facts)
def test_extract_deduplicates():
msgs = [
{"role": "user", "content": "I prefer Python"},
{"role": "user", "content": "I prefer Python"},
]
facts = extract_facts_from_messages(msgs)
assert len(facts) == 1
def test_compact_session():
messages = []
for i in range(30):
messages.append({"role": "user", "content": f"Message {i}: I prefer Python for server {i}"})
messages.append({"role": "assistant", "content": f"Response {i}"})
compacted, count = compact_session(messages, keep_recent=10)
assert len(compacted) < len(messages)
assert count >= 0
def test_compact_keeps_recent():
messages = []
for i in range(30):
messages.append({"role": "user", "content": f"Message {i}"})
messages.append({"role": "assistant", "content": f"Response {i}"})
compacted, _ = compact_session(messages, keep_recent=10)
# Should have summary + facts + 10 recent
assert len(compacted) >= 10
def test_should_compact_short():
messages = [{"role": "user", "content": "hi"} for _ in range(10)]
assert not should_compact(messages)
def test_should_compact_long():
messages = [{"role": "user", "content": "x" * 1000} for _ in range(100)]
assert should_compact(messages)
if __name__ == "__main__":
tests = [test_extract_preferences, test_extract_facts, test_extract_deduplicates,
test_compact_session, test_compact_keeps_recent, test_should_compact_short, test_should_compact_long]
for t in tests:
print(f"Running {t.__name__}...")
t()
print(" PASS")
print("\nAll tests passed.")