1#!/usr/bin/env python3
2"""Validate the context-engineering skill corpus and researcher harness.
3 
4This script is intentionally deterministic. It checks structure, manifests, and
5machine-readable artifacts without calling an LLM.
6"""
7 
8from __future__ import annotations
9 
10import argparse
11import json
12import re
13import sys
14from dataclasses import dataclass, asdict
15from pathlib import Path
16from typing import Any
17 
18 
19REQUIRED_RESEARCHER_FILES = [
20    "README.md",
21    "source-registry.md",
22    "mechanisms/README.md",
23    "mechanisms/registry.jsonl",
24    "mechanisms/ledgers/accepted.jsonl",
25    "mechanisms/ledgers/rejected.jsonl",
26    "claims/README.md",
27    "claims/index.jsonl",
28    "corpus/index.json",
29    "benchmarks/README.md",
30    "benchmarks/scenarios/adversarial.jsonl",
31    "benchmarks/goldens/adversarial-goldens.json",
32    "rubrics/content-curation.md",
33    "rubrics/skill-change.md",
34    "rubrics/harness-change.md",
35    "rubrics/pairwise-skill-revision.md",
36    "templates/source-evaluation.json",
37    "templates/skill-proposal.md",
38    "templates/mechanism-proposal.jsonl",
39    "templates/research-thread.md",
40    "runbooks/autonomous-research-loop.md",
41    "runbooks/pr-readiness.md",
42    "scripts/validate_repo.py",
43    "scripts/validate_run.py",
44    "scripts/research_loop.py",
45    "scripts/novelty_check.py",
46    "scripts/compare_skill_revisions.py",
47    "scripts/check_activation_cases.py",
48    "scripts/run_benchmarks.py",
49    "fixtures/activation-cases.jsonl",
50    "reports/benchmark-history.jsonl",
51]
52 
53 
54REQUIRED_SOURCE_EVAL_KEYS = {
55    "evaluation_id",
56    "timestamp",
57    "source",
58    "gatekeeper",
59    "scoring",
60    "decision",
61    "extraction",
62}
63 
64 
65@dataclass
66class Finding:
67    severity: str
68    path: str
69    message: str
70 
71 
72class Validator:
73    def __init__(self, root: Path) -> None:
74        self.root = root.resolve()
75        self.findings: list[Finding] = []
76 
77    def error(self, path: Path | str, message: str) -> None:
78        self.findings.append(Finding("error", self.rel(path), message))
79 
80    def warn(self, path: Path | str, message: str) -> None:
81        self.findings.append(Finding("warning", self.rel(path), message))
82 
83    def rel(self, path: Path | str) -> str:
84        p = Path(path)
85        if not p.is_absolute():
86            return str(p)
87        try:
88            return str(p.relative_to(self.root))
89        except ValueError:
90            return str(p)
91 
92    def run(self) -> dict[str, Any]:
93        skill_names = self.validate_skills()
94        self.validate_manifests(skill_names)
95        self.validate_docs(skill_names)
96        self.validate_researcher()
97        self.validate_rubrics()
98        self.validate_mechanisms(skill_names)
99        self.validate_claims(skill_names)
100        self.validate_corpus_index(skill_names)
101        self.validate_activation_cases(skill_names)
102        self.validate_benchmark_scenarios()
103        self.validate_runs()
104        self.validate_root_provenance()
105        self.validate_source_evaluations()
106        errors = sum(1 for f in self.findings if f.severity == "error")
107        warnings = sum(1 for f in self.findings if f.severity == "warning")
108        return {
109            "ok": errors == 0,
110            "summary": {
111                "errors": errors,
112                "warnings": warnings,
113                "skill_count": len(skill_names),
114            },
115            "findings": [asdict(f) for f in self.findings],
116        }
117 
118    def validate_skills(self) -> list[str]:
119        skills_dir = self.root / "skills"
120        if not skills_dir.exists():
121            self.error(skills_dir, "skills directory missing")
122            return []
123 
124        skill_names: list[str] = []
125        for skill_dir in sorted(p for p in skills_dir.iterdir() if p.is_dir()):
126            skill_file = skill_dir / "SKILL.md"
127            if not skill_file.exists():
128                self.error(skill_file, "missing SKILL.md")
129                continue
130 
131            text = skill_file.read_text(encoding="utf-8")
132            lines = text.splitlines()
133            frontmatter = self.parse_frontmatter(text, skill_file)
134            name = str(frontmatter.get("name", "")).strip()
135            description = str(frontmatter.get("description", "")).strip()
136            skill_names.append(skill_dir.name)
137 
138            if not name:
139                self.error(skill_file, "frontmatter missing name")
140            elif name != skill_dir.name:
141                self.error(skill_file, f"name '{name}' does not match directory '{skill_dir.name}'")
142            if name and not re.fullmatch(r"[a-z0-9]+(?:-[a-z0-9]+)*", name):
143                self.error(skill_file, "name must be lowercase kebab-case without repeated hyphens")
144            if not description:
145                self.error(skill_file, "frontmatter missing description")
146            elif len(description) > 1024:
147                self.error(skill_file, "description exceeds 1024 characters")
148            if re.search(r"\b(I can|Use me|You can use this)\b", description):
149                self.warn(skill_file, "description may not be third person")
150            if len(lines) > 500:
151                self.error(skill_file, f"SKILL.md exceeds 500 lines ({len(lines)})")
152            for section in [
153                "## When to Activate",
154                "## Core Concepts",
155                "## Practical Guidance",
156                "## Examples",
157                "## Guidelines",
158                "## Gotchas",
159                "## Integration",
160                "## References",
161            ]:
162                if section not in text:
163                    self.error(skill_file, f"missing required section: {section}")
164            if "Do not activate" not in text:
165                self.warn(skill_file, "missing explicit non-activation boundary")
166 
167        return sorted(skill_names)
168 
169    def parse_frontmatter(self, text: str, path: Path) -> dict[str, str]:
170        if not text.startswith("---\n"):
171            self.error(path, "missing opening frontmatter delimiter")
172            return {}
173        end = text.find("\n---", 4)
174        if end == -1:
175            self.error(path, "missing closing frontmatter delimiter")
176            return {}
177        data: dict[str, str] = {}
178        for line in text[4:end].splitlines():
179            if not line.strip() or line.startswith(" "):
180                continue
181            if ":" not in line:
182                continue
183            key, value = line.split(":", 1)
184            data[key.strip()] = value.strip().strip('"')
185        return data
186 
187    def validate_manifests(self, skill_names: list[str]) -> None:
188        marketplace_path = self.root / ".claude-plugin" / "marketplace.json"
189        plugin_path = self.root / ".plugin" / "plugin.json"
190        marketplace = self.load_json(marketplace_path)
191        plugin = self.load_json(plugin_path)
192        if not marketplace or not plugin:
193            return
194 
195        try:
196            plugins = marketplace["plugins"]
197            if len(plugins) != 1:
198                self.error(marketplace_path, "marketplace must contain exactly one bundled plugin")
199                return
200            plugin_entry = plugins[0]
201            if plugin_entry.get("source") != "./":
202                self.error(marketplace_path, "plugin source must be './'")
203            manifest_paths = plugin_entry["skills"]
204        except (KeyError, IndexError, TypeError):
205            self.error(marketplace_path, "plugins[0].skills missing or invalid")
206            return
207 
208        manifest_names = sorted(Path(p).name for p in manifest_paths)
209        if len(manifest_paths) != len(set(manifest_paths)):
210            self.error(marketplace_path, "duplicate skill paths in marketplace manifest")
211        if manifest_names != sorted(skill_names):
212            self.error(
213                marketplace_path,
214                f"manifest skills differ from skills directory: manifest={manifest_names} skills={skill_names}",
215            )
216        for raw_path in manifest_paths:
217            if Path(raw_path).is_absolute() or ".." in Path(raw_path).parts:
218                self.error(marketplace_path, f"skill path escapes repo: {raw_path}")
219                continue
220            path = self.root / raw_path
221            if not path.exists():
222                self.error(marketplace_path, f"skill path does not exist: {raw_path}")
223 
224        marketplace_version = str(marketplace.get("metadata", {}).get("version", ""))
225        plugin_version = str(plugin.get("version", ""))
226        if marketplace_version != plugin_version:
227            self.warn(
228                plugin_path,
229                f"plugin version {plugin_version} differs from marketplace metadata {marketplace_version}",
230            )
231 
232    def validate_docs(self, skill_names: list[str]) -> None:
233        readme = self.root / "README.md"
234        if not readme.exists():
235            self.error(readme, "required doc missing")
236        else:
237            text = readme.read_text(encoding="utf-8")
238            for skill in skill_names:
239                if f"skills/{skill}/" not in text and f"`{skill}`" not in text:
240                    self.error(readme, f"missing exact published skill mention: {skill}")
241 
242        root_skill = self.root / "SKILL.md"
243        if not root_skill.exists():
244            self.error(root_skill, "required doc missing")
245        else:
246            text = root_skill.read_text(encoding="utf-8")
247            for skill in skill_names:
248                if f"skills/{skill}/SKILL.md" not in text:
249                    self.error(root_skill, f"missing exact internal skill path: {skill}")
250 
251        claude = self.root / "CLAUDE.md"
252        if claude.exists():
253            text = claude.read_text(encoding="utf-8")
254            expected = f"{len(skill_names)} skill"
255            if expected not in text:
256                self.warn(claude, f"does not mention current skill count phrase '{expected}'")
257 
258    def validate_researcher(self) -> None:
259        researcher = self.root / "researcher"
260        for relative in REQUIRED_RESEARCHER_FILES:
261            path = researcher / relative
262            if not path.exists():
263                self.error(path, "required researcher OS file missing")
264        for relative in ["templates/source-evaluation.json"]:
265            self.load_json(researcher / relative)
266 
267        content_rubric = researcher / "rubrics" / "content-curation.md"
268        if content_rubric.exists():
269            text = content_rubric.read_text(encoding="utf-8")
270            if "any failed gate rejects" not in text.lower() and "any gate fails" not in text.lower():
271                self.warn(content_rubric, "gate failure semantics are not explicit")
272 
273    def validate_rubrics(self) -> None:
274        rubric_requirements = {
275            "content-curation.md": ["G1", "G2", "G3", "G4", "O1", "O2", "O3", "O4", "1.4", "0.9"],
276            "skill-change.md": ["S1", "S2", "S3", "S4", "S5", "1.4"],
277            "harness-change.md": ["H1", "H2", "H3", "H4", "H5", "1.5"],
278            "pairwise-skill-revision.md": ["Behavioral Improvement", "Evidence Fidelity", "Activation Clarity", "Corpus Fit", "Simplicity"],
279        }
280        rubric_dir = self.root / "researcher" / "rubrics"
281        for filename, required_terms in rubric_requirements.items():
282            path = rubric_dir / filename
283            if not path.exists():
284                continue
285            text = path.read_text(encoding="utf-8")
286            for term in required_terms:
287                if term not in text:
288                    self.error(path, f"missing rubric term or threshold: {term}")
289 
290    def validate_mechanisms(self, skill_names: list[str]) -> None:
291        path = self.root / "researcher" / "mechanisms" / "registry.jsonl"
292        if not path.exists():
293            return
294        claim_ids = {
295            entry.get("claim_id")
296            for _, entry in self.load_jsonl(self.root / "researcher" / "claims" / "index.jsonl")
297            if entry.get("claim_id")
298        }
299        seen: set[str] = set()
300        required = {"mechanism_id", "owning_skill", "status", "activation_scenario", "behavior_change", "evidence", "failure_modes"}
301        valid_status = {"accepted", "candidate", "deprecated", "rejected"}
302        for line_number, line in enumerate(path.read_text(encoding="utf-8").splitlines(), start=1):
303            if not line.strip():
304                continue
305            try:
306                entry = json.loads(line)
307            except json.JSONDecodeError as exc:
308                self.error(path, f"line {line_number} invalid JSON: {exc}")
309                continue
310            missing = required - set(entry)
311            if missing:
312                self.error(path, f"line {line_number} missing fields: {sorted(missing)}")
313            mechanism_id = entry.get("mechanism_id")
314            if mechanism_id in seen:
315                self.error(path, f"duplicate mechanism_id: {mechanism_id}")
316            if isinstance(mechanism_id, str):
317                seen.add(mechanism_id)
318                if not re.fullmatch(r"[a-z0-9]+(?:-[a-z0-9]+)*", mechanism_id):
319                    self.error(path, f"invalid mechanism_id: {mechanism_id}")
320            if entry.get("status") not in valid_status:
321                self.error(path, f"line {line_number} invalid status: {entry.get('status')}")
322            if entry.get("owning_skill") not in skill_names:
323                self.error(path, f"line {line_number} owning_skill is not published: {entry.get('owning_skill')}")
324            for key in ["activation_scenario", "behavior_change"]:
325                if not isinstance(entry.get(key), str) or not entry.get(key, "").strip():
326                    self.error(path, f"line {line_number} {key} must be a non-empty string")
327            if not isinstance(entry.get("evidence"), list) or not entry.get("evidence"):
328                self.error(path, f"line {line_number} evidence must be a non-empty list")
329            else:
330                for evidence in entry["evidence"]:
331                    if not isinstance(evidence, str):
332                        self.error(path, f"line {line_number} evidence entries must be strings")
333                        continue
334                    if re.match(r"https?://", evidence):
335                        continue
336                    if evidence in claim_ids:
337                        continue
338                    if not (self.root / evidence).exists():
339                        self.error(path, f"line {line_number} evidence path does not exist: {evidence}")
340            if not isinstance(entry.get("failure_modes"), list) or not entry.get("failure_modes"):
341                self.error(path, f"line {line_number} failure_modes must be a non-empty list")
342 
343        for ledger_name in ["accepted.jsonl", "rejected.jsonl"]:
344            ledger_path = self.root / "researcher" / "mechanisms" / "ledgers" / ledger_name
345            for line_number, entry in self.load_jsonl(ledger_path):
346                if not entry.get("mechanism_id"):
347                    self.error(ledger_path, f"line {line_number} missing mechanism_id")
348                if not entry.get("rationale"):
349                    self.error(ledger_path, f"line {line_number} missing rationale")
350 
351    def validate_claims(self, skill_names: list[str]) -> None:
352        path = self.root / "researcher" / "claims" / "index.jsonl"
353        seen: set[str] = set()
354        valid_strength = {"primary", "secondary", "anecdotal", "derived"}
355        valid_volatility = {"low", "medium", "high"}
356        for line_number, entry in self.load_jsonl(path):
357            claim_id = entry.get("claim_id")
358            if not isinstance(claim_id, str) or not claim_id:
359                self.error(path, f"line {line_number} missing claim_id")
360            elif claim_id in seen:
361                self.error(path, f"duplicate claim_id: {claim_id}")
362            else:
363                seen.add(claim_id)
364            if entry.get("owning_skill") not in skill_names:
365                self.error(path, f"line {line_number} owning_skill is not published: {entry.get('owning_skill')}")
366            for key in ["claim_text", "section", "source_url", "retrieved_at", "last_reviewed"]:
367                if not isinstance(entry.get(key), str) or not entry.get(key, "").strip():
368                    self.error(path, f"line {line_number} {key} must be a non-empty string")
369            source_url = entry.get("source_url")
370            if isinstance(source_url, str) and not re.match(r"https?://", source_url):
371                if not (self.root / source_url).exists():
372                    self.error(path, f"line {line_number} source path does not exist: {source_url}")
373            if entry.get("evidence_strength") not in valid_strength:
374                self.error(path, f"line {line_number} invalid evidence_strength")
375            if entry.get("volatility") not in valid_volatility:
376                self.error(path, f"line {line_number} invalid volatility")
377 
378    def validate_corpus_index(self, skill_names: list[str]) -> None:
379        path = self.root / "researcher" / "corpus" / "index.json"
380        data = self.load_json(path)
381        if not isinstance(data, dict):
382            return
383        skills = data.get("skills")
384        if not isinstance(skills, list):
385            self.error(path, "skills must be a list")
386            return
387        indexed_names = sorted(item.get("name") for item in skills if isinstance(item, dict))
388        if indexed_names != sorted(skill_names):
389            self.error(path, f"corpus skill index differs from skills directory: {indexed_names} != {skill_names}")
390        claim_ids = {entry.get("claim_id") for _, entry in self.load_jsonl(self.root / "researcher" / "claims" / "index.jsonl")}
391        mechanism_ids = {
392            entry.get("mechanism_id")
393            for _, entry in self.load_jsonl(self.root / "researcher" / "mechanisms" / "registry.jsonl")
394        }
395        for item in skills:
396            if not isinstance(item, dict):
397                self.error(path, "skill index entries must be objects")
398                continue
399            skill_path = item.get("path")
400            if not isinstance(skill_path, str) or not (self.root / skill_path).exists():
401                self.error(path, f"skill path missing: {skill_path}")
402            for claim_id in item.get("claim_ids", []):
403                if claim_id not in claim_ids:
404                    self.error(path, f"unknown claim_id in corpus index: {claim_id}")
405            for mechanism_id in item.get("mechanism_ids", []):
406                if mechanism_id not in mechanism_ids:
407                    self.error(path, f"unknown mechanism_id in corpus index: {mechanism_id}")
408 
409    def validate_activation_cases(self, skill_names: list[str]) -> None:
410        path = self.root / "researcher" / "fixtures" / "activation-cases.jsonl"
411        for line_number, entry in self.load_jsonl(path):
412            expected = entry.get("expected_primary_skill")
413            if expected not in skill_names:
414                self.error(path, f"line {line_number} expected_primary_skill is not published: {expected}")
415            for key in ["case_id", "prompt", "reason"]:
416                if not isinstance(entry.get(key), str) or not entry.get(key, "").strip():
417                    self.error(path, f"line {line_number} {key} must be a non-empty string")
418            for key in ["acceptable_secondary_skills", "rejected_skills"]:
419                values = entry.get(key)
420                if not isinstance(values, list):
421                    self.error(path, f"line {line_number} {key} must be a list")
422                    continue
423                for value in values:
424                    if value not in skill_names:
425                        self.error(path, f"line {line_number} unknown skill in {key}: {value}")
426 
427    def validate_benchmark_scenarios(self) -> None:
428        scenario_dir = self.root / "researcher" / "benchmarks" / "scenarios"
429        for path in sorted(scenario_dir.glob("*.jsonl")):
430            for line_number, entry in self.load_jsonl(path):
431                for key in ["scenario_id", "class", "description", "expected_gate", "deterministic_signal"]:
432                    if not isinstance(entry.get(key), str) or not entry.get(key, "").strip():
433                        self.error(path, f"line {line_number} {key} must be a non-empty string")
434 
435    def validate_runs(self) -> None:
436        runs_dir = self.root / "researcher" / "runs"
437        if not runs_dir.exists():
438            return
439        for run_dir in sorted(p for p in runs_dir.iterdir() if p.is_dir()):
440            required_paths = [
441                run_dir / "THREAD.md",
442                run_dir / "run-state.json",
443                run_dir / "sources" / "queue.jsonl",
444                run_dir / "sources" / "evaluations",
445                run_dir / "sources" / "evidence" / "raw",
446                run_dir / "proposals",
447                run_dir / "proposals" / "mechanism-proposal.jsonl",
448            ]
449            for path in required_paths:
450                if not path.exists():
451                    self.error(path, "run artifact missing")
452            queue = run_dir / "sources" / "queue.jsonl"
453            if queue.exists():
454                for line_number, line in enumerate(queue.read_text(encoding="utf-8").splitlines(), start=1):
455                    if not line.strip():
456                        continue
457                    try:
458                        json.loads(line)
459                    except json.JSONDecodeError as exc:
460                        self.error(queue, f"line {line_number} invalid JSONL: {exc}")
461            report = run_dir / "reports" / "validation-report.json"
462            if report.exists():
463                data = self.load_json(report)
464                if isinstance(data, dict) and data.get("ok") is not True:
465                    self.error(report, "run validation report is not passing")
466            state = run_dir / "run-state.json"
467            if state.exists():
468                data = self.load_json(state)
469                if isinstance(data, dict):
470                    if data.get("current_state") not in {
471                        "initialized",
472                        "retrieved",
473                        "evaluated",
474                        "proposed",
475                        "novelty_checked",
476                        "validated",
477                        "pr_ready",
478                        "closed",
479                    }:
480                        self.error(state, "invalid current_state")
481                    if not isinstance(data.get("state_history"), list) or not data["state_history"]:
482                        self.error(state, "state_history must be a non-empty list")
483 
484    def validate_root_provenance(self) -> None:
485        for path in self.root.glob("autonomous-research-*.json"):
486            self.error(
487                path,
488                "raw research artifact must live under researcher/runs/<run>/sources/evidence/raw or be excluded",
489            )
490 
491    def validate_source_evaluations(self) -> None:
492        candidates: list[Path] = []
493        for base in [self.root / "researcher" / "runs", self.root / "researcher" / "fixtures"]:
494            if base.exists():
495                candidates.extend(base.rglob("*.json"))
496 
497        for path in candidates:
498            data = self.load_json(path)
499            if not isinstance(data, dict):
500                continue
501            if REQUIRED_SOURCE_EVAL_KEYS.issubset(data.keys()):
502                if "draft" in path.stem:
503                    continue
504                self.validate_source_eval_shape(path, data)
505 
506    def validate_source_eval_shape(self, path: Path, data: dict[str, Any]) -> None:
507        source = data.get("source", {})
508        if source.get("retrieval_status") == "failed":
509            decision = data.get("decision", {}).get("verdict")
510            if decision != "REJECT":
511                self.error(path, "failed retrieval must reject or reroute before evaluation")
512        if source.get("retrieval_status") != "retrieved" and data.get("decision", {}).get("verdict") == "APPROVE":
513            self.error(path, "only retrieved sources may receive APPROVE")
514 
515        gatekeeper = data.get("gatekeeper", {})
516        gate_values = [
517            gatekeeper.get("G1_mechanism_specificity", {}).get("pass"),
518            gatekeeper.get("G2_implementable_artifacts", {}).get("pass"),
519            gatekeeper.get("G3_beyond_basics", {}).get("pass"),
520            gatekeeper.get("G4_source_verifiability", {}).get("pass"),
521        ]
522        if not all(isinstance(value, bool) for value in gate_values):
523            self.error(path, "all gate pass values must be booleans")
524            return
525        expected_gatekeeper = "PASS" if all(gate_values) else "REJECT"
526        if gatekeeper.get("verdict") != expected_gatekeeper:
527            self.error(path, f"gatekeeper verdict must be {expected_gatekeeper}")
528 
529        scoring = data.get("scoring", {})
530        score_keys = [
531            "D1_technical_depth_actionability",
532            "D2_repo_relevance",
533            "D3_evidence_rigor",
534            "D4_novelty_insight",
535        ]
536        scores: dict[str, float] = {}
537        for key in score_keys:
538            score = scoring.get(key, {}).get("score")
539            if not isinstance(score, (int, float)) or score < 0 or score > 2:
540                self.error(path, f"{key}.score must be a number from 0 to 2")
541                return
542            scores[key] = float(score)
543 
544        recomputed_total = (
545            scores["D1_technical_depth_actionability"] * 0.35
546            + scores["D2_repo_relevance"] * 0.30
547            + scores["D3_evidence_rigor"] * 0.20
548            + scores["D4_novelty_insight"] * 0.15
549        )
550        recorded_total = scoring.get("weighted_total")
551        if not isinstance(recorded_total, (int, float)):
552            self.error(path, "scoring.weighted_total must be numeric")
553            return
554        if abs(float(recorded_total) - recomputed_total) > 0.011:
555            self.error(
556                path,
557                f"weighted_total {recorded_total} does not match recomputed {recomputed_total:.3f}",
558            )
559 
560        expected_decision = self.expected_content_decision(all(gate_values), scores, recomputed_total)
561        decision = data.get("decision", {})
562        if decision.get("verdict") != expected_decision["verdict"]:
563            self.error(
564                path,
565                f"decision verdict must be {expected_decision['verdict']} under content-curation rubric",
566            )
567        expected_override = expected_decision["override_triggered"]
568        actual_override = decision.get("override_triggered")
569        if actual_override == "null":
570            actual_override = None
571        if actual_override != expected_override:
572            self.error(path, f"override_triggered must be {expected_override or 'null'}")
573 
574    def expected_content_decision(
575        self,
576        gates_pass: bool,
577        scores: dict[str, float],
578        total: float,
579    ) -> dict[str, str | None]:
580        if not gates_pass:
581            return {"verdict": "REJECT", "override_triggered": None}
582        if scores["D1_technical_depth_actionability"] == 0:
583            return {"verdict": "REJECT", "override_triggered": "O1"}
584        if scores["D2_repo_relevance"] == 0:
585            return {"verdict": "REJECT", "override_triggered": "O2"}
586        if scores["D3_evidence_rigor"] == 1 and total >= 1.4:
587            return {"verdict": "HUMAN_REVIEW", "override_triggered": "O3"}
588        if scores["D4_novelty_insight"] == 2 and total < 1.4:
589            return {"verdict": "HUMAN_REVIEW", "override_triggered": "O4"}
590        if total >= 1.4:
591            return {"verdict": "APPROVE", "override_triggered": None}
592        if total >= 0.9:
593            return {"verdict": "HUMAN_REVIEW", "override_triggered": None}
594        return {"verdict": "REJECT", "override_triggered": None}
595 
596    def load_json(self, path: Path) -> Any:
597        if not path.exists():
598            self.error(path, "JSON file missing")
599            return None
600        try:
601            return json.loads(path.read_text(encoding="utf-8"), object_pairs_hook=self.reject_duplicate_keys(path))
602        except json.JSONDecodeError as exc:
603            self.error(path, f"invalid JSON: {exc}")
604            return None
605 
606    def load_jsonl(self, path: Path) -> list[tuple[int, dict[str, Any]]]:
607        if not path.exists():
608            self.error(path, "JSONL file missing")
609            return []
610        entries: list[tuple[int, dict[str, Any]]] = []
611        for line_number, line in enumerate(path.read_text(encoding="utf-8").splitlines(), start=1):
612            if not line.strip():
613                continue
614            try:
615                data = json.loads(line, object_pairs_hook=self.reject_duplicate_keys(path))
616            except json.JSONDecodeError as exc:
617                self.error(path, f"line {line_number} invalid JSONL: {exc}")
618                continue
619            if not isinstance(data, dict):
620                self.error(path, f"line {line_number} JSONL entry must be an object")
621                continue
622            entries.append((line_number, data))
623        return entries
624 
625    def reject_duplicate_keys(self, path: Path):
626        def hook(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
627            result: dict[str, Any] = {}
628            for key, value in pairs:
629                if key in result:
630                    self.error(path, f"duplicate JSON key: {key}")
631                result[key] = value
632            return result
633 
634        return hook
635 
636 
637def main() -> int:
638    parser = argparse.ArgumentParser(description="Validate skill corpus and researcher harness")
639    parser.add_argument("--root", type=Path, default=Path(__file__).resolve().parents[2])
640    parser.add_argument("--json", action="store_true", help="print machine-readable JSON")
641    parser.add_argument("--strict", action="store_true", help="treat warnings as failures")
642    args = parser.parse_args()
643 
644    result = Validator(args.root).run()
645    if args.json:
646        print(json.dumps(result, indent=2))
647    else:
648        summary = result["summary"]
649        print(
650            f"Validation {'passed' if result['ok'] else 'failed'}: "
651            f"{summary['errors']} errors, {summary['warnings']} warnings, "
652            f"{summary['skill_count']} skills"
653        )
654        for finding in result["findings"]:
655            print(f"[{finding['severity']}] {finding['path']}: {finding['message']}")
656 
657    if result["summary"]["errors"] > 0:
658        return 1
659    if args.strict and result["summary"]["warnings"] > 0:
660        return 1
661    return 0
662 
663 
664if __name__ == "__main__":
665    sys.exit(main())
666