1#!/usr/bin/env python3
2"""Check whether a proposed skill idea overlaps existing corpus content."""
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    "from",
23    "into",
24    "when",
25    "agent",
26    "agents",
27    "skill",
28    "skills",
29    "context",
30    "activation",
31    "artifact",
32    "artifacts",
33    "author",
34    "candidate",
35    "change",
36    "changes",
37    "check",
38    "checks",
39    "claim",
40    "claims",
41    "decision",
42    "delta",
43    "evidence",
44    "existing",
45    "file",
46    "gaps",
47    "human",
48    "proposal",
49    "proposed",
50    "quality",
51    "retrieval",
52    "review",
53    "risk",
54    "risks",
55    "source",
56    "status",
57    "target",
58    "type",
59}
60 
61 
62STRUCTURED_SECTIONS = [
63    "Mechanism",
64    "Skill Target",
65    "Proposed Delta",
66    "Risks And Gaps",
67    "Recommendation",
68]
69 
70 
71def tokens(text: str) -> set[str]:
72    raw = re.findall(r"[a-zA-Z][a-zA-Z0-9_-]{2,}", text.lower())
73    return {t.replace("_", "-") for t in raw if t not in STOPWORDS}
74 
75 
76def jaccard(a: set[str], b: set[str]) -> float:
77    if not a or not b:
78        return 0.0
79    return len(a & b) / len(a | b)
80 
81 
82def read_input(args: argparse.Namespace) -> str:
83    parts: list[str] = []
84    if args.text:
85        parts.append(args.text)
86    if args.file:
87        parts.append(Path(args.file).read_text(encoding="utf-8"))
88    if not parts:
89        parts.append(sys.stdin.read())
90    return "\n".join(parts).strip()
91 
92 
93def extract_sections(text: str) -> dict[str, str]:
94    sections: dict[str, str] = {}
95    for match in re.finditer(r"^##\s+(.+?)\n(.+?)(?=^##\s+|\Z)", text, flags=re.DOTALL | re.MULTILINE):
96        sections[match.group(1).strip()] = match.group(2).strip()
97    return sections
98 
99 
100def structured_proposal_text(text: str) -> str:
101    sections = extract_sections(text)
102    values = [sections[name] for name in STRUCTURED_SECTIONS if sections.get(name)]
103    return "\n".join(values) if values else salient_text(text)
104 
105 
106def mechanism_text(entry: dict[str, Any]) -> str:
107    values: list[str] = []
108    for key in ["mechanism_id", "owning_skill", "activation_scenario", "behavior_change"]:
109        value = entry.get(key)
110        if isinstance(value, str):
111            values.append(value)
112    for key in ["failure_modes", "evidence"]:
113        value = entry.get(key)
114        if isinstance(value, list):
115            values.extend(str(item) for item in value)
116    return "\n".join(values)
117 
118 
119def load_mechanisms(root: Path) -> list[dict[str, Any]]:
120    path = root / "researcher" / "mechanisms" / "registry.jsonl"
121    if not path.exists():
122        return []
123    entries: list[dict[str, Any]] = []
124    for line_number, line in enumerate(path.read_text(encoding="utf-8").splitlines(), start=1):
125        if not line.strip():
126            continue
127        try:
128            entry = json.loads(line)
129        except json.JSONDecodeError:
130            entries.append(
131                {
132                    "mechanism_id": f"invalid-json-line-{line_number}",
133                    "status": "invalid",
134                    "activation_scenario": "",
135                    "behavior_change": line,
136                    "failure_modes": ["invalid registry entry"],
137                }
138            )
139            continue
140        if entry.get("status") in {"accepted", "candidate"}:
141            entries.append(entry)
142    return entries
143 
144 
145def salient_text(text: str) -> str:
146    """Keep the parts of a proposal most likely to contain real mechanism text."""
147    section_names = [
148        "Mechanism",
149        "Proposed Delta",
150        "Risks And Gaps",
151        "Recommendation",
152        "Core Concepts",
153        "Detailed Topics",
154        "Practical Guidance",
155        "Guidelines",
156        "Gotchas",
157    ]
158    chunks: list[str] = []
159    for section in section_names:
160        match = re.search(rf"^## {re.escape(section)}\n(.+?)(?:\n## |\Z)", text, flags=re.DOTALL | re.MULTILINE)
161        if match:
162            chunks.append(match.group(1))
163    reduced = "\n".join(chunks) if chunks else text
164    lines: list[str] = []
165    for line in reduced.splitlines():
166        stripped = line.strip()
167        if not stripped:
168            continue
169        if stripped.startswith("- ["):
170            continue
171        if stripped in {"| --- | --- | --- |", "| --- | --- |"}:
172            continue
173        if "[Short Title]" in stripped or '""' in stripped:
174            continue
175        lines.append(stripped)
176    return "\n".join(lines)
177 
178 
179def corpus_documents(root: Path, exclude: Path | None = None) -> list[dict[str, str]]:
180    docs: list[dict[str, str]] = []
181    for path in sorted((root / "skills").glob("*/SKILL.md")):
182        if exclude and path.resolve() == exclude.resolve():
183            continue
184        docs.append({"path": str(path.relative_to(root)), "text": salient_text(path.read_text(encoding="utf-8"))})
185    for path in sorted((root / "researcher" / "fixtures").glob("**/*.md")):
186        if exclude and path.resolve() == exclude.resolve():
187            continue
188        docs.append({"path": str(path.relative_to(root)), "text": salient_text(path.read_text(encoding="utf-8"))})
189    for path in sorted((root / "researcher" / "runs").glob("**/*.md")):
190        if exclude and path.resolve() == exclude.resolve():
191            continue
192        if path.name == "skill-proposal.md":
193            text = path.read_text(encoding="utf-8")
194            if "Retrieval status: partial" in text or "[Short Title]" in text:
195                continue
196        docs.append({"path": str(path.relative_to(root)), "text": salient_text(path.read_text(encoding="utf-8"))})
197    return docs
198 
199 
200def main() -> int:
201    parser = argparse.ArgumentParser(description="Check novelty of a skill proposal against corpus")
202    parser.add_argument("--root", type=Path, default=ROOT)
203    parser.add_argument("--text", help="proposal text to check")
204    parser.add_argument("--file", type=Path, help="proposal file to check")
205    parser.add_argument("--threshold", type=float, default=0.18)
206    parser.add_argument("--json", action="store_true")
207    args = parser.parse_args()
208 
209    proposal = read_input(args)
210    proposal_structured_text = structured_proposal_text(proposal)
211    proposal_tokens = tokens(proposal_structured_text)
212 
213    mechanism_overlaps: list[dict[str, Any]] = []
214    for entry in load_mechanisms(args.root):
215        entry_tokens = tokens(mechanism_text(entry))
216        score = jaccard(proposal_tokens, entry_tokens)
217        if score > 0:
218            mechanism_overlaps.append(
219                {
220                    "mechanism_id": entry.get("mechanism_id", ""),
221                    "owning_skill": entry.get("owning_skill", ""),
222                    "score": round(score, 4),
223                    "shared_terms": sorted(proposal_tokens & entry_tokens)[:30],
224                }
225            )
226    mechanism_overlaps.sort(key=lambda item: item["score"], reverse=True)
227 
228    overlaps: list[dict[str, Any]] = []
229    exclude = args.file.resolve() if args.file else None
230    for doc in corpus_documents(args.root, exclude=exclude):
231        doc_tokens = tokens(doc["text"])
232        score = jaccard(proposal_tokens, doc_tokens)
233        shared = sorted(proposal_tokens & doc_tokens)[:30]
234        if score > 0:
235            overlaps.append({"path": doc["path"], "score": round(score, 4), "shared_terms": shared})
236    overlaps.sort(key=lambda item: item["score"], reverse=True)
237 
238    max_mechanism_score = mechanism_overlaps[0]["score"] if mechanism_overlaps else 0.0
239    max_corpus_score = overlaps[0]["score"] if overlaps else 0.0
240    max_score = max(max_mechanism_score, max_corpus_score)
241    verdict = "pass"
242    if max_score >= args.threshold:
243        verdict = "human_review"
244    if max_score >= args.threshold * 1.75:
245        verdict = "likely_duplicate"
246 
247    result = {
248        "verdict": verdict,
249        "threshold": args.threshold,
250        "max_score": max_score,
251        "max_mechanism_score": max_mechanism_score,
252        "top_mechanism_overlaps": mechanism_overlaps[:10],
253        "max_corpus_score": max_corpus_score,
254        "top_overlaps": overlaps[:10],
255    }
256    if args.json:
257        print(json.dumps(result, indent=2))
258    else:
259        print(f"Novelty verdict: {verdict} (max overlap {max_score})")
260        if mechanism_overlaps:
261            print("Top mechanism overlaps:")
262            for item in mechanism_overlaps[:5]:
263                print(
264                    f"- {item['mechanism_id']} ({item['owning_skill']}): "
265                    f"{item['score']} shared={', '.join(item['shared_terms'][:8])}"
266                )
267        if overlaps:
268            print("Top corpus overlaps:")
269            for item in overlaps[:5]:
270                print(f"- {item['path']}: {item['score']} shared={', '.join(item['shared_terms'][:8])}")
271 
272    return 0 if verdict == "pass" else 2
273 
274 
275if __name__ == "__main__":
276    raise SystemExit(main())
277