1#!/usr/bin/env python3
2"""Deterministically smoke-test skill activation boundaries."""
3 
4from __future__ import annotations
5 
6import argparse
7import json
8import re
9import sys
10from pathlib import Path
11from typing import Any
12 
13 
14ROOT = Path(__file__).resolve().parents[2]
15STOPWORDS = {
16    "the",
17    "and",
18    "for",
19    "with",
20    "that",
21    "this",
22    "into",
23    "from",
24    "agent",
25    "agents",
26    "skill",
27    "skills",
28    "build",
29    "create",
30    "design",
31    "whether",
32    "output",
33    "outputs",
34    "before",
35    "later",
36    "can",
37    "should",
38}
39 
40 
41def tokens(text: str) -> set[str]:
42    raw = re.findall(r"[a-zA-Z][a-zA-Z0-9_-]{2,}", text.lower())
43    return {token.replace("_", "-") for token in raw if token not in STOPWORDS}
44 
45 
46def parse_frontmatter(text: str) -> dict[str, str]:
47    if not text.startswith("---\n"):
48        return {}
49    end = text.find("\n---", 4)
50    if end == -1:
51        return {}
52    result: dict[str, str] = {}
53    for line in text[4:end].splitlines():
54        if ":" not in line or line.startswith(" "):
55            continue
56        key, value = line.split(":", 1)
57        result[key.strip()] = value.strip().strip('"')
58    return result
59 
60 
61def activation_text(text: str) -> str:
62    frontmatter = parse_frontmatter(text)
63    parts = [frontmatter.get("description", "")]
64    match = re.search(r"## When to Activate\n(.+?)(?:\n## |\Z)", text, flags=re.DOTALL)
65    if match:
66        parts.append(match.group(1))
67    return "\n".join(parts)
68 
69 
70def load_skills(root: Path) -> dict[str, set[str]]:
71    skills: dict[str, set[str]] = {}
72    for path in sorted((root / "skills").glob("*/SKILL.md")):
73        skills[path.parent.name] = tokens(activation_text(path.read_text(encoding="utf-8")))
74    return skills
75 
76 
77def load_cases(path: Path) -> list[dict[str, Any]]:
78    cases: list[dict[str, Any]] = []
79    for line_number, line in enumerate(path.read_text(encoding="utf-8").splitlines(), start=1):
80        if not line.strip():
81            continue
82        try:
83            cases.append(json.loads(line))
84        except json.JSONDecodeError as exc:
85            raise ValueError(f"{path}:{line_number} invalid JSONL: {exc}") from exc
86    return cases
87 
88 
89def rank_skills(prompt: str, skills: dict[str, set[str]]) -> list[dict[str, Any]]:
90    prompt_tokens = tokens(prompt)
91    ranked: list[dict[str, Any]] = []
92    for name, skill_tokens in skills.items():
93        shared = prompt_tokens & skill_tokens
94        score = len(shared) / max(len(prompt_tokens), 1)
95        ranked.append({"skill": name, "score": round(score, 4), "shared_terms": sorted(shared)})
96    ranked.sort(key=lambda item: (item["score"], item["skill"]), reverse=True)
97    return ranked
98 
99 
100def evaluate(root: Path, cases_path: Path) -> dict[str, Any]:
101    skills = load_skills(root)
102    results: list[dict[str, Any]] = []
103    for case in load_cases(cases_path):
104        ranked = rank_skills(case["prompt"], skills)
105        top_three = [item["skill"] for item in ranked[:3]]
106        rejected_in_top_three = sorted(set(case.get("rejected_skills", [])) & set(top_three))
107        expected = case["expected_primary_skill"]
108        passed = expected in top_three and not rejected_in_top_three
109        results.append(
110            {
111                "case_id": case["case_id"],
112                "passed": passed,
113                "expected_primary_skill": expected,
114                "top_three": top_three,
115                "rejected_in_top_three": rejected_in_top_three,
116                "ranked": ranked[:5],
117            }
118        )
119    failures = [result for result in results if not result["passed"]]
120    return {
121        "ok": not failures,
122        "summary": {"cases": len(results), "failures": len(failures)},
123        "results": results,
124    }
125 
126 
127def main() -> int:
128    parser = argparse.ArgumentParser(description="Check deterministic skill activation cases")
129    parser.add_argument("--root", type=Path, default=ROOT)
130    parser.add_argument(
131        "--cases",
132        type=Path,
133        default=ROOT / "researcher" / "fixtures" / "activation-cases.jsonl",
134    )
135    parser.add_argument("--json", action="store_true")
136    args = parser.parse_args()
137 
138    result = evaluate(args.root, args.cases)
139    if args.json:
140        print(json.dumps(result, indent=2))
141    else:
142        print(
143            f"Activation cases {'passed' if result['ok'] else 'failed'}: "
144            f"{result['summary']['cases']} cases, {result['summary']['failures']} failures"
145        )
146        for item in result["results"]:
147            if not item["passed"]:
148                print(f"- {item['case_id']}: expected {item['expected_primary_skill']} top={item['top_three']}")
149    return 0 if result["ok"] else 1
150 
151 
152if __name__ == "__main__":
153    sys.exit(main())
154