feat(sync): enforce one-way yinduzhanxing import policy

docs/research/yinduzhanxing_one_way_import_2026_08_06.md

@@ -0,0 +1,24 @@
+# Yinduzhanxing -> Jyotisha one-way import
+
+## Authority boundary
+
+- Direction: `732642856/yinduzhanxing` research source -> `root/Jyotisha` commercial target only.
+- Reverse synchronization is forbidden and is not expressible by the v2 policy or importer CLI.
+- The user-provided source at `../yinduzhanxing-main` is a snapshot without usable Git metadata. Imports from it must record `source_commit=unknown` and a deterministic `source_tree_hash`; they must not claim parity with a GitHub commit.
+- Commercial frontend, identity, billing, database, deployment and commercial truth overlays remain protected.
+
+## Review flow
+
+```bash
+.venv/bin/python scripts/import_yinduzhanxing.py \
+  --source ../yinduzhanxing-main \
+  --policy references/cross_project_contract/sync_policy.v2.json \
+  --dry-run \
+  --output artifacts/yinduzhanxing-import-plan.json
+```
+
+A reviewer must inspect semantic merge rows. Only then may the mirror allowlist be applied. `SKILL.md`, `AGENTS.md`, orchestrators and API entrypoints are semantic-merge paths and are never overwritten by the importer.
+
+## Rollback
+
+Revert the commercial import commit. Preserve the manifest for audit history. Never modify or push the research snapshot as part of rollback.

references/cross_project_contract/import_manifest.schema.json

@@ -0,0 +1,66 @@
+{
+  "$schema": "https://json-schema.org/draft/2020-12/schema",
+  "$id": "https://jyotisha.chat/contracts/yinduzhanxing-import-manifest.v1.schema.json",
+  "title": "Yinduzhanxing one-way import manifest",
+  "type": "object",
+  "additionalProperties": false,
+  "required": [
+    "schema_version", "source_repository", "source_commit", "source_tree_hash",
+    "source_mode", "target_repository", "target_base_commit", "policy_version",
+    "generated_at", "mirror_files", "semantic_merge_files", "protected_rejections",
+    "tests_run", "privacy_scan", "operator_review_required"
+  ],
+  "properties": {
+    "schema_version": {"const": 1},
+    "source_repository": {"type": "string", "minLength": 1},
+    "source_commit": {"type": "string", "pattern": "^(unknown|[0-9a-f]{40})$"},
+    "source_tree_hash": {"type": "string", "pattern": "^[0-9a-f]{64}$"},
+    "source_mode": {"enum": ["git", "snapshot"]},
+    "target_repository": {"type": "string", "minLength": 1},
+    "target_base_commit": {"type": "string", "pattern": "^[0-9a-f]{40}$"},
+    "policy_version": {"const": 2},
+    "generated_at": {"type": "string", "format": "date-time"},
+    "mirror_files": {"type": "array", "items": {"$ref": "#/$defs/file"}},
+    "semantic_merge_files": {"type": "array", "items": {"$ref": "#/$defs/semantic"}},
+    "protected_rejections": {"type": "array", "items": {"type": "string"}},
+    "tests_run": {"type": "array", "items": {"type": "string"}},
+    "privacy_scan": {
+      "type": "object",
+      "additionalProperties": false,
+      "required": ["status", "scanned_files", "rejections"],
+      "properties": {
+        "status": {"enum": ["pass", "rejected"]},
+        "scanned_files": {"type": "integer", "minimum": 0},
+        "rejections": {"type": "array", "items": {"type": "string"}}
+      }
+    },
+    "operator_review_required": {"type": "boolean"}
+  },
+  "$defs": {
+    "file": {
+      "type": "object",
+      "additionalProperties": false,
+      "required": ["source", "target", "status", "source_sha256", "target_sha256_before", "target_sha256_after", "license"],
+      "properties": {
+        "source": {"type": "string"}, "target": {"type": "string"},
+        "status": {"enum": ["new", "update", "unchanged", "applied"]},
+        "source_sha256": {"type": "string", "pattern": "^[0-9a-f]{64}$"},
+        "target_sha256_before": {"type": ["string", "null"], "pattern": "^[0-9a-f]{64}$"},
+        "target_sha256_after": {"type": "string", "pattern": "^[0-9a-f]{64}$"},
+        "license": {"type": "string"}
+      }
+    },
+    "semantic": {
+      "type": "object",
+      "additionalProperties": false,
+      "required": ["path", "status", "source_sha256", "target_sha256", "diff_summary"],
+      "properties": {
+        "path": {"type": "string"},
+        "status": {"enum": ["unchanged", "source_missing", "target_missing", "review_required"]},
+        "source_sha256": {"type": ["string", "null"], "pattern": "^[0-9a-f]{64}$"},
+        "target_sha256": {"type": ["string", "null"], "pattern": "^[0-9a-f]{64}$"},
+        "diff_summary": {"type": "string"}
+      }
+    }
+  }
+}

