mirror of
https://git.openapi.site/https://github.com/desirecore/market.git
synced 2026-09-05 18:23:50 +08:00
feat: 迁移统一目录元数据契约 (#102)
## 中文 - 为 System Agent、34 个 Builtin Skill、28 个 Pointer/Collection 条目增加 catalog sidecar - 覆盖 147 个 collection child，并固定可证明的来源；无法证明的内容保持 listing-only/unknown - 增加 strict Schema、validator、collection check 与 CI 完整性门禁 ## English - Add catalog metadata sidecars for the System Agent, 34 built-in Skills, and 28 pointer/collection entries - Cover 147 collection children while keeping unverifiable facts listing-only or unknown - Add strict schemas, validators, deterministic collection checks, and CI completeness gates ## 验证 / Verification - Catalog validator 17/17 - Collection generator 4/4 - 63 sidecars, 147 children, zero errors
This commit is contained in:
GitHub
96
scripts/catalog/test_collection_generator.py
Normal file
96
scripts/catalog/test_collection_generator.py
Normal file
@@ -0,0 +1,96 @@
|
||||
#!/usr/bin/env -S uv run --script
|
||||
# /// script
|
||||
# requires-python = ">=3.10"
|
||||
# dependencies = ["pyyaml>=6.0"]
|
||||
# ///
|
||||
"""Unit tests for collection generation check mode."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib.util
|
||||
import json
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
|
||||
GENERATOR_PATH = Path(__file__).parents[1] / "gen-collection-children.py"
|
||||
SPEC = importlib.util.spec_from_file_location("market_collection_generator", GENERATOR_PATH)
|
||||
assert SPEC is not None and SPEC.loader is not None
|
||||
GENERATOR = importlib.util.module_from_spec(SPEC)
|
||||
sys.modules[SPEC.name] = GENERATOR
|
||||
SPEC.loader.exec_module(GENERATOR)
|
||||
|
||||
|
||||
class CollectionGeneratorCheckTests(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
self.tempdir = tempfile.TemporaryDirectory()
|
||||
self.skills_dir = Path(self.tempdir.name) / "skills"
|
||||
self.entry_path = self.skills_dir / "example-collection" / "entry.json"
|
||||
self.entry_path.parent.mkdir(parents=True)
|
||||
self.children = [{"id": "child-one", "path": "skills/child-one"}]
|
||||
self.write_entry(self.children)
|
||||
|
||||
def tearDown(self) -> None:
|
||||
self.tempdir.cleanup()
|
||||
|
||||
def write_entry(self, children: list[dict[str, str]]) -> None:
|
||||
self.entry_path.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"id": "example-collection",
|
||||
"source": {
|
||||
"kind": "git",
|
||||
"repoUrl": "https://example.com/example.git",
|
||||
"repoBranch": "main",
|
||||
"ref": "a" * 40,
|
||||
},
|
||||
"children": children,
|
||||
},
|
||||
indent=2,
|
||||
)
|
||||
+ "\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
def run_check(self, discovered: list[dict[str, str]]) -> bool:
|
||||
before = self.entry_path.read_bytes()
|
||||
with (
|
||||
patch.object(GENERATOR, "SKILLS_DIR", self.skills_dir),
|
||||
patch.object(GENERATOR, "clone_pinned"),
|
||||
patch.object(GENERATOR, "discover_children", return_value=discovered),
|
||||
):
|
||||
result = GENERATOR.process("example-collection", check=True)
|
||||
self.assertEqual(before, self.entry_path.read_bytes(), "--check must never write entry.json")
|
||||
return result
|
||||
|
||||
def test_check_accepts_current_children_without_writing(self) -> None:
|
||||
self.assertTrue(self.run_check(self.children))
|
||||
|
||||
def test_check_rejects_stale_children_without_writing(self) -> None:
|
||||
self.assertFalse(self.run_check([{"id": "child-two", "path": "skills/child-two"}]))
|
||||
|
||||
def test_check_rejects_collection_source_path(self) -> None:
|
||||
entry = json.loads(self.entry_path.read_text(encoding="utf-8"))
|
||||
entry["source"]["path"] = "skills"
|
||||
self.entry_path.write_text(json.dumps(entry, indent=2) + "\n", encoding="utf-8")
|
||||
self.assertFalse(self.run_check(self.children))
|
||||
|
||||
def test_check_skips_unpinned_collection_without_network_or_writes(self) -> None:
|
||||
entry = json.loads(self.entry_path.read_text(encoding="utf-8"))
|
||||
del entry["source"]["ref"]
|
||||
self.entry_path.write_text(json.dumps(entry, indent=2) + "\n", encoding="utf-8")
|
||||
before = self.entry_path.read_bytes()
|
||||
with (
|
||||
patch.object(GENERATOR, "SKILLS_DIR", self.skills_dir),
|
||||
patch.object(GENERATOR, "clone_pinned") as clone,
|
||||
):
|
||||
self.assertTrue(GENERATOR.process("example-collection", check=True))
|
||||
clone.assert_not_called()
|
||||
self.assertEqual(before, self.entry_path.read_bytes())
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
440
scripts/catalog/test_validate_catalog_metadata.py
Normal file
440
scripts/catalog/test_validate_catalog_metadata.py
Normal file
@@ -0,0 +1,440 @@
|
||||
#!/usr/bin/env -S uv run --script
|
||||
# /// script
|
||||
# requires-python = ">=3.10"
|
||||
# dependencies = ["jsonschema>=4.23,<5", "pyyaml>=6.0"]
|
||||
# ///
|
||||
"""Unit tests for the Market catalog metadata sidecar validator."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import copy
|
||||
import importlib.util
|
||||
import json
|
||||
import shutil
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
VALIDATOR_PATH = Path(__file__).with_name("validate_catalog_metadata.py")
|
||||
SPEC = importlib.util.spec_from_file_location("market_catalog_metadata_validator", VALIDATOR_PATH)
|
||||
assert SPEC is not None and SPEC.loader is not None
|
||||
VALIDATOR = importlib.util.module_from_spec(SPEC)
|
||||
sys.modules[SPEC.name] = VALIDATOR
|
||||
SPEC.loader.exec_module(VALIDATOR)
|
||||
SOURCE_SCHEMA = VALIDATOR_PATH.parents[2] / "schemas" / "catalog-metadata.v1.schema.json"
|
||||
|
||||
|
||||
def unknown() -> dict[str, str]:
|
||||
return {"state": "unknown"}
|
||||
|
||||
|
||||
def valid_entry() -> dict[str, object]:
|
||||
return {
|
||||
"id": "example-skill",
|
||||
"name": "Example Skill",
|
||||
"category": "development",
|
||||
"tags": ["example"],
|
||||
"i18n": {
|
||||
"zh-CN": {"name": "示例技能", "shortDesc": "中文简介"},
|
||||
"en-US": {"name": "Example Skill", "shortDesc": "English summary"},
|
||||
},
|
||||
"maintainer": {"name": "Example Maintainer", "verified": False},
|
||||
"stewardship": "community",
|
||||
"license": "MIT",
|
||||
"redistribution": "allowed",
|
||||
"source": {
|
||||
"kind": "git",
|
||||
"repoUrl": "https://example.com/example-skill.git",
|
||||
"repoBranch": "main",
|
||||
"ref": "a" * 40,
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def valid_sidecar() -> dict[str, object]:
|
||||
return {
|
||||
"$schema": "../../schemas/catalog-metadata.v1.schema.json",
|
||||
"schemaVersion": 1,
|
||||
"identity": {"kind": "skill", "id": "example-skill"},
|
||||
"presentation": {
|
||||
"defaultLocale": "en-US",
|
||||
"i18n": {
|
||||
"zh-CN": {"name": "示例技能", "summary": "中文简介"},
|
||||
"en-US": {"name": "Example Skill", "summary": "English summary"},
|
||||
},
|
||||
"category": "development",
|
||||
"tags": ["example"],
|
||||
},
|
||||
"release": unknown(),
|
||||
"timestamps": {
|
||||
"catalogUpdatedAt": unknown(),
|
||||
"releasePublishedAt": unknown(),
|
||||
"reviewedAt": unknown(),
|
||||
"upstreamObservedAt": unknown(),
|
||||
},
|
||||
"provenance": {
|
||||
"content": {
|
||||
"kind": "git",
|
||||
"url": "https://example.com/example-skill.git",
|
||||
"ref": "a" * 40,
|
||||
}
|
||||
},
|
||||
"governance": {
|
||||
"stewardship": "community",
|
||||
"availability": "listing-only",
|
||||
"license": {"state": "known", "value": "MIT"},
|
||||
"redistribution": "allowed",
|
||||
"upstreamMaintainer": {"name": "Example Maintainer", "verified": False},
|
||||
},
|
||||
"compatibility": {"platforms": unknown()},
|
||||
"spec": {"kind": "skill", "riskLevel": "low"},
|
||||
}
|
||||
|
||||
|
||||
class CatalogMetadataValidatorTests(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
self.tempdir = tempfile.TemporaryDirectory()
|
||||
self.root = Path(self.tempdir.name)
|
||||
(self.root / "schemas").mkdir()
|
||||
shutil.copyfile(SOURCE_SCHEMA, self.root / "schemas" / SOURCE_SCHEMA.name)
|
||||
(self.root / "agents").mkdir()
|
||||
(self.root / "skills" / "example-skill").mkdir(parents=True)
|
||||
self.write_json(
|
||||
self.root / "manifest.json",
|
||||
{
|
||||
"supportedLocales": ["zh-CN", "en-US"],
|
||||
"stats": {"totalAgents": 0, "totalSkills": 1},
|
||||
},
|
||||
)
|
||||
self.entry_path = self.root / "skills" / "example-skill" / "entry.json"
|
||||
self.sidecar_path = self.root / "skills" / "example-skill" / VALIDATOR.SIDECAR_NAME
|
||||
self.write_json(self.entry_path, valid_entry())
|
||||
self.write_json(self.sidecar_path, valid_sidecar())
|
||||
|
||||
def tearDown(self) -> None:
|
||||
self.tempdir.cleanup()
|
||||
|
||||
@staticmethod
|
||||
def write_json(path: Path, value: object) -> None:
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
||||
|
||||
def validate(self, *, require_complete: bool = False):
|
||||
return VALIDATOR.validate_repository(self.root, require_complete=require_complete)
|
||||
|
||||
def rewrite_sidecar(self, mutate) -> None:
|
||||
payload = valid_sidecar()
|
||||
mutate(payload)
|
||||
self.write_json(self.sidecar_path, payload)
|
||||
|
||||
def write_agent_case(self, install_policy: str | None, update_policy: str | None, availability: str) -> None:
|
||||
agent_dir = self.root / "agents" / "system-agent"
|
||||
agent_dir.mkdir()
|
||||
self.write_json(
|
||||
agent_dir / "agent.json",
|
||||
{
|
||||
"id": "system-agent",
|
||||
"category": "development",
|
||||
"i18n": {
|
||||
"default_locale": "en-US",
|
||||
"source_locale": "zh-CN",
|
||||
"locales": ["zh-CN", "en-US"],
|
||||
"zh-CN": {"name": "系统智能体", "shortDesc": "系统智能体简介"},
|
||||
"en-US": {"name": "System Agent", "shortDesc": "System agent summary"},
|
||||
},
|
||||
},
|
||||
)
|
||||
sidecar = valid_sidecar()
|
||||
sidecar["identity"] = {"kind": "agent", "id": "system-agent"}
|
||||
sidecar["presentation"] = {
|
||||
"defaultLocale": "en-US",
|
||||
"i18n": {
|
||||
"zh-CN": {"name": "系统智能体", "summary": "系统智能体简介"},
|
||||
"en-US": {"name": "System Agent", "summary": "System agent summary"},
|
||||
},
|
||||
"category": "development",
|
||||
"tags": [],
|
||||
}
|
||||
sidecar["provenance"] = {}
|
||||
sidecar["governance"] = {
|
||||
"availability": availability,
|
||||
"license": unknown(),
|
||||
"redistribution": "verify-package-terms",
|
||||
}
|
||||
spec = {"kind": "agent"}
|
||||
if install_policy is not None:
|
||||
spec["installPolicy"] = install_policy
|
||||
if update_policy is not None:
|
||||
spec["updatePolicy"] = update_policy
|
||||
sidecar["spec"] = spec
|
||||
self.write_json(agent_dir / VALIDATOR.SIDECAR_NAME, sidecar)
|
||||
manifest = json.loads((self.root / "manifest.json").read_text(encoding="utf-8"))
|
||||
manifest["stats"]["totalAgents"] = 1
|
||||
self.write_json(self.root / "manifest.json", manifest)
|
||||
|
||||
def test_accepts_valid_listing_only_pointer(self) -> None:
|
||||
report = self.validate(require_complete=True)
|
||||
self.assertFalse(report.has_errors, report.issues)
|
||||
self.assertEqual(1, report.stats["sidecars"])
|
||||
self.assertEqual(1, report.stats["publishableSkills"])
|
||||
|
||||
def test_accepts_system_repository_listing_only_agent_policy(self) -> None:
|
||||
self.write_agent_case("system", "repository", "listing-only")
|
||||
report = self.validate()
|
||||
self.assertFalse(report.has_errors, report.issues)
|
||||
|
||||
def test_rejects_system_agent_content_release(self) -> None:
|
||||
self.write_agent_case("system", "repository", "listing-only")
|
||||
sidecar_path = self.root / "agents" / "system-agent" / VALIDATOR.SIDECAR_NAME
|
||||
payload = json.loads(sidecar_path.read_text(encoding="utf-8"))
|
||||
payload["release"] = {"state": "known", "version": "1.2.0", "versionScheme": "semver"}
|
||||
self.write_json(sidecar_path, payload)
|
||||
issues = self.validate().issues
|
||||
self.assertTrue(any(issue.rule in {"catalog-schema", "agent-policy"} for issue in issues))
|
||||
|
||||
def test_rejects_system_agent_content_release_timestamp(self) -> None:
|
||||
self.write_agent_case("system", "repository", "listing-only")
|
||||
sidecar_path = self.root / "agents" / "system-agent" / VALIDATOR.SIDECAR_NAME
|
||||
payload = json.loads(sidecar_path.read_text(encoding="utf-8"))
|
||||
payload["timestamps"]["releasePublishedAt"] = {
|
||||
"state": "known",
|
||||
"value": "2026-08-13",
|
||||
"precision": "day",
|
||||
}
|
||||
self.write_json(sidecar_path, payload)
|
||||
issues = self.validate().issues
|
||||
self.assertTrue(any(issue.rule == "agent-policy" for issue in issues))
|
||||
|
||||
def test_accepts_market_market_listing_only_agent_policy(self) -> None:
|
||||
self.write_agent_case("market", "market", "listing-only")
|
||||
report = self.validate()
|
||||
self.assertFalse(report.has_errors, report.issues)
|
||||
|
||||
def test_rejects_invalid_agent_policy_combinations(self) -> None:
|
||||
cases = [
|
||||
("system", "market", "listing-only"),
|
||||
("system", "repository", "installable"),
|
||||
("market", "repository", "listing-only"),
|
||||
("system", None, "listing-only"),
|
||||
(None, "repository", "listing-only"),
|
||||
]
|
||||
for index, (install_policy, update_policy, availability) in enumerate(cases):
|
||||
with self.subTest(
|
||||
installPolicy=install_policy,
|
||||
updatePolicy=update_policy,
|
||||
availability=availability,
|
||||
):
|
||||
if index:
|
||||
shutil.rmtree(self.root / "agents" / "system-agent")
|
||||
self.write_agent_case(install_policy, update_policy, availability)
|
||||
self.assertTrue(any(issue.rule == "catalog-schema" for issue in self.validate().issues))
|
||||
|
||||
def test_market_agent_installable_still_requires_governance_evidence(self) -> None:
|
||||
self.write_agent_case("market", "market", "installable")
|
||||
issues = self.validate().issues
|
||||
evidence = [issue for issue in issues if issue.rule == "installable-evidence"]
|
||||
self.assertGreaterEqual(len(evidence), 4)
|
||||
|
||||
def test_rejects_provider_identity_catalog_trust_and_runtime_fields(self) -> None:
|
||||
def mutate(payload):
|
||||
payload["identity"]["catalogSourceId"] = "market:official"
|
||||
payload["provenance"]["catalog"] = {"trust": "official"}
|
||||
payload["installStatus"] = "installed"
|
||||
|
||||
self.rewrite_sidecar(mutate)
|
||||
issues = self.validate().issues
|
||||
schema_messages = [issue.message for issue in issues if issue.rule == "catalog-schema"]
|
||||
self.assertTrue(any("Additional properties" in message for message in schema_messages))
|
||||
|
||||
def test_rejects_wrong_time_precision_and_non_utc_second(self) -> None:
|
||||
def mutate(payload):
|
||||
payload["timestamps"]["catalogUpdatedAt"] = {
|
||||
"state": "known",
|
||||
"value": "2026-08-30T12:00:00+08:00",
|
||||
"precision": "day",
|
||||
}
|
||||
|
||||
self.rewrite_sidecar(mutate)
|
||||
self.assertTrue(any(issue.rule == "catalog-schema" for issue in self.validate().issues))
|
||||
|
||||
def test_rejects_legacy_duplicate_mismatch(self) -> None:
|
||||
self.rewrite_sidecar(lambda payload: payload["presentation"].update(category="research"))
|
||||
issues = self.validate().issues
|
||||
self.assertTrue(any(issue.rule == "legacy-consistency" for issue in issues))
|
||||
|
||||
def test_keeps_unverified_legacy_license_unknown_with_warning(self) -> None:
|
||||
def mutate(payload):
|
||||
payload["governance"]["license"] = unknown()
|
||||
|
||||
self.rewrite_sidecar(mutate)
|
||||
report = self.validate()
|
||||
self.assertFalse(report.has_errors, report.issues)
|
||||
self.assertTrue(any(issue.rule == "legacy-license-unverified" for issue in report.issues))
|
||||
|
||||
def test_rejects_installable_without_review_or_immutable_source(self) -> None:
|
||||
def mutate(payload):
|
||||
payload["governance"] = {
|
||||
"availability": "installable",
|
||||
"license": unknown(),
|
||||
"redistribution": "verify-package-terms",
|
||||
}
|
||||
payload["provenance"]["content"]["ref"] = "main"
|
||||
|
||||
self.rewrite_sidecar(mutate)
|
||||
issues = self.validate().issues
|
||||
evidence = [issue.message for issue in issues if issue.rule == "installable-evidence"]
|
||||
self.assertTrue(any("immutable" in message for message in evidence))
|
||||
self.assertTrue(any("known license" in message for message in evidence))
|
||||
self.assertTrue(any("known reviewedAt" in message for message in evidence))
|
||||
self.assertTrue(any("compliance" in message for message in evidence))
|
||||
|
||||
def test_allows_builtin_installation_with_stable_incomplete_governance_warnings(self) -> None:
|
||||
self.entry_path.unlink()
|
||||
skill_path = self.entry_path.with_name("SKILL.md")
|
||||
skill_path.write_text(
|
||||
"""---
|
||||
name: example-skill
|
||||
description: Example builtin skill.
|
||||
type: procedural
|
||||
risk_level: low
|
||||
tags: [example]
|
||||
metadata:
|
||||
author: example
|
||||
i18n:
|
||||
default_locale: en-US
|
||||
source_locale: zh-CN
|
||||
locales: [zh-CN, en-US]
|
||||
zh-CN:
|
||||
name: 示例技能
|
||||
short_desc: 中文简介
|
||||
en-US:
|
||||
name: Example Skill
|
||||
short_desc: English summary
|
||||
market:
|
||||
category: development
|
||||
---
|
||||
Body.
|
||||
""",
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
def mutate(payload):
|
||||
payload["provenance"] = {}
|
||||
payload["governance"] = {
|
||||
"stewardship": "community",
|
||||
"availability": "installable",
|
||||
"license": unknown(),
|
||||
"redistribution": "verify-package-terms",
|
||||
"listingMaintainer": {"name": "Candidate", "verified": False},
|
||||
}
|
||||
payload["spec"] = {"kind": "skill", "riskLevel": "low", "skillType": "procedural"}
|
||||
|
||||
self.rewrite_sidecar(mutate)
|
||||
report = self.validate()
|
||||
self.assertFalse(report.has_errors, report.issues)
|
||||
warning_rules = {issue.rule for issue in report.issues if issue.severity == "warning"}
|
||||
self.assertTrue(
|
||||
{
|
||||
"missing-license-evidence",
|
||||
"missing-review-evidence",
|
||||
"missing-maintainer-evidence",
|
||||
"missing-content-evidence",
|
||||
}.issubset(warning_rules)
|
||||
)
|
||||
|
||||
def test_validates_collection_identity_and_does_not_invent_child_version(self) -> None:
|
||||
entry = valid_entry()
|
||||
entry["children"] = [
|
||||
{
|
||||
"id": "child-one",
|
||||
"path": "skills/child-one",
|
||||
"i18n": {
|
||||
"zh-CN": {"shortDesc": "Same child summary"},
|
||||
"en-US": {"shortDesc": "Same child summary"},
|
||||
},
|
||||
}
|
||||
]
|
||||
self.write_json(self.entry_path, entry)
|
||||
|
||||
def mutate(payload):
|
||||
child_presentation = copy.deepcopy(payload["presentation"])
|
||||
child_presentation["tags"] = []
|
||||
child_presentation["i18n"] = {
|
||||
"zh-CN": {"name": "child-one", "summary": "Same child summary"},
|
||||
"en-US": {"name": "child-one", "summary": "Same child summary"},
|
||||
}
|
||||
payload["spec"]["collection"] = {
|
||||
"role": "parent",
|
||||
"childCount": 1,
|
||||
"children": [
|
||||
{
|
||||
"identity": {
|
||||
"kind": "skill",
|
||||
"id": "child-one",
|
||||
"parentId": "example-skill",
|
||||
},
|
||||
"path": "skills/child-one",
|
||||
"presentation": child_presentation,
|
||||
"release": unknown(),
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
self.rewrite_sidecar(mutate)
|
||||
report = self.validate()
|
||||
self.assertFalse(report.has_errors)
|
||||
self.assertTrue(any(issue.rule == "i18n-freshness" for issue in report.issues))
|
||||
|
||||
def invent_version(payload):
|
||||
mutate(payload)
|
||||
payload["spec"]["collection"]["children"][0]["release"] = {
|
||||
"state": "known",
|
||||
"version": "1.0.0",
|
||||
"versionScheme": "semver",
|
||||
}
|
||||
|
||||
self.rewrite_sidecar(invent_version)
|
||||
self.assertTrue(any(issue.rule == "collection-version" for issue in self.validate().issues))
|
||||
|
||||
def test_marks_identical_locale_payloads_for_review_without_failing(self) -> None:
|
||||
def mutate(payload):
|
||||
payload["presentation"]["i18n"] = {
|
||||
"zh-CN": {"name": "Same", "summary": "Same summary"},
|
||||
"en-US": {"name": "Same", "summary": "Same summary"},
|
||||
}
|
||||
entry = valid_entry()
|
||||
entry["i18n"] = {
|
||||
"zh-CN": {"name": "Same", "shortDesc": "Same summary"},
|
||||
"en-US": {"name": "Same", "shortDesc": "Same summary"},
|
||||
}
|
||||
self.write_json(self.entry_path, entry)
|
||||
|
||||
self.rewrite_sidecar(mutate)
|
||||
report = self.validate()
|
||||
self.assertFalse(report.has_errors, report.issues)
|
||||
self.assertTrue(any(issue.rule == "i18n-freshness" for issue in report.issues))
|
||||
|
||||
def test_rejects_sidecar_outside_fixed_path(self) -> None:
|
||||
rogue = self.root / "metadata" / VALIDATOR.SIDECAR_NAME
|
||||
self.write_json(rogue, valid_sidecar())
|
||||
issues = self.validate().issues
|
||||
self.assertTrue(any(issue.path.startswith("metadata/") and issue.rule == "fixed-sidecar-path" for issue in issues))
|
||||
|
||||
def test_reports_manifest_stats_and_optional_complete_coverage(self) -> None:
|
||||
manifest = {
|
||||
"supportedLocales": ["zh-CN", "en-US"],
|
||||
"stats": {"totalAgents": 1, "totalSkills": 99},
|
||||
}
|
||||
self.write_json(self.root / "manifest.json", manifest)
|
||||
issues = self.validate().issues
|
||||
self.assertEqual(2, sum(issue.rule == "market-stats" for issue in issues))
|
||||
|
||||
self.sidecar_path.unlink()
|
||||
issues = self.validate(require_complete=True).issues
|
||||
self.assertTrue(any(issue.rule == "sidecar-coverage" for issue in issues))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
787
scripts/catalog/validate_catalog_metadata.py
Normal file
787
scripts/catalog/validate_catalog_metadata.py
Normal file
@@ -0,0 +1,787 @@
|
||||
#!/usr/bin/env -S uv run --script
|
||||
# /// script
|
||||
# requires-python = ">=3.10"
|
||||
# dependencies = ["jsonschema>=4.23,<5", "pyyaml>=6.0"]
|
||||
# ///
|
||||
"""Validate Market ``catalog-metadata.v1.json`` sidecars.
|
||||
|
||||
The sidecar is deliberately discovered at one fixed path next to a legacy
|
||||
``agent.json``, ``SKILL.md`` or ``entry.json``. It supplements the legacy
|
||||
file; it never selects an arbitrary metadata path and never declares the
|
||||
trusted catalog provider identity.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import re
|
||||
import sys
|
||||
from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
from typing import Any, Iterable
|
||||
|
||||
import yaml
|
||||
from jsonschema import Draft7Validator, FormatChecker
|
||||
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[2]
|
||||
SCHEMA_PATH = REPO_ROOT / "schemas" / "catalog-metadata.v1.schema.json"
|
||||
SIDECAR_NAME = "catalog-metadata.v1.json"
|
||||
EXPECTED_SCHEMA_REF = "../../schemas/catalog-metadata.v1.schema.json"
|
||||
FRONTMATTER_RE = re.compile(r"^---\s*\n(.*?)\n---\s*\n", re.DOTALL)
|
||||
FULL_GIT_REF_RE = re.compile(r"^[a-fA-F0-9]{40}(?:[a-fA-F0-9]{24})?$")
|
||||
CONTAINER_DIGEST_RE = re.compile(r"^sha256:[a-fA-F0-9]{64}$")
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Issue:
|
||||
path: str
|
||||
rule: str
|
||||
message: str
|
||||
severity: str = "error"
|
||||
|
||||
def to_dict(self) -> dict[str, str]:
|
||||
return {
|
||||
"path": self.path,
|
||||
"rule": self.rule,
|
||||
"message": self.message,
|
||||
"severity": self.severity,
|
||||
}
|
||||
|
||||
|
||||
@dataclass
|
||||
class Report:
|
||||
issues: list[Issue] = field(default_factory=list)
|
||||
stats: dict[str, int] = field(default_factory=dict)
|
||||
|
||||
def add(self, path: str, rule: str, message: str, severity: str = "error") -> None:
|
||||
self.issues.append(Issue(path, rule, message, severity))
|
||||
|
||||
@property
|
||||
def has_errors(self) -> bool:
|
||||
return any(issue.severity == "error" for issue in self.issues)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class LegacyItem:
|
||||
kind: str
|
||||
item_id: str
|
||||
source_type: str
|
||||
data: dict[str, Any]
|
||||
i18n: dict[str, dict[str, Any]]
|
||||
default_locale: str | None
|
||||
version: str | None
|
||||
category: str | None
|
||||
tags: list[str] | None
|
||||
|
||||
|
||||
def _rel(path: Path, root: Path) -> str:
|
||||
try:
|
||||
return path.relative_to(root).as_posix()
|
||||
except ValueError:
|
||||
return path.as_posix()
|
||||
|
||||
|
||||
def _read_json(path: Path, report: Report, root: Path, rule: str) -> dict[str, Any] | None:
|
||||
try:
|
||||
value = json.loads(path.read_text(encoding="utf-8"))
|
||||
except (OSError, json.JSONDecodeError) as exc:
|
||||
report.add(_rel(path, root), rule, f"cannot read JSON object: {exc}")
|
||||
return None
|
||||
if not isinstance(value, dict):
|
||||
report.add(_rel(path, root), rule, "JSON root must be an object")
|
||||
return None
|
||||
return value
|
||||
|
||||
|
||||
def _read_skill_frontmatter(path: Path, report: Report, root: Path) -> dict[str, Any] | None:
|
||||
try:
|
||||
text = path.read_text(encoding="utf-8")
|
||||
except OSError as exc:
|
||||
report.add(_rel(path, root), "legacy-read", f"cannot read SKILL.md: {exc}")
|
||||
return None
|
||||
match = FRONTMATTER_RE.match(text)
|
||||
if not match:
|
||||
report.add(_rel(path, root), "legacy-read", "SKILL.md has no YAML frontmatter")
|
||||
return None
|
||||
try:
|
||||
value = yaml.safe_load(match.group(1)) or {}
|
||||
except yaml.YAMLError as exc:
|
||||
report.add(_rel(path, root), "legacy-read", f"cannot parse YAML frontmatter: {exc}")
|
||||
return None
|
||||
if not isinstance(value, dict):
|
||||
report.add(_rel(path, root), "legacy-read", "SKILL.md frontmatter must be a mapping")
|
||||
return None
|
||||
return value
|
||||
|
||||
|
||||
def _legacy_i18n(raw: Any, *, short_key: str) -> tuple[dict[str, dict[str, Any]], str | None]:
|
||||
if not isinstance(raw, dict):
|
||||
return {}, None
|
||||
default_locale = raw.get("defaultLocale") or raw.get("default_locale")
|
||||
locales_value = raw.get("locales")
|
||||
if isinstance(locales_value, dict):
|
||||
candidates = locales_value.items()
|
||||
else:
|
||||
candidates = (
|
||||
(key, value)
|
||||
for key, value in raw.items()
|
||||
if re.fullmatch(r"[a-z]{2,3}(?:-[A-Z]{2})?", str(key))
|
||||
)
|
||||
locales: dict[str, dict[str, Any]] = {}
|
||||
for locale, payload in candidates:
|
||||
if not isinstance(payload, dict):
|
||||
continue
|
||||
normalized: dict[str, Any] = {}
|
||||
if isinstance(payload.get("name"), str):
|
||||
normalized["name"] = payload["name"]
|
||||
summary = payload.get(short_key)
|
||||
if isinstance(summary, str):
|
||||
normalized["summary"] = summary
|
||||
description = payload.get("description") or payload.get("fullDesc")
|
||||
if isinstance(description, str):
|
||||
normalized["description"] = description
|
||||
locales[str(locale)] = normalized
|
||||
return locales, default_locale if isinstance(default_locale, str) else None
|
||||
|
||||
|
||||
def load_legacy(sidecar: Path, report: Report, root: Path) -> LegacyItem | None:
|
||||
parent = sidecar.parent
|
||||
root_kind = parent.parent.name
|
||||
if root_kind == "agents":
|
||||
legacy_path = parent / "agent.json"
|
||||
data = _read_json(legacy_path, report, root, "legacy-read")
|
||||
if data is None:
|
||||
return None
|
||||
i18n, default_locale = _legacy_i18n(data.get("i18n"), short_key="shortDesc")
|
||||
return LegacyItem(
|
||||
kind="agent",
|
||||
item_id=parent.name,
|
||||
source_type="agent",
|
||||
data=data,
|
||||
i18n=i18n,
|
||||
default_locale=default_locale,
|
||||
version=str(data["version"]) if data.get("version") is not None else None,
|
||||
category=data.get("category") if isinstance(data.get("category"), str) else None,
|
||||
tags=data.get("tags") if isinstance(data.get("tags"), list) else None,
|
||||
)
|
||||
|
||||
if root_kind != "skills":
|
||||
report.add(
|
||||
_rel(sidecar, root),
|
||||
"fixed-sidecar-path",
|
||||
f"{SIDECAR_NAME} is only allowed at agents/<id>/ or skills/<id>/",
|
||||
)
|
||||
return None
|
||||
|
||||
pointer_path = parent / "entry.json"
|
||||
builtin_path = parent / "SKILL.md"
|
||||
if pointer_path.is_file() == builtin_path.is_file():
|
||||
report.add(
|
||||
_rel(sidecar, root),
|
||||
"fixed-sidecar-path",
|
||||
"skill sidecar must be next to exactly one legacy entry.json or SKILL.md",
|
||||
)
|
||||
return None
|
||||
|
||||
if pointer_path.is_file():
|
||||
data = _read_json(pointer_path, report, root, "legacy-read")
|
||||
if data is None:
|
||||
return None
|
||||
i18n, default_locale = _legacy_i18n(data.get("i18n"), short_key="shortDesc")
|
||||
return LegacyItem(
|
||||
kind="skill",
|
||||
item_id=parent.name,
|
||||
source_type="pointer",
|
||||
data=data,
|
||||
i18n=i18n,
|
||||
default_locale=default_locale,
|
||||
version=str(data["version"]) if data.get("version") is not None else None,
|
||||
category=data.get("category") if isinstance(data.get("category"), str) else None,
|
||||
tags=data.get("tags") if isinstance(data.get("tags"), list) else None,
|
||||
)
|
||||
|
||||
data = _read_skill_frontmatter(builtin_path, report, root)
|
||||
if data is None:
|
||||
return None
|
||||
metadata = data.get("metadata") if isinstance(data.get("metadata"), dict) else {}
|
||||
i18n_raw = metadata.get("i18n") if isinstance(metadata, dict) else None
|
||||
i18n, default_locale = _legacy_i18n(i18n_raw, short_key="short_desc")
|
||||
market = data.get("market") if isinstance(data.get("market"), dict) else {}
|
||||
tags = data.get("tags") if isinstance(data.get("tags"), list) else None
|
||||
return LegacyItem(
|
||||
kind="skill",
|
||||
item_id=parent.name,
|
||||
source_type="builtin",
|
||||
data=data,
|
||||
i18n=i18n,
|
||||
default_locale=default_locale,
|
||||
version=str(data["version"]) if data.get("version") is not None else None,
|
||||
category=(market.get("category") if isinstance(market.get("category"), str) else None),
|
||||
tags=tags,
|
||||
)
|
||||
|
||||
|
||||
def _json_path(error: Any) -> str:
|
||||
parts = [str(part) for part in error.absolute_path]
|
||||
return ".".join(parts) if parts else "$"
|
||||
|
||||
|
||||
def _compare(report: Report, path: str, field: str, sidecar: Any, legacy: Any) -> None:
|
||||
if legacy is not None and sidecar != legacy:
|
||||
report.add(path, "legacy-consistency", f"{field} differs from the legacy value")
|
||||
|
||||
|
||||
def _known_timestamp_value(value: Any) -> str | None:
|
||||
if isinstance(value, dict) and value.get("state") == "known" and isinstance(value.get("value"), str):
|
||||
return value["value"]
|
||||
return None
|
||||
|
||||
|
||||
def _content_is_immutable(content: Any) -> bool:
|
||||
if not isinstance(content, dict):
|
||||
return False
|
||||
kind = content.get("kind")
|
||||
ref = content.get("ref")
|
||||
digest = content.get("sha256")
|
||||
if kind == "git":
|
||||
return isinstance(ref, str) and FULL_GIT_REF_RE.fullmatch(ref) is not None
|
||||
if kind == "container":
|
||||
return isinstance(ref, str) and CONTAINER_DIGEST_RE.fullmatch(ref) is not None
|
||||
if kind in {"web", "zip", "release", "package"}:
|
||||
return isinstance(digest, str) and re.fullmatch(r"[a-fA-F0-9]{64}", digest) is not None
|
||||
return False
|
||||
|
||||
|
||||
def _validate_i18n(
|
||||
report: Report,
|
||||
rel: str,
|
||||
sidecar: dict[str, Any],
|
||||
legacy: LegacyItem,
|
||||
supported_locales: set[str],
|
||||
) -> None:
|
||||
presentation = sidecar.get("presentation")
|
||||
if not isinstance(presentation, dict):
|
||||
return
|
||||
default_locale = presentation.get("defaultLocale")
|
||||
i18n = presentation.get("i18n")
|
||||
if not isinstance(i18n, dict):
|
||||
return
|
||||
if default_locale not in i18n:
|
||||
report.add(rel, "i18n", "presentation.defaultLocale must exist in presentation.i18n")
|
||||
missing = sorted(supported_locales - set(i18n))
|
||||
if missing:
|
||||
report.add(rel, "i18n", f"presentation.i18n is missing Market locales {missing}")
|
||||
if legacy.default_locale is not None:
|
||||
_compare(report, rel, "presentation.defaultLocale", default_locale, legacy.default_locale)
|
||||
|
||||
for locale, legacy_payload in legacy.i18n.items():
|
||||
payload = i18n.get(locale)
|
||||
if not isinstance(payload, dict):
|
||||
report.add(rel, "legacy-consistency", f"presentation.i18n.{locale} is missing")
|
||||
continue
|
||||
for key in ("name", "summary", "description"):
|
||||
if key in legacy_payload:
|
||||
_compare(
|
||||
report,
|
||||
rel,
|
||||
f"presentation.i18n.{locale}.{key}",
|
||||
payload.get(key),
|
||||
legacy_payload[key],
|
||||
)
|
||||
|
||||
payloads = [
|
||||
(locale, value.get("name"), value.get("summary"))
|
||||
for locale, value in sorted(i18n.items())
|
||||
if isinstance(value, dict)
|
||||
]
|
||||
if len(payloads) > 1 and len({(name, summary) for _, name, summary in payloads}) == 1:
|
||||
report.add(
|
||||
rel,
|
||||
"i18n-freshness",
|
||||
"all locale name/summary payloads are identical; translation requires explicit review",
|
||||
severity="warning",
|
||||
)
|
||||
|
||||
|
||||
def _validate_content_consistency(
|
||||
report: Report,
|
||||
rel: str,
|
||||
sidecar: dict[str, Any],
|
||||
legacy: LegacyItem,
|
||||
) -> None:
|
||||
if legacy.source_type != "pointer":
|
||||
return
|
||||
provenance = sidecar.get("provenance")
|
||||
content = provenance.get("content") if isinstance(provenance, dict) else None
|
||||
source = legacy.data.get("source")
|
||||
if not isinstance(content, dict) or not isinstance(source, dict):
|
||||
return
|
||||
mapping = {
|
||||
"kind": "kind",
|
||||
"url": "repoUrl",
|
||||
"path": "path",
|
||||
"ref": "ref",
|
||||
"sha256": "sha256",
|
||||
}
|
||||
for sidecar_key, legacy_key in mapping.items():
|
||||
legacy_value = source.get(legacy_key)
|
||||
if legacy_value is not None:
|
||||
_compare(
|
||||
report,
|
||||
rel,
|
||||
f"provenance.content.{sidecar_key}",
|
||||
content.get(sidecar_key),
|
||||
legacy_value,
|
||||
)
|
||||
|
||||
|
||||
def _validate_governance_consistency(
|
||||
report: Report,
|
||||
rel: str,
|
||||
sidecar: dict[str, Any],
|
||||
legacy: LegacyItem,
|
||||
) -> None:
|
||||
governance = sidecar.get("governance")
|
||||
if not isinstance(governance, dict):
|
||||
return
|
||||
legacy_governance = legacy.data
|
||||
if legacy.source_type == "builtin":
|
||||
market = legacy.data.get("market")
|
||||
legacy_governance = market if isinstance(market, dict) else {}
|
||||
|
||||
for field in ("stewardship", "redistribution"):
|
||||
if legacy_governance.get(field) is not None and governance.get(field) is not None:
|
||||
_compare(report, rel, f"governance.{field}", governance.get(field), legacy_governance[field])
|
||||
|
||||
license_value = legacy.data.get("license")
|
||||
license_fact = governance.get("license")
|
||||
if license_value is not None and isinstance(license_fact, dict):
|
||||
if license_fact.get("state") != "known":
|
||||
report.add(
|
||||
rel,
|
||||
"legacy-license-unverified",
|
||||
"legacy license text has no sidecar evidence and remains unknown",
|
||||
severity="warning",
|
||||
)
|
||||
else:
|
||||
_compare(report, rel, "governance.license.value", license_fact.get("value"), license_value)
|
||||
|
||||
legacy_maintainer = legacy_governance.get("maintainer")
|
||||
maintainer_field = "upstreamMaintainer" if legacy.source_type == "pointer" else "listingMaintainer"
|
||||
sidecar_maintainer = governance.get(maintainer_field)
|
||||
if isinstance(legacy_maintainer, dict) and isinstance(sidecar_maintainer, dict):
|
||||
for field in ("name", "url", "verified"):
|
||||
if legacy_maintainer.get(field) is not None:
|
||||
_compare(
|
||||
report,
|
||||
rel,
|
||||
f"governance.{maintainer_field}.{field}",
|
||||
sidecar_maintainer.get(field),
|
||||
legacy_maintainer[field],
|
||||
)
|
||||
|
||||
|
||||
def _validate_collection(
|
||||
report: Report,
|
||||
rel: str,
|
||||
sidecar: dict[str, Any],
|
||||
legacy: LegacyItem,
|
||||
supported_locales: set[str],
|
||||
) -> None:
|
||||
legacy_children = legacy.data.get("children")
|
||||
spec = sidecar.get("spec")
|
||||
collection = spec.get("collection") if isinstance(spec, dict) else None
|
||||
sidecar_children = collection.get("children") if isinstance(collection, dict) else None
|
||||
|
||||
if isinstance(legacy_children, list) != isinstance(sidecar_children, list):
|
||||
report.add(rel, "collection-identity", "sidecar collection presence must match legacy children")
|
||||
return
|
||||
if not isinstance(legacy_children, list) or not isinstance(sidecar_children, list):
|
||||
return
|
||||
|
||||
_compare(report, rel, "spec.collection.role", collection.get("role"), "parent")
|
||||
_compare(report, rel, "spec.collection.childCount", collection.get("childCount"), len(legacy_children))
|
||||
|
||||
identities = [child.get("identity") for child in sidecar_children if isinstance(child, dict)]
|
||||
ids = [identity.get("id") for identity in identities if isinstance(identity, dict)]
|
||||
paths = [child.get("path") for child in sidecar_children if isinstance(child, dict)]
|
||||
if len(ids) != len(set(ids)):
|
||||
report.add(rel, "collection-identity", "collection child IDs must be unique")
|
||||
if len(paths) != len(set(paths)):
|
||||
report.add(rel, "collection-identity", "collection child paths must be unique")
|
||||
|
||||
if len(sidecar_children) != len(legacy_children):
|
||||
report.add(
|
||||
rel,
|
||||
"collection-identity",
|
||||
f"sidecar declares {len(sidecar_children)} children but legacy declares {len(legacy_children)}",
|
||||
)
|
||||
return
|
||||
|
||||
identical_locale_children: list[str] = []
|
||||
for index, (child, legacy_child) in enumerate(zip(sidecar_children, legacy_children)):
|
||||
if not isinstance(child, dict) or not isinstance(legacy_child, dict):
|
||||
continue
|
||||
identity = child.get("identity")
|
||||
if not isinstance(identity, dict):
|
||||
continue
|
||||
_compare(
|
||||
report,
|
||||
rel,
|
||||
f"spec.collection.children[{index}].identity.id",
|
||||
identity.get("id"),
|
||||
legacy_child.get("id"),
|
||||
)
|
||||
_compare(
|
||||
report,
|
||||
rel,
|
||||
f"spec.collection.children[{index}].identity.parentId",
|
||||
identity.get("parentId"),
|
||||
legacy.item_id,
|
||||
)
|
||||
_compare(
|
||||
report,
|
||||
rel,
|
||||
f"spec.collection.children[{index}].path",
|
||||
child.get("path"),
|
||||
legacy_child.get("path"),
|
||||
)
|
||||
presentation = child.get("presentation")
|
||||
child_i18n = presentation.get("i18n") if isinstance(presentation, dict) else None
|
||||
if isinstance(child_i18n, dict):
|
||||
missing_locales = sorted(supported_locales - set(child_i18n))
|
||||
if missing_locales:
|
||||
report.add(
|
||||
rel,
|
||||
"i18n",
|
||||
f"collection child {identity.get('id')!r} is missing Market locales {missing_locales}",
|
||||
)
|
||||
locale_payloads = [
|
||||
(payload.get("name"), payload.get("summary"))
|
||||
for payload in child_i18n.values()
|
||||
if isinstance(payload, dict)
|
||||
]
|
||||
if len(locale_payloads) > 1 and len(set(locale_payloads)) == 1:
|
||||
identical_locale_children.append(str(identity.get("id")))
|
||||
|
||||
legacy_child_i18n = legacy_child.get("i18n")
|
||||
if isinstance(legacy_child_i18n, dict):
|
||||
for locale, legacy_payload in legacy_child_i18n.items():
|
||||
payload = child_i18n.get(locale)
|
||||
if not isinstance(payload, dict) or not isinstance(legacy_payload, dict):
|
||||
report.add(
|
||||
rel,
|
||||
"legacy-consistency",
|
||||
f"collection child {identity.get('id')!r} is missing legacy locale {locale!r}",
|
||||
)
|
||||
continue
|
||||
if legacy_payload.get("shortDesc") is not None:
|
||||
_compare(
|
||||
report,
|
||||
rel,
|
||||
f"spec.collection.children[{index}].presentation.i18n.{locale}.summary",
|
||||
payload.get("summary"),
|
||||
legacy_payload.get("shortDesc"),
|
||||
)
|
||||
if legacy_child.get("name") is not None:
|
||||
_compare(
|
||||
report,
|
||||
rel,
|
||||
f"spec.collection.children[{index}].presentation.i18n.{locale}.name",
|
||||
payload.get("name"),
|
||||
legacy_child.get("name"),
|
||||
)
|
||||
legacy_version = legacy_child.get("version")
|
||||
release = child.get("release")
|
||||
if legacy_version is not None and isinstance(release, dict):
|
||||
if release.get("state") != "known":
|
||||
report.add(
|
||||
rel,
|
||||
"legacy-consistency",
|
||||
f"spec.collection.children[{index}].release must preserve legacy version",
|
||||
)
|
||||
else:
|
||||
_compare(
|
||||
report,
|
||||
rel,
|
||||
f"spec.collection.children[{index}].release.version",
|
||||
release.get("version"),
|
||||
str(legacy_version),
|
||||
)
|
||||
if legacy_version is None and isinstance(release, dict) and release.get("state") == "known":
|
||||
report.add(
|
||||
rel,
|
||||
"collection-version",
|
||||
f"child {identity.get('id')!r} has no legacy/upstream version; sidecar must not synthesize one",
|
||||
)
|
||||
|
||||
if identical_locale_children:
|
||||
preview = ", ".join(identical_locale_children[:5])
|
||||
suffix = "" if len(identical_locale_children) <= 5 else ", …"
|
||||
report.add(
|
||||
rel,
|
||||
"i18n-freshness",
|
||||
f"{len(identical_locale_children)} collection child locale payload(s) are identical "
|
||||
f"({preview}{suffix}); translation requires explicit review",
|
||||
severity="warning",
|
||||
)
|
||||
|
||||
|
||||
def validate_sidecar(
|
||||
sidecar_path: Path,
|
||||
validator: Draft7Validator,
|
||||
report: Report,
|
||||
root: Path,
|
||||
supported_locales: set[str],
|
||||
) -> None:
|
||||
rel = _rel(sidecar_path, root)
|
||||
sidecar = _read_json(sidecar_path, report, root, "catalog-schema")
|
||||
if sidecar is None:
|
||||
return
|
||||
for error in sorted(validator.iter_errors(sidecar), key=lambda item: list(item.absolute_path)):
|
||||
report.add(rel, "catalog-schema", f"{_json_path(error)}: {error.message}")
|
||||
if any(issue.path == rel and issue.rule == "catalog-schema" for issue in report.issues):
|
||||
return
|
||||
|
||||
legacy = load_legacy(sidecar_path, report, root)
|
||||
if legacy is None:
|
||||
return
|
||||
identity = sidecar["identity"]
|
||||
_compare(report, rel, "identity.kind", identity.get("kind"), legacy.kind)
|
||||
_compare(report, rel, "identity.id", identity.get("id"), legacy.item_id)
|
||||
if identity.get("parentId") is not None:
|
||||
report.add(rel, "collection-identity", "top-level Market sidecar identity must not declare parentId")
|
||||
if sidecar.get("$schema") not in (None, EXPECTED_SCHEMA_REF):
|
||||
report.add(rel, "fixed-sidecar-path", f"$schema must be {EXPECTED_SCHEMA_REF!r}")
|
||||
|
||||
presentation = sidecar.get("presentation", {})
|
||||
_compare(report, rel, "presentation.category", presentation.get("category"), legacy.category)
|
||||
if legacy.tags is not None:
|
||||
_compare(report, rel, "presentation.tags", presentation.get("tags"), legacy.tags)
|
||||
_validate_i18n(report, rel, sidecar, legacy, supported_locales)
|
||||
|
||||
release = sidecar.get("release")
|
||||
spec = sidecar.get("spec")
|
||||
system_agent = (
|
||||
legacy.kind == "agent"
|
||||
and isinstance(spec, dict)
|
||||
and spec.get("installPolicy") == "system"
|
||||
and spec.get("updatePolicy") == "repository"
|
||||
)
|
||||
if system_agent:
|
||||
if isinstance(release, dict) and release.get("state") != "unknown":
|
||||
report.add(rel, "agent-policy", "system Agent metadata revision must not be exposed as a content release")
|
||||
release_published_at = sidecar.get("timestamps", {}).get("releasePublishedAt")
|
||||
if isinstance(release_published_at, dict) and release_published_at.get("state") != "unknown":
|
||||
report.add(
|
||||
rel,
|
||||
"agent-policy",
|
||||
"system Agent metadata revision must not be exposed as a content release timestamp",
|
||||
)
|
||||
elif legacy.version is not None and isinstance(release, dict):
|
||||
if release.get("state") != "known":
|
||||
report.add(rel, "legacy-consistency", "release must preserve the legacy version")
|
||||
else:
|
||||
_compare(report, rel, "release.version", release.get("version"), legacy.version)
|
||||
|
||||
_validate_content_consistency(report, rel, sidecar, legacy)
|
||||
_validate_governance_consistency(report, rel, sidecar, legacy)
|
||||
_validate_collection(report, rel, sidecar, legacy, supported_locales)
|
||||
|
||||
governance = sidecar.get("governance")
|
||||
provenance = sidecar.get("provenance")
|
||||
timestamps = sidecar.get("timestamps")
|
||||
if isinstance(governance, dict) and governance.get("availability") == "installable":
|
||||
content = provenance.get("content") if isinstance(provenance, dict) else None
|
||||
license_fact = governance.get("license")
|
||||
reviewed_at = timestamps.get("reviewedAt") if isinstance(timestamps, dict) else None
|
||||
compliance = governance.get("compliance")
|
||||
maintainer = governance.get("listingMaintainer")
|
||||
|
||||
# Builtin Skills are already delivered by the trusted Market bootstrap.
|
||||
# Missing governance evidence must remain visible, but it must not silently
|
||||
# disable their existing installation path. A Market-installed Agent still
|
||||
# uses the strict evidence gate below; a system Agent is schema-locked to
|
||||
# listing-only and therefore never enters this installable branch.
|
||||
if legacy.source_type == "builtin":
|
||||
if not isinstance(license_fact, dict) or license_fact.get("state") != "known" or not license_fact.get("evidencePath"):
|
||||
report.add(
|
||||
rel,
|
||||
"missing-license-evidence",
|
||||
"installable builtin content has no verified per-item license evidence",
|
||||
severity="warning",
|
||||
)
|
||||
if _known_timestamp_value(reviewed_at) is None or not isinstance(compliance, dict):
|
||||
report.add(
|
||||
rel,
|
||||
"missing-review-evidence",
|
||||
"installable builtin content has no ref-bound governance review",
|
||||
severity="warning",
|
||||
)
|
||||
if not isinstance(maintainer, dict) or maintainer.get("verified") is not True:
|
||||
report.add(
|
||||
rel,
|
||||
"missing-maintainer-evidence",
|
||||
"installable builtin content has no verified listing maintainer",
|
||||
severity="warning",
|
||||
)
|
||||
if governance.get("stewardship") != "official" and not isinstance(content, dict):
|
||||
report.add(
|
||||
rel,
|
||||
"missing-content-evidence",
|
||||
"vendored builtin content has no explicit upstream content provenance",
|
||||
severity="warning",
|
||||
)
|
||||
return
|
||||
|
||||
if not _content_is_immutable(content):
|
||||
report.add(rel, "installable-evidence", "installable content must have an immutable ref or SHA-256 digest")
|
||||
if not isinstance(license_fact, dict) or license_fact.get("state") != "known":
|
||||
report.add(rel, "installable-evidence", "installable content must have a known license")
|
||||
if _known_timestamp_value(reviewed_at) is None:
|
||||
report.add(rel, "installable-evidence", "installable content must have a known reviewedAt timestamp")
|
||||
if not isinstance(compliance, dict):
|
||||
report.add(rel, "installable-evidence", "installable content must include compliance evidence")
|
||||
else:
|
||||
content_ref = None
|
||||
if isinstance(content, dict):
|
||||
content_ref = content.get("ref") or content.get("sha256")
|
||||
if content_ref is not None and compliance.get("reviewedRef") != content_ref:
|
||||
report.add(rel, "installable-evidence", "compliance.reviewedRef must match the immutable content ref/digest")
|
||||
if compliance.get("reviewedAt") != _known_timestamp_value(reviewed_at):
|
||||
report.add(rel, "installable-evidence", "compliance.reviewedAt must equal timestamps.reviewedAt")
|
||||
|
||||
|
||||
def _discover_sidecars(root: Path) -> list[Path]:
|
||||
return sorted(path for path in root.rglob(SIDECAR_NAME) if ".git" not in path.parts)
|
||||
|
||||
|
||||
def _load_supported_locales(root: Path, report: Report) -> set[str]:
|
||||
manifest = _read_json(root / "manifest.json", report, root, "market-stats")
|
||||
if manifest is None:
|
||||
return set()
|
||||
locales = manifest.get("supportedLocales")
|
||||
if not isinstance(locales, list) or not all(isinstance(locale, str) for locale in locales):
|
||||
report.add("manifest.json", "market-stats", "supportedLocales must be a list of strings")
|
||||
return set()
|
||||
return set(locales)
|
||||
|
||||
|
||||
def _validate_stats(root: Path, report: Report, sidecars: Iterable[Path]) -> None:
|
||||
agent_count = sum(1 for _ in (root / "agents").glob("*/agent.json")) + sum(
|
||||
1 for _ in (root / "agents").glob("*/entry.json")
|
||||
)
|
||||
builtin_count = sum(1 for _ in (root / "skills").glob("*/SKILL.md"))
|
||||
pointer_count = sum(1 for _ in (root / "skills").glob("*/entry.json"))
|
||||
collection_count = 0
|
||||
child_count = 0
|
||||
for entry_path in (root / "skills").glob("*/entry.json"):
|
||||
try:
|
||||
entry = json.loads(entry_path.read_text(encoding="utf-8"))
|
||||
except (OSError, json.JSONDecodeError):
|
||||
continue
|
||||
children = entry.get("children") if isinstance(entry, dict) else None
|
||||
if isinstance(children, list):
|
||||
collection_count += 1
|
||||
child_count += len(children)
|
||||
sidecar_list = list(sidecars)
|
||||
report.stats = {
|
||||
"agents": agent_count,
|
||||
"builtinSkills": builtin_count,
|
||||
"pointerSkills": pointer_count,
|
||||
"publishableSkills": builtin_count + pointer_count,
|
||||
"collections": collection_count,
|
||||
"collectionChildren": child_count,
|
||||
"sidecars": len(sidecar_list),
|
||||
}
|
||||
|
||||
manifest = _read_json(root / "manifest.json", report, root, "market-stats")
|
||||
stats = manifest.get("stats") if isinstance(manifest, dict) else None
|
||||
if not isinstance(stats, dict):
|
||||
report.add("manifest.json", "market-stats", "stats must be an object")
|
||||
return
|
||||
if stats.get("totalAgents") != agent_count:
|
||||
report.add("manifest.json", "market-stats", f"stats.totalAgents must be {agent_count}")
|
||||
if stats.get("totalSkills") != builtin_count + pointer_count:
|
||||
report.add("manifest.json", "market-stats", f"stats.totalSkills must be {builtin_count + pointer_count}")
|
||||
|
||||
|
||||
def validate_repository(root: Path = REPO_ROOT, *, require_complete: bool = False) -> Report:
|
||||
root = root.resolve()
|
||||
report = Report()
|
||||
schema_path = root / "schemas" / "catalog-metadata.v1.schema.json"
|
||||
schema = _read_json(schema_path, report, root, "catalog-schema")
|
||||
if schema is None:
|
||||
return report
|
||||
try:
|
||||
Draft7Validator.check_schema(schema)
|
||||
except Exception as exc: # jsonschema raises several SchemaError subclasses
|
||||
report.add(_rel(schema_path, root), "catalog-schema", f"invalid JSON Schema: {exc}")
|
||||
return report
|
||||
validator = Draft7Validator(schema, format_checker=FormatChecker())
|
||||
sidecars = _discover_sidecars(root)
|
||||
supported_locales = _load_supported_locales(root, report)
|
||||
|
||||
for sidecar in sidecars:
|
||||
if sidecar.parent.parent not in {root / "agents", root / "skills"}:
|
||||
report.add(
|
||||
_rel(sidecar, root),
|
||||
"fixed-sidecar-path",
|
||||
f"{SIDECAR_NAME} is only allowed at agents/<id>/ or skills/<id>/",
|
||||
)
|
||||
continue
|
||||
validate_sidecar(sidecar, validator, report, root, supported_locales)
|
||||
|
||||
_validate_stats(root, report, sidecars)
|
||||
if require_complete:
|
||||
expected = report.stats.get("agents", 0) + report.stats.get("publishableSkills", 0)
|
||||
actual = report.stats.get("sidecars", 0)
|
||||
if actual != expected:
|
||||
report.add(
|
||||
".",
|
||||
"sidecar-coverage",
|
||||
f"complete migration requires {expected} sidecars, found {actual}",
|
||||
)
|
||||
return report
|
||||
|
||||
|
||||
def main(argv: list[str] | None = None) -> int:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("--json", action="store_true", help="emit stable machine-readable JSON")
|
||||
parser.add_argument(
|
||||
"--require-complete",
|
||||
action="store_true",
|
||||
help="require one sidecar for every publishable top-level Agent and Skill",
|
||||
)
|
||||
args = parser.parse_args(argv)
|
||||
report = validate_repository(require_complete=args.require_complete)
|
||||
if args.json:
|
||||
json.dump(
|
||||
{"stats": report.stats, "issues": [issue.to_dict() for issue in report.issues]},
|
||||
sys.stdout,
|
||||
indent=2,
|
||||
ensure_ascii=False,
|
||||
sort_keys=True,
|
||||
)
|
||||
sys.stdout.write("\n")
|
||||
else:
|
||||
for issue in report.issues:
|
||||
marker = "ERROR" if issue.severity == "error" else "WARN "
|
||||
print(f"[{marker}] {issue.path} :: {issue.rule} :: {issue.message}")
|
||||
summary = ", ".join(f"{key}={value}" for key, value in report.stats.items())
|
||||
if report.issues:
|
||||
errors = sum(issue.severity == "error" for issue in report.issues)
|
||||
warnings = sum(issue.severity == "warning" for issue in report.issues)
|
||||
print(f"\n{errors} error(s), {warnings} warning(s). {summary}")
|
||||
else:
|
||||
print(f"OK: catalog metadata valid. {summary}")
|
||||
return 1 if report.has_errors else 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
sys.exit(main())
|
||||
except KeyboardInterrupt:
|
||||
sys.exit(130)
|
||||
Reference in New Issue
Block a user