root/Jyotisha
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(governance): pin selective upstream import contract

Browse Source
This commit is contained in:
Jesse_Chen
2026-08-15 12:13:09 +08:00
parent 0fd111d16b
commit 8abd248f49
11 changed files with 1053 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_import_yinduzhanxing.py
+224 -15
View File
@@ -16,6 +16,7 @@ if str(SCRIPTS) not in sys.path:
import import_yinduzhanxing as importer # noqa: E402
POLICY = ROOT / "references/cross_project_contract/sync_policy.v2.json"
ARCHIVE_COMMIT = "5db72537741fcedaa7b5498502d4a31b0f9fc147"
def _git(root: Path, *args: str) -> str:
@@ -47,10 +48,31 @@ def _source(tmp_path: Path) -> Path:
return source
def _policy(tmp_path: Path, *, mirror: list[dict[str, str]] | None = None) -> Path:
def _git_source(tmp_path: Path) -> Path:
source = _source(tmp_path)
subprocess.run(["git", "init", "-q", str(source)], check=True)
subprocess.run(["git", "-C", str(source), "config", "user.email", "tests@example.invalid"], check=True)
subprocess.run(["git", "-C", str(source), "config", "user.name", "Tests"], check=True)
subprocess.run(
["git", "-C", str(source), "remote", "add", "origin", "https://github.com/732642856/yinduzhanxing.git"],
check=True,
)
subprocess.run(["git", "-C", str(source), "add", "."], check=True)
subprocess.run(["git", "-C", str(source), "commit", "-qm", "research base"], check=True)
return source
def _policy(
tmp_path: Path,
*,
mirror: list[dict[str, str]] | None = None,
protected: list[str] | None = None,
) -> Path:
value = json.loads(POLICY.read_text(encoding="utf-8"))
if mirror is not None:
value["modes"]["mirror"] = mirror
if protected is not None:
value["modes"]["protected"] = protected
path = tmp_path / "policy.json"
path.write_text(json.dumps(value), encoding="utf-8")
return path
@@ -60,37 +82,186 @@ def test_v2_policy_is_strictly_one_way_and_has_no_reverse_gate() -> None:
policy = importer.load_policy(POLICY)
assert policy["direction"] == "research_to_commercial_only"
assert policy["reverse_sync"] == "forbidden"
assert policy["source_repository_url"] == "https://github.com/732642856/yinduzhanxing"
assert "commercial_to_research" not in json.dumps(policy, sort_keys=True)
assert policy["modes"]["protected"]
def test_dry_run_writes_no_target_file_and_snapshot_commit_is_unknown(tmp_path: Path) -> None:
def test_policy_cannot_remove_mandatory_commercial_protections(tmp_path: Path) -> None:
weakened = sorted(importer.REQUIRED_PROTECTED_PATTERNS - {"frontend/**"})
policy = _policy(tmp_path, protected=weakened)
with pytest.raises(importer.ImportRejected, match="required_protected_patterns_missing:frontend/\\*\\*"):
importer.load_policy(policy)
def _archive_kwargs(source: Path) -> dict[str, str]:
return {
"source_mode": "archive",
"source_commit": ARCHIVE_COMMIT,
"expected_tree_sha256": importer.source_tree_hash(source),
}
def test_dry_run_writes_no_target_file_and_records_bound_archive_identity(tmp_path: Path) -> None:
source, target = _source(tmp_path), _target(tmp_path)
before = _git(target, "status", "--porcelain=v1")
manifest = importer.build_manifest(source, policy_path=POLICY, target=target)
manifest = importer.build_manifest(source, policy_path=POLICY, target=target, **_archive_kwargs(source))
after = _git(target, "status", "--porcelain=v1")
assert before == after == ""
assert manifest["source_mode"] == "snapshot"
assert manifest["source_commit"] == "unknown"
assert manifest["source_mode"] == "archive"
assert manifest["source_commit"] == ARCHIVE_COMMIT
assert manifest["source_tree_hash"] == importer.source_tree_hash(source)
assert manifest["mirror_files"][0]["status"] == "new"
assert not (target / "references/upstream/yinduzhanxing/SKILL.md").exists()
@pytest.mark.parametrize(
("source_commit", "expected_tree_sha256", "error"),
[
(ARCHIVE_COMMIT, None, "archive_expected_tree_sha256_required"),
(None, "0" * 64, "archive_source_commit_required"),
(ARCHIVE_COMMIT, "0" * 64, "source_tree_sha256_mismatch"),
],
)
def test_archive_identity_requires_commit_and_matching_tree_hash(
tmp_path: Path, source_commit: str | None, expected_tree_sha256: str | None, error: str
) -> None:
source, target = _source(tmp_path), _target(tmp_path)
with pytest.raises(importer.ImportRejected, match=error):
importer.build_manifest(
source,
policy_path=POLICY,
target=target,
source_mode="archive",
source_commit=source_commit,
expected_tree_sha256=expected_tree_sha256,
)
@pytest.mark.parametrize(
("source_commit", "expected_tree_sha256"),
[(ARCHIVE_COMMIT, None), (None, "0" * 64)],
)
def test_explicit_provenance_pins_must_be_provided_together_for_git_sources(
tmp_path: Path, source_commit: str | None, expected_tree_sha256: str | None
) -> None:
source, target = _git_source(tmp_path), _target(tmp_path)
with pytest.raises(importer.ImportRejected, match="source_commit_and_expected_tree_sha256_must_be_provided_together"):
importer.build_manifest(
source,
policy_path=POLICY,
target=target,
source_mode="git",
source_commit=source_commit,
expected_tree_sha256=expected_tree_sha256,
)
def test_unbound_non_git_source_no_longer_emits_unknown_snapshot(tmp_path: Path) -> None:
source, target = _source(tmp_path), _target(tmp_path)
with pytest.raises(importer.ImportRejected, match="archive_source_commit_required"):
importer.build_manifest(source, policy_path=POLICY, target=target)
def test_git_source_must_be_clean_and_is_bound_to_head(tmp_path: Path) -> None:
source, target = _git_source(tmp_path), _target(tmp_path)
manifest = importer.build_manifest(source, policy_path=POLICY, target=target)
assert manifest["source_mode"] == "git"
assert manifest["source_commit"] == _git(source, "rev-parse", "HEAD")
(source / "SKILL.md").write_text("# Dirty tracked research skill\n", encoding="utf-8")
with pytest.raises(importer.ImportRejected, match="git_source_must_be_clean"):
importer.build_manifest(source, policy_path=POLICY, target=target)
subprocess.run(["git", "-C", str(source), "restore", "SKILL.md"], check=True)
(source / "untracked.txt").write_text("not committed\n", encoding="utf-8")
with pytest.raises(importer.ImportRejected, match="git_source_must_be_clean"):
importer.build_manifest(source, policy_path=POLICY, target=target)
def test_git_source_cannot_be_forced_through_archive_mode(tmp_path: Path) -> None:
source, target = _git_source(tmp_path), _target(tmp_path)
with pytest.raises(importer.ImportRejected, match="archive_source_must_not_be_git_checkout"):
importer.build_manifest(source, policy_path=POLICY, target=target, **_archive_kwargs(source))
def test_git_source_origin_must_match_policy_repository(tmp_path: Path) -> None:
source, target = _git_source(tmp_path), _target(tmp_path)
subprocess.run(
["git", "-C", str(source), "remote", "set-url", "origin", "https://github.com/example/not-the-upstream.git"],
check=True,
)
with pytest.raises(importer.ImportRejected, match="git_source_repository_mismatch"):
importer.build_manifest(source, policy_path=POLICY, target=target)
subprocess.run(
["git", "-C", str(source), "remote", "set-url", "origin", "https://evil.example/732642856/yinduzhanxing.git"],
check=True,
)
with pytest.raises(importer.ImportRejected, match="git_source_repository_mismatch"):
importer.build_manifest(source, policy_path=POLICY, target=target)
subprocess.run(
["git", "-C", str(source), "remote", "set-url", "origin", "https://github.com/attacker/732642856/yinduzhanxing.git"],
check=True,
)
with pytest.raises(importer.ImportRejected, match="git_source_repository_mismatch"):
importer.build_manifest(source, policy_path=POLICY, target=target)
def test_manifest_output_cannot_overwrite_commercial_repository_files(tmp_path: Path) -> None:
protected = ROOT / "SKILL.md"
before = protected.read_bytes()
with pytest.raises(importer.ImportRejected, match="manifest_output_must_be_external_or_import_record_json"):
importer.validate_output_path(protected)
assert protected.read_bytes() == before
external = tmp_path / "manifest.json"
assert importer.validate_output_path(external) == external.resolve()
internal = ROOT / "references/cross_project_contract/imports/test-output.json"
assert importer.validate_output_path(internal) == internal.resolve()
source, target = _source(tmp_path), _target(tmp_path)
with pytest.raises(importer.ImportRejected, match="manifest_output_must_not_overlap_source"):
importer.validate_output_path(source / "manifest.json", target=target, source=source)
with pytest.raises(importer.ImportRejected, match="manifest_output_must_be_json"):
importer.validate_output_path(tmp_path / "manifest.txt", target=target, source=source)
def test_fixed_archive_inputs_build_byte_equivalent_manifests(tmp_path: Path) -> None:
source, target = _source(tmp_path), _target(tmp_path)
kwargs = _archive_kwargs(source)
recorded_tests = ["tests/test_upstream_import_plan.py", "tests/test_import_yinduzhanxing.py", "tests/test_import_yinduzhanxing.py"]
first = importer.build_manifest(source, policy_path=POLICY, target=target, tests_run=recorded_tests, **kwargs)
second = importer.build_manifest(source, policy_path=POLICY, target=target, tests_run=list(reversed(recorded_tests)), **kwargs)
assert first == second
assert json.dumps(first, ensure_ascii=False, sort_keys=True) == json.dumps(second, ensure_ascii=False, sort_keys=True)
assert first["source_skill_sha256"] == hashlib.sha256((source / "SKILL.md").read_bytes()).hexdigest()
assert first["tests_run"] == ["tests/test_import_yinduzhanxing.py", "tests/test_upstream_import_plan.py"]
def test_apply_handles_new_update_and_unchanged_with_recomputable_hashes(tmp_path: Path) -> None:
source, target = _source(tmp_path), _target(tmp_path)
destination = target / "references/upstream/yinduzhanxing/SKILL.md"
first = importer.build_manifest(source, policy_path=POLICY, target=target, apply=True)
first = importer.build_manifest(source, policy_path=POLICY, target=target, apply=True, **_archive_kwargs(source))
assert first["mirror_files"][0]["status"] == "applied"
assert destination.read_bytes() == (source / "SKILL.md").read_bytes()
unchanged = importer.build_manifest(source, policy_path=POLICY, target=target)
subprocess.run(["git", "-C", str(target), "add", "references/upstream/yinduzhanxing/SKILL.md"], check=True)
subprocess.run(["git", "-C", str(target), "commit", "-qm", "record mirrored skill"], check=True)
unchanged = importer.build_manifest(source, policy_path=POLICY, target=target, **_archive_kwargs(source))
assert unchanged["mirror_files"][0]["status"] == "unchanged"
assert unchanged["mirror_files"][0]["source_sha256"] == hashlib.sha256(destination.read_bytes()).hexdigest()
(source / "SKILL.md").write_text("# Updated research skill\n", encoding="utf-8")
update = importer.build_manifest(source, policy_path=POLICY, target=target)
update = importer.build_manifest(source, policy_path=POLICY, target=target, **_archive_kwargs(source))
assert update["mirror_files"][0]["status"] == "update"
assert update["mirror_files"][0]["target_sha256_before"] != update["mirror_files"][0]["target_sha256_after"]
@@ -98,9 +269,11 @@ def test_apply_handles_new_update_and_unchanged_with_recomputable_hashes(tmp_pat
def test_semantic_files_never_overwrite_target(tmp_path: Path) -> None:
source, target = _source(tmp_path), _target(tmp_path)
(target / "SKILL.md").write_text("commercial skill\n", encoding="utf-8")
subprocess.run(["git", "-C", str(target), "add", "SKILL.md"], check=True)
subprocess.run(["git", "-C", str(target), "commit", "-qm", "add commercial skill"], check=True)
before = (target / "SKILL.md").read_bytes()
manifest = importer.build_manifest(source, policy_path=POLICY, target=target, apply=True)
manifest = importer.build_manifest(source, policy_path=POLICY, target=target, apply=True, **_archive_kwargs(source))
assert (target / "SKILL.md").read_bytes() == before
row = next(item for item in manifest["semantic_merge_files"] if item["path"] == "SKILL.md")
@@ -112,16 +285,31 @@ def test_protected_mapping_is_rejected_instead_of_silently_skipped(tmp_path: Pat
source, target = _source(tmp_path), _target(tmp_path)
policy = _policy(tmp_path, mirror=[{"source": "SKILL.md", "target": "frontend/src/stolen.ts", "license": "MIT"}])
with pytest.raises(importer.ImportRejected, match="protected_rejected"):
importer.build_manifest(source, policy_path=policy, target=target, apply=True)
importer.build_manifest(source, policy_path=policy, target=target, apply=True, **_archive_kwargs(source))
assert not (target / "frontend/src/stolen.ts").exists()
@pytest.mark.parametrize(
"target_path",
[
"SKILL.md",
"skills/jyotish-vedic-astrology/SKILL.md",
"skills/jyotish-birth-time-rectification/SKILL.md",
"skills/jyotish-birth-time-rectification/references/private.md",
],
)
def test_commercial_skill_surfaces_cannot_be_mirror_targets(tmp_path: Path, target_path: str) -> None:
policy = _policy(tmp_path, mirror=[{"source": "SKILL.md", "target": target_path, "license": "MIT"}])
with pytest.raises(importer.ImportRejected, match="protected_rejected"):
importer.load_policy(policy)
@pytest.mark.parametrize("path", ["../escape", "/absolute/path", "safe/../../escape"])
def test_path_traversal_and_absolute_paths_are_rejected(tmp_path: Path, path: str) -> None:
source, target = _source(tmp_path), _target(tmp_path)
policy = _policy(tmp_path, mirror=[{"source": "SKILL.md", "target": path, "license": "MIT"}])
with pytest.raises(importer.ImportRejected, match="unsafe_path"):
importer.build_manifest(source, policy_path=policy, target=target)
importer.build_manifest(source, policy_path=policy, target=target, **_archive_kwargs(source))
def test_symlink_escape_is_rejected(tmp_path: Path) -> None:
@@ -130,8 +318,29 @@ def test_symlink_escape_is_rejected(tmp_path: Path) -> None:
outside.write_text("outside\n", encoding="utf-8")
(source / "linked.md").symlink_to(outside)
policy = _policy(tmp_path, mirror=[{"source": "linked.md", "target": "references/upstream/linked.md", "license": "MIT"}])
with pytest.raises(importer.ImportRejected, match="symlink_escape|non_regular_or_symlink"):
importer.build_manifest(source, policy_path=policy, target=target)
with pytest.raises(importer.ImportRejected, match="source_tree_symlink_rejected|symlink_escape|non_regular_or_symlink"):
importer.build_manifest(source, policy_path=policy, target=target, **_archive_kwargs(source))
def test_target_parent_symlink_cannot_redirect_mirror_into_protected_surface(tmp_path: Path) -> None:
source, target = _source(tmp_path), _target(tmp_path)
(target / "frontend").mkdir()
(target / "frontend/.keep").write_text("protected\n", encoding="utf-8")
(target / "references/upstream").mkdir(parents=True)
(target / "references/upstream/yinduzhanxing").symlink_to("../../frontend")
subprocess.run(["git", "-C", str(target), "add", "frontend", "references/upstream/yinduzhanxing"], check=True)
subprocess.run(["git", "-C", str(target), "commit", "-qm", "add redirecting symlink"], check=True)
with pytest.raises(importer.ImportRejected, match="symlink_component_rejected"):
importer.build_manifest(source, policy_path=POLICY, target=target, apply=True, **_archive_kwargs(source))
assert not (target / "frontend/SKILL.md").exists()
def test_target_worktree_must_be_clean_for_deterministic_manifest(tmp_path: Path) -> None:
source, target = _source(tmp_path), _target(tmp_path)
(target / "untracked.txt").write_text("dirty target\n", encoding="utf-8")
with pytest.raises(importer.ImportRejected, match="target_must_be_clean"):
importer.build_manifest(source, policy_path=POLICY, target=target, **_archive_kwargs(source))
def test_sensitive_source_is_rejected(tmp_path: Path) -> None:
@@ -139,7 +348,7 @@ def test_sensitive_source_is_rejected(tmp_path: Path) -> None:
(source / ".env").write_text("OPENAI_API_KEY=not-a-real-key\n", encoding="utf-8")
policy = _policy(tmp_path, mirror=[{"source": ".env", "target": "references/upstream/env.txt", "license": "MIT"}])
with pytest.raises(importer.ImportRejected, match="sensitive_filename|sensitive_content"):
importer.build_manifest(source, policy_path=policy, target=target)
importer.build_manifest(source, policy_path=policy, target=target, **_archive_kwargs(source))
def test_overlapping_source_and_target_are_rejected(tmp_path: Path) -> None:
@@ -169,6 +378,6 @@ def test_apply_rolls_back_earlier_files_when_later_replace_fails(tmp_path: Path,
monkeypatch.setattr(importer.os, "replace", fail_second)
with pytest.raises(OSError, match="simulated"):
importer.build_manifest(source, policy_path=policy, target=target, apply=True)
importer.build_manifest(source, policy_path=policy, target=target, apply=True, **_archive_kwargs(source))
assert not (target / "references/upstream/one.md").exists()
assert not (target / "references/upstream/two.md").exists()
tests/test_report_orchestrator_reader_contract.py
+6 -3
View File
@@ -3,6 +3,7 @@ from __future__ import annotations
import hashlib
import json
import re
from pathlib import Path
from scripts.report_orchestrator import render_reader_report
@@ -42,9 +43,11 @@ def test_upstream_skill_snapshot_matches_manifest_and_commercial_root_remains_ro
root_skill = (ROOT / "SKILL.md").read_text(encoding="utf-8")
linked_skill = ROOT / "skills/jyotish-vedic-astrology/SKILL.md"
assert hashlib.sha256(snapshot.read_bytes()).hexdigest() == manifest["skill_sha256"]
assert manifest["source_mode"] == "snapshot"
assert manifest["source_commit"] is None
snapshot_hash = hashlib.sha256(snapshot.read_bytes()).hexdigest()
assert snapshot_hash == manifest.get("source_skill_sha256", manifest["skill_sha256"])
assert manifest["source_mode"] == "archive"
assert re.fullmatch(r"[0-9a-f]{40}", manifest["source_commit"])
assert re.fullmatch(r"[0-9a-f]{64}", manifest["source_tree_sha256"])
assert linked_skill.resolve() == (ROOT / "SKILL.md").resolve()
assert "商业运行时路由（最高优先级）" in root_skill
assert "references/upstream/yinduzhanxing/SKILL.md" in root_skill
tests/test_upstream_import_plan.py
+157
View File
@@ -0,0 +1,157 @@
"""PR-0 selective upstream import-plan and provenance contracts."""
from __future__ import annotations
import hashlib
import json
import re
from pathlib import Path
from jsonschema import Draft202012Validator, FormatChecker
ROOT = Path(__file__).resolve().parents[1]
UPSTREAM = ROOT / "references/upstream/yinduzhanxing"
PLAN = UPSTREAM / "import-plan.json"
SOURCE_MANIFEST = UPSTREAM / "source-manifest.json"
IMPORT_SCHEMA = ROOT / "references/cross_project_contract/import_manifest.schema.json"
LEGACY_SNAPSHOT = ROOT / "references/cross_project_contract/imports/snapshot-9034e1967032d09c.json"
PINNED_IMPORT_RECORD = ROOT / "references/cross_project_contract/imports/commit-5db72537741fcedaa7b5498502d4a31b0f9fc147.json"
EXPECTED_SOURCES = {
"scripts/evidence_maturity.py",
"scripts/assertion_policy.py",
"scripts/evidence_labeled_reporting.py",
"scripts/calculation_profile_contract.py",
"scripts/report_pack_contract.py",
"scripts/domain_profile_builder.py",
"scripts/unified_calculation_archive.py",
"scripts/specialized_evidence_packets.py",
"scripts/timing_evidence_packets.py",
}
EXPECTED_CATEGORIES = {
"calculation_contracts",
"evidence_assertion_contracts",
"report_domain_profile_contracts",
"research_oracle_assets",
}
EXPECTED_REJECTIONS = {
"fixed_event_domain_count_threshold",
"sparse_start_mid_end_rectification_scan",
"weaken_three_engine_gate",
"weaken_required_layers",
"weaken_stability_gate",
"remove_commercial_routes",
}
EXPECTED_UPSTREAM_IDENTITY = {
"source_repository": "732642856/yinduzhanxing",
"source_repository_url": "https://github.com/732642856/yinduzhanxing",
"source_commit": "5db72537741fcedaa7b5498502d4a31b0f9fc147",
"source_tree_sha256": "18e3122ef73c2776a950bfec128efeb09ac0524ec1d2179390260166b87ca814",
}
EXPECTED_SOURCE_GIT_TREE = "16935cb68a6fa1ef72661cfca4650a42c60e9b2c"
EXPECTED_ARCHIVE_SHA256 = "07d6b71af311160c544433eb9d51beb5b0a5e4bd0afabd4a7d43d3edb573518d"
EXPECTED_ARCHIVE_FILE_COUNT = 2901
EXPECTED_SOURCE_SKILL_SHA256 = "ef453dd8dd4a9da72b56010ec33bb7dab57fd72333d8785cac0986d5456e17c5"
def _load(path: Path) -> dict:
return json.loads(path.read_text(encoding="utf-8"))
def test_import_plan_records_all_nine_upstream_sources_and_semantic_destinations() -> None:
plan = _load(PLAN)
entries = plan["entries"]
assert len(entries) == 9
assert {entry["source"] for entry in entries} == EXPECTED_SOURCES
assert {entry["category"] for entry in entries} == EXPECTED_CATEGORIES
assert [entry["action"] for entry in entries].count("adapt") == 8
assert [entry["action"] for entry in entries].count("reference_only") == 1
for entry in entries:
assert set(("source", "category", "action", "reason", "targets", "tests")) <= set(entry)
assert entry["source"].startswith("scripts/")
assert entry["reason"]
assert entry["targets"] and all(target.startswith(("scripts/", "references/")) for target in entry["targets"])
assert entry["tests"] and all(test.startswith("tests/") for test in entry["tests"])
archive = next(entry for entry in entries if entry["source"] == "scripts/unified_calculation_archive.py")
assert archive["action"] == "reference_only"
assert "runtime truth" in archive["reason"]
assert "none of these research scripts is copied" in plan["boundary"]
def test_rejected_regressions_protect_server_truth_gates_and_commercial_routes() -> None:
plan = _load(PLAN)
rejections = plan["rejected_regressions"]
assert {item["id"] for item in rejections} == EXPECTED_REJECTIONS
assert all(item["action"] == "reject" for item in rejections)
by_id = {item["id"]: item for item in rejections}
fixed_count = by_id["fixed_event_domain_count_threshold"]["reason"].lower()
assert "3 events" in fixed_count and "2 domains" in fixed_count
assert "not sufficient" in fixed_count and "cannot bypass" in fixed_count
sparse_scan = by_id["sparse_start_mid_end_rectification_scan"]["reason"].lower()
assert all(token in sparse_scan for token in ("start", "midpoint", "end"))
assert "not a complete" in sparse_scan
for rejection_id in (
"weaken_three_engine_gate",
"weaken_required_layers",
"weaken_stability_gate",
):
assert by_id[rejection_id]["action"] == "reject"
assert by_id[rejection_id]["commercial_owner"].startswith("scripts/")
routes = by_id["remove_commercial_routes"]
assert routes["preserved_routes"] == ["migration", "family", "annual"]
assert "commercial-runtime-owned" in routes["reason"]
def test_formal_varga_is_evidence_only_and_cannot_close_local_truth() -> None:
policy = _load(PLAN)["formal_varga_policy"]
assert policy["action"] == "reference_only"
reason = policy["reason"].lower()
for token in ("evidence_only", "cannot claim", "cannot", "api key"):
assert token in reason
assert "local-formula parity" in reason
assert "required layers" in reason
def test_source_manifest_pins_valid_archive_commit_tree_and_root_skill_hash() -> None:
manifest = _load(SOURCE_MANIFEST)
plan = _load(PLAN)
assert manifest["source_mode"] == "archive"
assert manifest["import_policy_version"] == 2
assert manifest["imported_at"] != manifest["source_committed_at"]
assert {key: manifest[key] for key in EXPECTED_UPSTREAM_IDENTITY} == EXPECTED_UPSTREAM_IDENTITY
assert {key: plan[key] for key in EXPECTED_UPSTREAM_IDENTITY} == EXPECTED_UPSTREAM_IDENTITY
assert manifest["source_git_tree"] == EXPECTED_SOURCE_GIT_TREE
assert manifest["archive_sha256"] == EXPECTED_ARCHIVE_SHA256
assert manifest["archive_file_count"] == EXPECTED_ARCHIVE_FILE_COUNT
assert manifest["source_committed_at"] == "2026-08-13T18:03:53Z"
assert re.fullmatch(r"[0-9a-f]{40}", manifest["source_commit"])
assert re.fullmatch(r"[0-9a-f]{64}", manifest["source_tree_sha256"])
source_skill_sha256 = manifest.get("source_skill_sha256", manifest["skill_sha256"])
assert re.fullmatch(r"[0-9a-f]{64}", source_skill_sha256)
snapshot_hash = hashlib.sha256((UPSTREAM / "SKILL.md").read_bytes()).hexdigest()
assert source_skill_sha256 == snapshot_hash == EXPECTED_SOURCE_SKILL_SHA256
assert manifest["skill_sha256"] == manifest["source_skill_sha256"]
def test_import_manifest_schema_accepts_archive_and_legacy_snapshot_records() -> None:
schema = _load(IMPORT_SCHEMA)
validator = Draft202012Validator(schema, format_checker=FormatChecker())
legacy = _load(LEGACY_SNAPSHOT)
pinned = _load(PINNED_IMPORT_RECORD)
validator.validate(legacy)
validator.validate(pinned)
invalid_git = dict(pinned, source_mode="git", source_commit="unknown")
assert list(validator.iter_errors(invalid_git))
assert "archive" in schema["properties"]["source_mode"]["enum"]
source_manifest = _load(SOURCE_MANIFEST)
plan = _load(PLAN)
assert pinned["source_repository"] == source_manifest["source_repository"] == plan["source_repository"]
assert pinned["source_repository_url"] == source_manifest["source_repository_url"] == plan["source_repository_url"] == EXPECTED_UPSTREAM_IDENTITY["source_repository_url"]
assert pinned["source_commit"] == source_manifest["source_commit"] == plan["source_commit"]
assert pinned["source_tree_hash"] == source_manifest["source_tree_sha256"] == plan["source_tree_sha256"]
assert pinned["source_skill_sha256"] == source_manifest["source_skill_sha256"] == EXPECTED_SOURCE_SKILL_SHA256