references/cross_project_contract/sync_policy.v2.json

@@ -0,0 +1,52 @@
+{
+  "schema_version": 2,
+  "direction": "research_to_commercial_only",
+  "source_repository": "732642856/yinduzhanxing",
+  "target_repository": "root/Jyotisha",
+  "reverse_sync": "forbidden",
+  "modes": {
+    "mirror": [
+      {
+        "source": "SKILL.md",
+        "target": "references/upstream/yinduzhanxing/SKILL.md",
+        "license": "MIT"
+      }
+    ],
+    "semantic_merge": [
+      "SKILL.md",
+      "AGENTS.md",
+      "references/strict-workflow-router.md",
+      "scripts/unified_consultation_orchestrator.py",
+      "scripts/report_orchestrator.py",
+      "scripts/jyotish_api_server.py"
+    ],
+    "protected": [
+      "frontend/**",
+      "deploy/**",
+      ".gitea/**",
+      ".github/**",
+      "frontend/db/**",
+      "frontend/supabase/**",
+      "references/oracle/commercial_skill_truth_overlay.v1.json",
+      "**/.env",
+      "**/.env.*",
+      "**/*payment*",
+      "**/*billing*",
+      "**/*subscription*",
+      "**/*entitlement*",
+      "**/*admin*",
+      "**/*service_role*"
+    ]
+  },
+  "privacy": {
+    "forbidden_extensions": [".pem", ".key", ".p12", ".pfx", ".dump", ".sql.gz", ".pdf"],
+    "forbidden_basenames": [".env", "cookies.txt", "id_rsa", "id_ed25519"],
+    "content_markers": ["-----BEGIN PRIVATE KEY-----", "SUPABASE_SERVICE_ROLE_KEY=", "VEDASTRO_API_KEY=", "OPENAI_API_KEY=", "DATABASE_URL=postgres"]
+  },
+  "notes": [
+    "Only explicit mirror mappings may be copied byte-for-byte.",
+    "Semantic-merge paths are review inputs and are never overwritten by the importer.",
+    "Commercial product, identity, billing, database and deployment surfaces are protected.",
+    "There is no commercial-to-research mode or gate in schema v2."
+  ]
+}

scripts/import_yinduzhanxing.py

@@ -0,0 +1,348 @@
+#!/usr/bin/env python3
+"""Safely import an explicit research allowlist into the commercial repository.
+
+The importer is deliberately one-way and offline.  It never writes semantic-merge
+or protected paths, and defaults to a dry run.  A source without usable Git
+metadata is recorded as an auditable snapshot with ``source_commit=unknown``.
+"""
+
+from __future__ import annotations
+
+import argparse
+import difflib
+import fnmatch
+import hashlib
+import json
+import os
+import subprocess
+import sys
+import tempfile
+from datetime import UTC, datetime
+from pathlib import Path, PurePosixPath
+from typing import Any
+
+ROOT = Path(__file__).resolve().parents[1]
+DEFAULT_POLICY = ROOT / "references/cross_project_contract/sync_policy.v2.json"
+EXIT_OK = 0
+EXIT_SEMANTIC_REVIEW = 2
+EXIT_POLICY_REJECTED = 3
+EXIT_VALIDATION_FAILED = 4
+
+
+class ImportRejected(ValueError):
+    """The requested import violates a direction, path, privacy or repository guard."""
+
+
+def sha256_bytes(value: bytes) -> str:
+    return hashlib.sha256(value).hexdigest()
+
+
+def sha256_file(path: Path) -> str:
+    return sha256_bytes(path.read_bytes())
+
+
+def _safe_relative(value: str) -> str:
+    if not isinstance(value, str) or not value.strip():
+        raise ImportRejected("empty_path")
+    candidate = PurePosixPath(value.replace("\\", "/"))
+    if candidate.is_absolute() or ".." in candidate.parts or "." in candidate.parts:
+        raise ImportRejected(f"unsafe_path:{value}")
+    normalized = candidate.as_posix()
+    if normalized.startswith("/") or "\x00" in normalized:
+        raise ImportRejected(f"unsafe_path:{value}")
+    return normalized
+
+
+def _resolved_file(root: Path, relative: str, *, must_exist: bool) -> Path:
+    path = root / _safe_relative(relative)
+    root_resolved = root.resolve()
+    if must_exist and not path.exists():
+        return path
+    resolved = path.resolve(strict=must_exist)
+    if resolved != root_resolved and root_resolved not in resolved.parents:
+        raise ImportRejected(f"symlink_escape:{relative}")
+    if must_exist and (not path.is_file() or path.is_symlink()):
+        raise ImportRejected(f"non_regular_or_symlink:{relative}")
+    return path
+
+
+def _matches(path: str, patterns: list[str]) -> bool:
+    lowered = path.casefold()
+    return any(fnmatch.fnmatchcase(lowered, pattern.casefold()) for pattern in patterns)
+
+
+def load_policy(path: Path) -> dict[str, Any]:
+    policy = json.loads(path.read_text(encoding="utf-8"))
+    if policy.get("schema_version") != 2:
+        raise ImportRejected("policy_schema_version_must_be_2")
+    if policy.get("direction") != "research_to_commercial_only":
+        raise ImportRejected("direction_must_be_research_to_commercial_only")
+    if policy.get("reverse_sync") != "forbidden":
+        raise ImportRejected("reverse_sync_must_be_forbidden")
+    if "commercial_to_research" in json.dumps(policy, sort_keys=True):
+        raise ImportRejected("reverse_direction_is_not_expressible")
+    modes = policy.get("modes")
+    if not isinstance(modes, dict) or set(modes) != {"mirror", "semantic_merge", "protected"}:
+        raise ImportRejected("policy_modes_invalid")
+    if not isinstance(modes["mirror"], list) or not isinstance(modes["semantic_merge"], list):
+        raise ImportRejected("policy_allowlists_invalid")
+    protected = [_safe_relative(value) if "*" not in value else value.replace("\\", "/") for value in modes["protected"]]
+    for mapping in modes["mirror"]:
+        if not isinstance(mapping, dict) or not {"source", "target", "license"} <= set(mapping):
+            raise ImportRejected("mirror_mapping_invalid")
+        _safe_relative(mapping["source"])
+        target = _safe_relative(mapping["target"])
+        if _matches(target, protected):
+            raise ImportRejected(f"protected_rejected:{target}")
+    for value in modes["semantic_merge"]:
+        _safe_relative(value)
+    return policy
+
+
+def _git_output(root: Path, *args: str) -> str | None:
+    try:
+        return subprocess.check_output(
+            ["git", "-C", str(root), *args], stderr=subprocess.DEVNULL, text=True, timeout=8
+        ).strip()
+    except (OSError, subprocess.SubprocessError):
+        return None
+
+
+def source_identity(source: Path) -> tuple[str, str]:
+    inside = _git_output(source, "rev-parse", "--show-toplevel")
+    commit = _git_output(source, "rev-parse", "HEAD") if inside and Path(inside).resolve() == source.resolve() else None
+    if commit and len(commit) == 40 and all(ch in "0123456789abcdef" for ch in commit.lower()):
+        return "git", commit.lower()
+    return "snapshot", "unknown"
+
+
+def _normalized_remote(root: Path) -> str | None:
+    remote = _git_output(root, "remote", "get-url", "origin")
+    if not remote:
+        return None
+    value = remote.strip().lower().removesuffix(".git").replace("git@", "ssh://git@")
+    return value.rstrip("/")
+
+
+def validate_roots(source: Path, target: Path) -> None:
+    source = source.expanduser().resolve(strict=True)
+    target = target.expanduser().resolve(strict=True)
+    if not source.is_dir() or not target.is_dir():
+        raise ImportRejected("source_and_target_must_be_directories")
+    if source == target or source in target.parents or target in source.parents:
+        raise ImportRejected("source_and_target_must_not_overlap")
+    source_remote = _normalized_remote(source)
+    target_remote = _normalized_remote(target)
+    if source_remote and target_remote and source_remote == target_remote:
+        raise ImportRejected("source_and_target_git_remote_must_differ")
+
+
+def source_tree_hash(source: Path) -> str:
+    digest = hashlib.sha256()
+    for path in sorted(source.rglob("*"), key=lambda item: item.relative_to(source).as_posix()):
+        relative = path.relative_to(source).as_posix()
+        if ".git" in PurePosixPath(relative).parts or path.is_symlink() or not path.is_file():
+            continue
+        digest.update(relative.encode("utf-8"))
+        digest.update(b"\0")
+        digest.update(bytes.fromhex(sha256_file(path)))
+        digest.update(b"\0")
+    return digest.hexdigest()
+
+
+def _privacy_rejection(path: Path, relative: str, privacy: dict[str, Any]) -> str | None:
+    lowered = relative.casefold()
+    basenames = {str(value).casefold() for value in privacy.get("forbidden_basenames", [])}
+    extensions = [str(value).casefold() for value in privacy.get("forbidden_extensions", [])]
+    if path.name.casefold() in basenames or any(lowered.endswith(ext) for ext in extensions):
+        return f"sensitive_filename:{relative}"
+    data = path.read_bytes()
+    if b"\x00" in data[:8192]:
+        return f"binary_content_rejected:{relative}"
+    text = data.decode("utf-8", errors="replace")
+    for marker in privacy.get("content_markers", []):
+        if str(marker).casefold() in text.casefold():
+            return f"sensitive_content:{relative}"
+    return None
+
+
+def _diff_summary(source: Path | None, target: Path | None) -> str:
+    if source is None:
+        return "source missing"
+    if target is None:
+        return "target missing; manual semantic merge required"
+    try:
+        before = target.read_text(encoding="utf-8").splitlines()
+        after = source.read_text(encoding="utf-8").splitlines()
+    except UnicodeDecodeError:
+        return "binary difference; manual semantic merge required"
+    added = removed = 0
+    for line in difflib.ndiff(before, after):
+        added += line.startswith("+ ")
+        removed += line.startswith("- ")
+    return f"manual semantic merge required: +{added}/-{removed} lines"
+
+
+def build_manifest(
+    source: Path,
+    *,
+    policy_path: Path = DEFAULT_POLICY,
+    target: Path = ROOT,
+    apply: bool = False,
+) -> dict[str, Any]:
+    source = source.expanduser().resolve(strict=True)
+    target = target.expanduser().resolve(strict=True)
+    validate_roots(source, target)
+    policy = load_policy(policy_path)
+    protected = list(policy["modes"]["protected"])
+    privacy = policy.get("privacy") if isinstance(policy.get("privacy"), dict) else {}
+    source_mode, source_commit = source_identity(source)
+    target_commit = _git_output(target, "rev-parse", "HEAD")
+    if not target_commit or len(target_commit) != 40:
+        raise ImportRejected("target_base_commit_unavailable")
+
+    mirror_files: list[dict[str, Any]] = []
+    privacy_rejections: list[str] = []
+    pending_writes: list[tuple[Path, bytes, dict[str, Any]]] = []
+    for mapping in policy["modes"]["mirror"]:
+        source_rel = _safe_relative(mapping["source"])
+        target_rel = _safe_relative(mapping["target"])
+        if _matches(target_rel, protected):
+            raise ImportRejected(f"protected_rejected:{target_rel}")
+        source_file = _resolved_file(source, source_rel, must_exist=True)
+        if not source_file.exists():
+            raise ImportRejected(f"mirror_source_missing:{source_rel}")
+        target_file = _resolved_file(target, target_rel, must_exist=False)
+        rejection = _privacy_rejection(source_file, source_rel, privacy)
+        if rejection:
+            privacy_rejections.append(rejection)
+            continue
+        data = source_file.read_bytes()
+        source_hash = sha256_bytes(data)
+        before_hash = sha256_file(target_file) if target_file.exists() and target_file.is_file() else None
+        status = "unchanged" if source_hash == before_hash else "update" if before_hash else "new"
+        row = {
+            "source": source_rel,
+            "target": target_rel,
+            "status": status,
+            "source_sha256": source_hash,
+            "target_sha256_before": before_hash,
+            "target_sha256_after": source_hash,
+            "license": str(mapping["license"]),
+        }
+        mirror_files.append(row)
+        if status != "unchanged":
+            pending_writes.append((target_file, data, row))
+
+    if privacy_rejections:
+        raise ImportRejected(";".join(privacy_rejections))
+
+    semantic_files: list[dict[str, Any]] = []
+    for relative_value in policy["modes"]["semantic_merge"]:
+        relative = _safe_relative(relative_value)
+        source_file = _resolved_file(source, relative, must_exist=False)
+        target_file = _resolved_file(target, relative, must_exist=False)
+        source_exists = source_file.exists() and source_file.is_file() and not source_file.is_symlink()
+        target_exists = target_file.exists() and target_file.is_file() and not target_file.is_symlink()
+        source_hash = sha256_file(source_file) if source_exists else None
+        target_hash = sha256_file(target_file) if target_exists else None
+        if not source_exists:
+            status = "source_missing"
+        elif not target_exists:
+            status = "target_missing"
+        elif source_hash == target_hash:
+            status = "unchanged"
+        else:
+            status = "review_required"
+        semantic_files.append({
+            "path": relative,
+            "status": status,
+            "source_sha256": source_hash,
+            "target_sha256": target_hash,
+            "diff_summary": _diff_summary(source_file if source_exists else None, target_file if target_exists else None),
+        })
+
+    if apply and pending_writes:
+        backups: list[tuple[Path, bytes | None]] = []
+        try:
+            for destination, data, row in pending_writes:
+                destination.parent.mkdir(parents=True, exist_ok=True)
+                backups.append((destination, destination.read_bytes() if destination.exists() else None))
+                fd, temp_name = tempfile.mkstemp(prefix=f".{destination.name}.", dir=destination.parent)
+                try:
+                    with os.fdopen(fd, "wb") as handle:
+                        handle.write(data)
+                        handle.flush()
+                        os.fsync(handle.fileno())
+                    os.chmod(temp_name, 0o644)
+                    os.replace(temp_name, destination)
+                finally:
+                    if os.path.exists(temp_name):
+                        os.unlink(temp_name)
+                row["status"] = "applied"
+        except Exception:
+            for destination, previous in reversed(backups):
+                if previous is None:
+                    destination.unlink(missing_ok=True)
+                else:
+                    destination.write_bytes(previous)
+            raise
+
+    return {
+        "schema_version": 1,
+        "source_repository": policy["source_repository"],
+        "source_commit": source_commit,
+        "source_tree_hash": source_tree_hash(source),
+        "source_mode": source_mode,
+        "target_repository": policy["target_repository"],
+        "target_base_commit": target_commit.lower(),
+        "policy_version": 2,
+        "generated_at": datetime.now(UTC).isoformat().replace("+00:00", "Z"),
+        "mirror_files": mirror_files,
+        "semantic_merge_files": semantic_files,
+        "protected_rejections": [],
+        "tests_run": [],
+        "privacy_scan": {"status": "pass", "scanned_files": len(mirror_files), "rejections": []},
+        "operator_review_required": any(row["status"] in {"review_required", "target_missing"} for row in semantic_files),
+    }
+
+
+def write_json_atomic(path: Path, value: dict[str, Any]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    data = (json.dumps(value, ensure_ascii=False, indent=2, sort_keys=True) + "\n").encode("utf-8")
+    fd, temp_name = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent)
+    try:
+        with os.fdopen(fd, "wb") as handle:
+            handle.write(data)
+            handle.flush()
+            os.fsync(handle.fileno())
+        os.replace(temp_name, path)
+    finally:
+        if os.path.exists(temp_name):
+            os.unlink(temp_name)
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--source", type=Path, required=True)
+    parser.add_argument("--policy", type=Path, default=DEFAULT_POLICY)
+    mode = parser.add_mutually_exclusive_group()
+    mode.add_argument("--dry-run", action="store_true")
+    mode.add_argument("--apply", action="store_true")
+    parser.add_argument("--output", type=Path, required=True)
+    args = parser.parse_args(argv)
+    try:
+        manifest = build_manifest(args.source, policy_path=args.policy, apply=args.apply)
+        write_json_atomic(args.output, manifest)
+    except ImportRejected as error:
+        print(json.dumps({"status": "rejected", "reason": str(error)}, ensure_ascii=False), file=sys.stderr)
+        return EXIT_POLICY_REJECTED
+    except (OSError, ValueError, json.JSONDecodeError) as error:
+        print(json.dumps({"status": "invalid", "reason": str(error)}, ensure_ascii=False), file=sys.stderr)
+        return EXIT_VALIDATION_FAILED
+    print(json.dumps(manifest, ensure_ascii=False, indent=2, sort_keys=True))
+    return EXIT_SEMANTIC_REVIEW if manifest["operator_review_required"] else EXIT_OK
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/test_import_yinduzhanxing.py

@@ -0,0 +1,174 @@
+from __future__ import annotations
+
+import hashlib
+import json
+import subprocess
+import sys
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[1]
+SCRIPTS = ROOT / "scripts"
+if str(SCRIPTS) not in sys.path:
+    sys.path.insert(0, str(SCRIPTS))
+
+import import_yinduzhanxing as importer  # noqa: E402
+
+POLICY = ROOT / "references/cross_project_contract/sync_policy.v2.json"
+
+
+def _git(root: Path, *args: str) -> str:
+    return subprocess.check_output(["git", "-C", str(root), *args], text=True).strip()
+
+
+def _target(tmp_path: Path) -> Path:
+    target = tmp_path / "commercial"
+    target.mkdir()
+    subprocess.run(["git", "init", "-q", str(target)], check=True)
+    subprocess.run(["git", "-C", str(target), "config", "user.email", "tests@example.invalid"], check=True)
+    subprocess.run(["git", "-C", str(target), "config", "user.name", "Tests"], check=True)
+    (target / "README.md").write_text("commercial\n", encoding="utf-8")
+    subprocess.run(["git", "-C", str(target), "add", "README.md"], check=True)
+    subprocess.run(["git", "-C", str(target), "commit", "-qm", "base"], check=True)
+    return target
+
+
+def _source(tmp_path: Path) -> Path:
+    source = tmp_path / "snapshot"
+    (source / "scripts").mkdir(parents=True)
+    (source / "references").mkdir()
+    (source / "SKILL.md").write_text("# Research skill\n", encoding="utf-8")
+    (source / "AGENTS.md").write_text("research agent\n", encoding="utf-8")
+    (source / "references/strict-workflow-router.md").write_text("router\n", encoding="utf-8")
+    (source / "scripts/unified_consultation_orchestrator.py").write_text("# source\n", encoding="utf-8")
+    (source / "scripts/report_orchestrator.py").write_text("# report\n", encoding="utf-8")
+    (source / "scripts/jyotish_api_server.py").write_text("# api\n", encoding="utf-8")
+    return source
+
+
+def _policy(tmp_path: Path, *, mirror: list[dict[str, str]] | None = None) -> Path:
+    value = json.loads(POLICY.read_text(encoding="utf-8"))
+    if mirror is not None:
+        value["modes"]["mirror"] = mirror
+    path = tmp_path / "policy.json"
+    path.write_text(json.dumps(value), encoding="utf-8")
+    return path
+
+
+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 "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:
+    source, target = _source(tmp_path), _target(tmp_path)
+    before = _git(target, "status", "--porcelain=v1")
+    manifest = importer.build_manifest(source, policy_path=POLICY, target=target)
+    after = _git(target, "status", "--porcelain=v1")
+
+    assert before == after == ""
+    assert manifest["source_mode"] == "snapshot"
+    assert manifest["source_commit"] == "unknown"
+    assert manifest["mirror_files"][0]["status"] == "new"
+    assert not (target / "references/upstream/yinduzhanxing/SKILL.md").exists()
+
+
+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)
+    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)
+    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)
+    assert update["mirror_files"][0]["status"] == "update"
+    assert update["mirror_files"][0]["target_sha256_before"] != update["mirror_files"][0]["target_sha256_after"]
+
+
+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")
+    before = (target / "SKILL.md").read_bytes()
+
+    manifest = importer.build_manifest(source, policy_path=POLICY, target=target, apply=True)
+
+    assert (target / "SKILL.md").read_bytes() == before
+    row = next(item for item in manifest["semantic_merge_files"] if item["path"] == "SKILL.md")
+    assert row["status"] == "review_required"
+    assert manifest["operator_review_required"] is True
+
+
+def test_protected_mapping_is_rejected_instead_of_silently_skipped(tmp_path: Path) -> None:
+    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)
+    assert not (target / "frontend/src/stolen.ts").exists()
+
+
+@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)
+
+
+def test_symlink_escape_is_rejected(tmp_path: Path) -> None:
+    source, target = _source(tmp_path), _target(tmp_path)
+    outside = tmp_path / "outside.md"
+    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)
+
+
+def test_sensitive_source_is_rejected(tmp_path: Path) -> None:
+    source, target = _source(tmp_path), _target(tmp_path)
+    (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)
+
+
+def test_overlapping_source_and_target_are_rejected(tmp_path: Path) -> None:
+    target = _target(tmp_path)
+    source = target / "source"
+    source.mkdir()
+    with pytest.raises(importer.ImportRejected, match="must_not_overlap"):
+        importer.build_manifest(source, policy_path=POLICY, target=target)
+
+
+def test_apply_rolls_back_earlier_files_when_later_replace_fails(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    source, target = _source(tmp_path), _target(tmp_path)
+    (source / "second.md").write_text("second\n", encoding="utf-8")
+    policy = _policy(tmp_path, mirror=[
+        {"source": "SKILL.md", "target": "references/upstream/one.md", "license": "MIT"},
+        {"source": "second.md", "target": "references/upstream/two.md", "license": "MIT"},
+    ])
+    original_replace = importer.os.replace
+    calls = 0
+
+    def fail_second(source_path: str, target_path: Path) -> None:
+        nonlocal calls
+        calls += 1
+        if calls == 2:
+            raise OSError("simulated replace failure")
+        original_replace(source_path, target_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)
+    assert not (target / "references/upstream/one.md").exists()
+    assert not (target / "references/upstream/two.md").exists()