fix: 对齐 Agent 指针目录与客户端校验契约 (#103)

## 问题 / Problem

市场 CI 要求每个 Agent 条目都有 catalog sidecar,但 sidecar 校验器只接受相邻的
agent.json。客户端已支持的 Agent entry.json 指针因此无法满足完整覆盖门禁。

The catalog completeness gate requires sidecars for every Agent, while
the validator previously read only agent.json. Valid Agent pointer
entries could not satisfy CI.

## 修改 / Changes

- Agent 主文件严格二选一:内联 agent.json 或指针 entry.json;目录 slug 与安装源 UUID 保持分离。
- 使用从主仓固定提交导出的完整客户端 Agent entry Schema,先校验原始类型、字段、版本、路径和策略组合,再比较
sidecar。
- 对齐 latestVersion、固定来源、许可/治理、兼容性和有效的缺省 market/market 策略,拒绝 sidecar
单方面提升托管权限。
- 未修改产品条目、manifest、现有 sidecar Schema 或 CI 完整性要求。

Agent pointers now pass the pinned client input schema before source
metadata consistency checks. Effective default policies are compared
without promoting catalog metadata into authority. Existing inline
Agents and Skills retain their validation paths; no entries or CI
requirements change.

## 验证 / Validation

- 29 + 4 + 9 项定向 Python 测试通过。
- 完整 i18n、catalog --require-complete、translation freshness 检查通过;116
条既有警告与基线一致,无新增。
- 原始客户端 Schema 与固定主仓来源逐项比较一致(仅增加溯源注释)。
- 独立代码审查发现的原始契约与缺省策略问题已修复并复核。
- 公开信息边界检查覆盖完整工作树、隐藏文件、链接目标与新增 Git 元数据,通过。

Targeted tests, full catalog/i18n/freshness validation, source-schema
comparison and independent review passed. Existing warnings are
unchanged. This PR repairs catalog validation; it does not claim runtime
installation of a new product.

---------

Co-authored-by: yige <yige@yigedeMacBook-Neo.local>
This commit is contained in:
2026-08-31 06:21:04 -04:00
committed by GitHub
parent a205ef20f0
commit d76db984b6
4 changed files with 568 additions and 8 deletions

View File

@@ -99,6 +99,7 @@ class CatalogMetadataValidatorTests(unittest.TestCase):
self.root = Path(self.tempdir.name)
(self.root / "schemas").mkdir()
shutil.copyfile(SOURCE_SCHEMA, self.root / "schemas" / SOURCE_SCHEMA.name)
shutil.copyfile(SOURCE_SCHEMA.with_name(VALIDATOR.AGENT_ENTRY_SCHEMA_NAME), self.root / "schemas" / VALIDATOR.AGENT_ENTRY_SCHEMA_NAME)
(self.root / "agents").mkdir()
(self.root / "skills" / "example-skill").mkdir(parents=True)
self.write_json(
@@ -174,6 +175,201 @@ class CatalogMetadataValidatorTests(unittest.TestCase):
manifest["stats"]["totalAgents"] = 1
self.write_json(self.root / "manifest.json", manifest)
def write_agent_pointer_case(self, mutate_entry=None, mutate_sidecar=None) -> tuple[Path, Path]:
agent_dir = self.root / "agents" / "example-agent"
entry = valid_entry()
entry.update(
id="example-agent", name="Example Agent", latestVersion="1.2.3",
requiredClientVersion="10.0.0", installPolicy="market", updatePolicy="market",
icon='<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 24 24"><path d="M0 0h24v24H0z"/></svg>',
)
entry["i18n"] = {
"zh-CN": {"name": "示例智能体", "shortDesc": "通用示例智能体"},
"en-US": {"name": "Example Agent", "shortDesc": "General-purpose example agent"},
}
entry["source"]["repoUrl"] = "https://example.com/example-agent.git"
sidecar = valid_sidecar()
sidecar["identity"] = {"kind": "agent", "id": "example-agent"}
sidecar["presentation"]["i18n"] = {
locale: {"name": value["name"], "summary": value["shortDesc"]}
for locale, value in entry["i18n"].items()
}
sidecar["release"] = {"state": "known", "version": "1.2.3", "versionScheme": "semver"}
sidecar["provenance"]["content"]["url"] = entry["source"]["repoUrl"]
sidecar["timestamps"]["reviewedAt"] = {"state": "known", "value": "2026-08-30", "precision": "day"}
sidecar["governance"].update(
availability="installable",
compliance={"licenseEvidencePath": "LICENSE", "reviewedRef": "a" * 40,
"reviewedAt": "2026-08-30", "reviewedBy": "Example Reviewer", "upstreamEndorsed": False},
)
sidecar["governance"]["license"]["evidencePath"] = "LICENSE"
sidecar["compatibility"]["requiredClientVersion"] = "10.0.0"
sidecar["spec"] = {"kind": "agent", "installPolicy": "market", "updatePolicy": "market"}
if mutate_entry:
mutate_entry(entry)
if mutate_sidecar:
mutate_sidecar(sidecar)
entry_path = agent_dir / "entry.json"
sidecar_path = agent_dir / VALIDATOR.SIDECAR_NAME
self.write_json(entry_path, entry)
self.write_json(sidecar_path, sidecar)
self.write_json(self.root / "manifest.json", {
"supportedLocales": ["zh-CN", "en-US"], "stats": {"totalAgents": 1, "totalSkills": 1},
})
return entry_path, sidecar_path
def test_accepts_complete_installable_agent_pointer_without_inline_agent(self) -> None:
entry_path, _ = self.write_agent_pointer_case()
report = self.validate(require_complete=True)
self.assertFalse(report.has_errors, report.issues)
self.assertFalse(entry_path.with_name("agent.json").exists())
self.assertEqual(1, report.stats["agents"])
self.assertEqual(2, report.stats["sidecars"])
def test_listing_only_agent_pointer_allows_empty_root_path_and_unpinned_ref(self) -> None:
def metadata_change(payload):
payload["provenance"]["content"].pop("ref")
payload["governance"]["availability"] = "listing-only"
self.write_agent_pointer_case(
lambda payload: payload["source"].update(path="", ref=""), metadata_change,
)
report = self.validate(require_complete=True)
self.assertFalse(report.has_errors, report.issues)
def test_accepts_agent_web_and_zip_pointers_with_matching_byte_digest(self) -> None:
for kind in ("web", "zip"):
with self.subTest(kind=kind):
def entry_change(payload):
payload["source"] = {"kind": kind, "repoUrl": "https://example.com/agent.zip", "sha256": "b" * 64}
def metadata_change(payload):
payload["provenance"]["content"] = {"kind": kind, "url": "https://example.com/agent.zip", "sha256": "b" * 64}
payload["governance"]["compliance"]["reviewedRef"] = "b" * 64
self.write_agent_pointer_case(entry_change, metadata_change)
report = self.validate(require_complete=True)
self.assertFalse(report.has_errors, report.issues)
def test_rejects_agent_pointer_sidecar_conflicts(self) -> None:
mutations = {
"identity": lambda p: p["identity"].update(id="other-agent"),
"kind": lambda p: (p["identity"].update(kind="skill"), p.update(spec={"kind": "skill"})),
"version": lambda p: p["release"].update(version="9.9.9"),
"source-kind": lambda p: p["provenance"]["content"].update(kind="web"),
"source-url": lambda p: p["provenance"]["content"].update(url="https://example.com/other.git"),
"source-ref": lambda p: p["provenance"]["content"].update(ref="b" * 40),
"source-path": lambda p: p["provenance"]["content"].update(path="another-agent"),
"source-digest": lambda p: p["provenance"]["content"].update(sha256="b" * 64),
"category": lambda p: p["presentation"].update(category="research"),
"tags": lambda p: p["presentation"].update(tags=["other"]),
"summary": lambda p: p["presentation"]["i18n"]["en-US"].update(summary="Different"),
"license": lambda p: p["governance"]["license"].update(value="Apache-2.0"),
"redistribution": lambda p: p["governance"].update(redistribution="source-pointer-only"),
"stewardship": lambda p: p["governance"].update(stewardship="official"),
"maintainer": lambda p: p["governance"]["upstreamMaintainer"].update(name="Other Maintainer"),
"compatibility": lambda p: p["compatibility"].update(requiredClientVersion="9.0.0"),
"policy": lambda p: (p.update(spec={"kind": "agent", "installPolicy": "system", "updatePolicy": "repository"}),
p.update(release=unknown()), p["governance"].update(availability="listing-only")),
}
for name, mutation in mutations.items():
with self.subTest(name=name):
self.write_agent_pointer_case(mutate_sidecar=mutation)
self.assertTrue(any(issue.rule == "legacy-consistency" for issue in self.validate().issues))
def test_rejects_agent_pointer_entry_id_that_is_not_the_catalog_slug(self) -> None:
self.write_agent_pointer_case(mutate_entry=lambda p: p.update(id="00000000-0000-4000-8000-000000000000"))
self.assertTrue(any("entry.id" in issue.message for issue in self.validate().issues))
def test_agent_pointer_must_keep_both_immutable_source_and_governance_evidence(self) -> None:
cases = [
(lambda p: p["source"].pop("ref"), None),
(lambda p: p["source"].update(ref="main"),
lambda p: (p["provenance"]["content"].update(ref="main"), p["governance"]["compliance"].update(reviewedRef="main"))),
(None, lambda p: p["governance"].update(license=unknown())),
(None, lambda p: p["governance"].pop("compliance")),
(None, lambda p: p["timestamps"].update(reviewedAt=unknown())),
(None, lambda p: p["governance"]["compliance"].update(reviewedRef="b" * 40)),
]
for index, (entry_change, metadata_change) in enumerate(cases):
with self.subTest(index=index):
self.write_agent_pointer_case(entry_change, metadata_change)
self.assertTrue(any(issue.rule == "installable-evidence" for issue in self.validate().issues))
def test_rejects_invalid_raw_agent_entry_even_if_sidecar_repeats_it(self) -> None:
cases = [
(lambda p: p.update(latestVersion="1.2.3-beta"), lambda p: p["release"].update(version="1.2.3-beta")),
(lambda p: p.update(latestVersion=123), lambda p: p["release"].update(version="123", versionScheme="opaque")),
(lambda p: p.update(latestVersion="١.٢.٣"), lambda p: p["release"].update(version="١.٢.٣", versionScheme="opaque")),
(lambda p: p.update(requiredClientVersion="1.2.3-beta"), lambda p: p["compatibility"].update(requiredClientVersion="1.2.3-beta")),
(lambda p: p.update(requiredClientVersion=None), None),
(lambda p: p.pop("updatePolicy"), None),
(lambda p: p.pop("installPolicy"), None),
(lambda p: p.update(installPolicy="system", updatePolicy="market"), None),
(lambda p: p["source"].update(repoBranch=123), None),
(lambda p: p["source"].update(path="C:/outside"), lambda p: p["provenance"]["content"].update(path="C:/outside")),
(lambda p: p["source"].update(path="C:\\outside"), None),
(lambda p: p["source"].update(path="../outside"), None),
(lambda p: p["source"].update(path=0), None),
(lambda p: p["source"].update(ref=None), None),
(lambda p: p["source"].update(unknown=True), None),
(lambda p: p.update(version="1.2.3"), None),
(lambda p: p.update(name=42), None),
(lambda p: p.update(i18n={"en-US": {"name": 42}}), None),
(lambda p: p["maintainer"].update(verified="true"), None),
]
for index, (entry_change, metadata_change) in enumerate(cases):
with self.subTest(index=index):
self.write_agent_pointer_case(entry_change, metadata_change)
self.assertTrue(any(issue.rule == "agent-entry-schema" for issue in self.validate().issues))
def test_agent_pointer_default_policy_cannot_be_replaced_by_sidecar(self) -> None:
def default_policy(payload):
payload.pop("installPolicy")
payload.pop("updatePolicy")
self.write_agent_pointer_case(default_policy)
self.assertFalse(self.validate().has_errors)
self.write_agent_pointer_case(default_policy, lambda p: p.update(spec={"kind": "agent"}))
self.assertFalse(self.validate().has_errors)
def system_policy(payload):
payload["spec"].update(installPolicy="system", updatePolicy="repository")
payload["governance"]["availability"] = "listing-only"
payload["release"] = unknown()
self.write_agent_pointer_case(default_policy, system_policy)
self.assertTrue(any(issue.rule == "legacy-consistency" for issue in self.validate().issues))
def test_accepts_explicit_system_agent_pointer_with_matching_listing_policy(self) -> None:
def system_policy(payload):
payload["spec"].update(installPolicy="system", updatePolicy="repository")
payload["governance"]["availability"] = "listing-only"
payload["release"] = unknown()
self.write_agent_pointer_case(lambda p: p.update(installPolicy="system", updatePolicy="repository"), system_policy)
report = self.validate(require_complete=True)
self.assertFalse(report.has_errors, report.issues)
def test_agent_pointer_sidecar_does_not_allow_provider_or_runtime_fields(self) -> None:
self.write_agent_pointer_case(mutate_sidecar=lambda p: p["identity"].update(catalogSourceId="market:official"))
self.assertTrue(any(issue.rule == "catalog-schema" for issue in self.validate().issues))
def test_agent_sidecar_rejects_missing_or_ambiguous_legacy_file(self) -> None:
entry, sidecar = self.write_agent_pointer_case()
entry.unlink()
report = self.validate()
self.assertTrue(any(issue.rule == "fixed-sidecar-path" and "agent.json or entry.json" in issue.message for issue in report.issues))
self.write_agent_pointer_case()
self.write_json(entry.with_name("agent.json"), {"id": "example-agent"})
report = self.validate()
self.assertTrue(any(issue.rule == "fixed-sidecar-path" and "exactly one" in issue.message for issue in report.issues))
def test_agent_pointer_requires_sidecar_under_complete_coverage(self) -> None:
_, sidecar = self.write_agent_pointer_case()
sidecar.unlink()
self.assertTrue(any(issue.rule == "sidecar-coverage" for issue in self.validate(require_complete=True).issues))
def test_accepts_valid_listing_only_pointer(self) -> None:
report = self.validate(require_complete=True)
self.assertFalse(report.has_errors, report.issues)

View File

@@ -22,11 +22,12 @@ from pathlib import Path
from typing import Any, Iterable
import yaml
from jsonschema import Draft7Validator, FormatChecker
from jsonschema import Draft7Validator, FormatChecker, ValidationError, validators
REPO_ROOT = Path(__file__).resolve().parents[2]
SCHEMA_PATH = REPO_ROOT / "schemas" / "catalog-metadata.v1.schema.json"
AGENT_ENTRY_SCHEMA_NAME = "market-agent-entry.client.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)
@@ -34,6 +35,16 @@ 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}$")
def _client_pattern(validator, pattern, instance, schema):
# The exported client patterns use ECMAScript ASCII \d; Python's default
# Unicode \d would admit version strings that the client rejects.
if validator.is_type(instance, "string") and re.search(pattern, instance, re.ASCII) is None:
yield ValidationError(f"{instance!r} does not match {pattern!r}")
AgentEntryValidator = validators.extend(Draft7Validator, {"pattern": _client_pattern})
@dataclass(frozen=True)
class Issue:
path: str
@@ -146,23 +157,46 @@ def _legacy_i18n(raw: Any, *, short_key: str) -> tuple[dict[str, dict[str, Any]]
return locales, default_locale if isinstance(default_locale, str) else None
def load_legacy(sidecar: Path, report: Report, root: Path) -> LegacyItem | None:
def load_legacy(sidecar: Path, report: Report, root: Path, agent_entry_validator) -> LegacyItem | None:
parent = sidecar.parent
root_kind = parent.parent.name
if root_kind == "agents":
legacy_path = parent / "agent.json"
inline_path = parent / "agent.json"
pointer_path = parent / "entry.json"
if inline_path.is_file() == pointer_path.is_file():
report.add(
_rel(sidecar, root),
"fixed-sidecar-path",
"agent sidecar must be next to exactly one legacy agent.json or entry.json",
)
return None
is_pointer = pointer_path.is_file()
legacy_path = pointer_path if is_pointer else inline_path
data = _read_json(legacy_path, report, root, "legacy-read")
if data is None:
return None
if is_pointer:
errors = list(agent_entry_validator.iter_errors(data))
for error in errors:
report.add(_rel(legacy_path, root), "agent-entry-schema", f"{_json_path(error)}: {error.message}")
if errors:
return None
if is_pointer and data.get("id") != parent.name:
report.add(_rel(legacy_path, root), "legacy-consistency", "entry.id must equal the catalog directory slug")
# The pointer ID is a listing slug, not the upstream AgentFS UUID.
# Client indexing reads latestVersion for pointers and version for inline metadata.
version = data.get("latestVersion") if is_pointer else (
str(data["version"]) if data.get("version") is not None else None
)
i18n, default_locale = _legacy_i18n(data.get("i18n"), short_key="shortDesc")
return LegacyItem(
kind="agent",
item_id=parent.name,
source_type="agent",
source_type="pointer" if is_pointer else "agent",
data=data,
i18n=i18n,
default_locale=default_locale,
version=str(data["version"]) if data.get("version") is not None else None,
version=version,
category=data.get("category") if isinstance(data.get("category"), str) else None,
tags=data.get("tags") if isinstance(data.get("tags"), list) else None,
)
@@ -327,7 +361,16 @@ def _validate_content_consistency(
}
for sidecar_key, legacy_key in mapping.items():
legacy_value = source.get(legacy_key)
if legacy_value is not None:
if legacy.kind == "agent":
declared_value = content.get(sidecar_key)
# A sidecar must describe the exact pointer that the client installs,
# not add a different subdirectory or pin that entry.json never uses.
if sidecar_key in {"path", "ref"}:
declared_value = None if declared_value == "" else declared_value
legacy_value = None if legacy_value == "" else legacy_value
if declared_value != legacy_value:
report.add(rel, "legacy-consistency", f"provenance.content.{sidecar_key} differs from the Agent pointer")
elif legacy_value is not None:
_compare(
report,
rel,
@@ -535,6 +578,7 @@ def validate_sidecar(
report: Report,
root: Path,
supported_locales: set[str],
agent_entry_validator,
) -> None:
rel = _rel(sidecar_path, root)
sidecar = _read_json(sidecar_path, report, root, "catalog-schema")
@@ -545,7 +589,7 @@ def validate_sidecar(
if any(issue.path == rel and issue.rule == "catalog-schema" for issue in report.issues):
return
legacy = load_legacy(sidecar_path, report, root)
legacy = load_legacy(sidecar_path, report, root, agent_entry_validator)
if legacy is None:
return
identity = sidecar["identity"]
@@ -586,6 +630,16 @@ def validate_sidecar(
else:
_compare(report, rel, "release.version", release.get("version"), legacy.version)
if legacy.kind == "agent" and legacy.source_type == "pointer":
for field in ("installPolicy", "updatePolicy"):
# Ordinary pointers default to market/market in the client. A sidecar
# cannot turn an omitted pair into a system/repository listing.
_compare(report, rel, f"spec.{field}", spec.get(field, "market"), legacy.data.get(field, "market"))
_compare(
report, rel, "compatibility.requiredClientVersion",
sidecar["compatibility"].get("requiredClientVersion"), legacy.data.get("requiredClientVersion"),
)
_validate_content_consistency(report, rel, sidecar, legacy)
_validate_governance_consistency(report, rel, sidecar, legacy)
_validate_collection(report, rel, sidecar, legacy, supported_locales)
@@ -636,6 +690,13 @@ def validate_sidecar(
)
return
if legacy.kind == "agent" and legacy.source_type == "pointer":
if not _content_is_immutable(legacy.data.get("source")):
report.add(
rel, "installable-evidence",
"installable Agent entry.source must itself declare an immutable ref or SHA-256 digest",
)
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":
@@ -720,6 +781,15 @@ def validate_repository(root: Path = REPO_ROOT, *, require_complete: bool = Fals
except Exception as exc: # jsonschema raises several SchemaError subclasses
report.add(_rel(schema_path, root), "catalog-schema", f"invalid JSON Schema: {exc}")
return report
agent_entry_schema = _read_json(root / "schemas" / AGENT_ENTRY_SCHEMA_NAME, report, root, "agent-entry-schema")
if agent_entry_schema is None:
return report
try:
AgentEntryValidator.check_schema(agent_entry_schema)
except Exception as exc:
report.add(f"schemas/{AGENT_ENTRY_SCHEMA_NAME}", "agent-entry-schema", f"invalid JSON Schema: {exc}")
return report
agent_entry_validator = AgentEntryValidator(agent_entry_schema, format_checker=FormatChecker())
validator = Draft7Validator(schema, format_checker=FormatChecker())
sidecars = _discover_sidecars(root)
supported_locales = _load_supported_locales(root, report)
@@ -732,7 +802,7 @@ def validate_repository(root: Path = REPO_ROOT, *, require_complete: bool = Fals
f"{SIDECAR_NAME} is only allowed at agents/<id>/ or skills/<id>/",
)
continue
validate_sidecar(sidecar, validator, report, root, supported_locales)
validate_sidecar(sidecar, validator, report, root, supported_locales, agent_entry_validator)
_validate_stats(root, report, sidecars)
if require_complete: