diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 3c42eee..0804070 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -14,11 +14,21 @@ jobs: timeout-minutes: 20 steps: - uses: actions/checkout@v6 + with: + fetch-depth: 0 - uses: actions/setup-python@v6 with: python-version: "3.11" - name: Install package run: python -m pip install -e '.[test]' + - name: Validate changed live signal pair structure + env: + PAIR_EVENT: ${{ github.event_name }} + PAIR_BASE_SHA: ${{ github.event.pull_request.base.sha || github.event.before }} + PAIR_HEAD_SHA: ${{ github.event.pull_request.head.sha || github.sha }} + run: | + python scripts/validate_latest_signal.py --changed-pair \ + --event "$PAIR_EVENT" --base-sha "$PAIR_BASE_SHA" --head-sha "$PAIR_HEAD_SHA" - name: Run tests run: python -m pytest -q - name: Validate example signal diff --git a/docs/architecture.md b/docs/architecture.md index a24697c..8fde2c9 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -78,6 +78,52 @@ existing QuantStrategyLab `symbol,as_of,close` price-history files. Large source files should stay in their owning strategy repositories or object storage; this repository only stores small extracted replay inputs when needed for research. +## Strict Live Signal-Pair Structure Gate + +Legacy examples and replay retain their existing schema reader. A separate +strict publication check applies to the canonical pair +`data/output/latest_signal.json` and `data/output/latest_signal.manifest.json`. +It requires an integer-v2 manifest while retaining the consumer's supported +signal schemas 1 and 2; it never upgrades or rewrites retained data. + +The strict check reads each input once, binds the exact signal bytes to the +manifest hash, rejects undeclared fields, and checks matching dates, mode, +timezone-aware generation, bounded finite confidence, and no-execution policy. +Its success label is `PAIR_STRUCTURE_VALID`. Actual context bytes, model use, +and producer-run provenance remain `UNVERIFIED`: a formatted input digest or a +repository commit ancestor does not prove a real model run used that context. +Authentic context/run linkage and source rights remain separate publication +review conditions; no downloader or synthetic provenance default is introduced. + +The existing required `test` job retains `contents: read`. For PRs it checks +the merge-base-to-head change; for main pushes it checks before-to-head. If +either canonical live path changes, both must change and exist as regular head +tree blobs. Deletion, rename away from a canonical path, single-file updates, +invalid comparison commits, or workspace substitution fail closed. Validation +uses committed head bytes and verifies producer ancestry as repository lineage +only. Unrelated changes still run ordinary tests without relabeling the retained +legacy pair trusted or blocking all work because that old pair is stale. + +Commands are `validate_latest_signal.py --require-manifest` for an +explicit file-pair check, and `--changed-pair --event pull_request|push +--base-sha --head-sha ` for the local Git-only CI gate. Schema-only +examples and the existing `--allow-missing` diagnostic remain available, but +cannot bypass the strict gate. No clock-relative freshness is asserted here; +the downstream report evaluates its own fixed cutoff. + +Recurring publisher identity and permission scope are separate decisions from +this structural validation. Public derived-snapshot publication must follow its +explicitly approved scope and applicable source-data rights; a one-time import +approval does not establish a recurring publication service or a general exception +to the generated-artifact contribution policy. +Repository MIT is not proof of market-data redistribution rights. This patch +does not change permissions or CONTRIBUTING and does not import any real data. + +中文摘要：保留旧样例与回放读取；仅当正式 signal/manifest 路径发生变更时，要求成对提交并 +验证原始字节、字段、时间和政策。成功只表示 PAIR_STRUCTURE_VALID，不代表实际 context、 +模型或生产运行已核实。无关 PR 不被旧数据阻塞；单侧修改、删除、改名、非法 SHA 或工作树 +替换均拒绝。持续发布身份和派生数据公开权利另行决策，本补丁不导入数据或扩大权限。 + ## Risk Notes The artifact is research evidence, not a trading instruction. Missing evidence, diff --git a/scripts/validate_latest_signal.py b/scripts/validate_latest_signal.py index 24f2b09..c76dfec 100644 --- a/scripts/validate_latest_signal.py +++ b/scripts/validate_latest_signal.py @@ -6,6 +6,7 @@ import hashlib import json import re +import subprocess import sys from pathlib import Path from typing import Any @@ -14,14 +15,50 @@ sys.path.insert(0, str(ROOT / "src")) from research_signal_context_pipelines import SignalValidationError, validate_signal # noqa: E402 +from research_signal_context_pipelines.schema import validate_signal_for_publication # noqa: E402 DEFAULT_SIGNAL_PATH = Path("data/output/latest_signal.json") EXPECTED_MANIFEST_TYPE = "research_signal_context" EXPECTED_ARTIFACT_PATH = "data/output/latest_signal.json" EXPECTED_PRODUCER_REPOSITORY = "QuantStrategyLab/ResearchSignalContextPipelines" -COMMIT_SHA_PATTERN = re.compile(r"[0-9a-fA-F]{40}") +COMMIT_SHA_PATTERN = re.compile(r"[0-9a-f]{40}") SHA256_PATTERN = re.compile(r"[0-9a-f]{64}") +MANIFEST_PATH = "data/output/latest_signal.manifest.json" +PAIR_PATHS = frozenset({EXPECTED_ARTIFACT_PATH, MANIFEST_PATH}) +MANIFEST_KEYS = frozenset({ + "manifest_type", "schema_version", "artifact", "as_of", "generated_at", + "expires_at", "mode", "producer", "input_digest", "policy", +}) +STRUCTURE_SUCCESS = ( + "PAIR_STRUCTURE_VALID context_provenance=UNVERIFIED " + "producer_run_provenance=UNVERIFIED model_use=UNVERIFIED" +) + + +def _parse_json(raw: bytes, label: str) -> Mapping[str, Any]: + def unique_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]: + result: dict[str, Any] = {} + for key, value in pairs: + if key in result: + raise SignalValidationError("duplicate JSON key") + result[key] = value + return result + + def reject_constant(value: str) -> None: + raise SignalValidationError("non-finite JSON constant") + + try: + value = json.loads(raw.decode("utf-8"), object_pairs_hook=unique_object, parse_constant=reject_constant) + except (UnicodeError, json.JSONDecodeError) as exc: + raise SignalValidationError(f"invalid {label} JSON") from exc + return _require_mapping(value, label) + + +def _exact_keys(value: Mapping[str, Any], expected: frozenset[str], label: str) -> None: + if set(value) != expected: + missing = ", ".join(sorted(expected - set(value))) or "none" + raise SignalValidationError(f"{label} keys mismatch; missing: {missing}; unknown fields are not permitted") def _require_mapping(value: Any, name: str) -> Mapping[str, Any]: @@ -40,23 +77,31 @@ def validate_manifest_v2( signal_path: Path, signal: Mapping[str, Any], manifest: Mapping[str, Any], + *, + signal_bytes: bytes | None = None, ) -> None: + raw = signal_path.read_bytes() if signal_bytes is None else signal_bytes + actual_signal = _parse_json(raw, "signal") + if json.dumps(actual_signal, sort_keys=True, allow_nan=False) != json.dumps(signal, sort_keys=True, allow_nan=False): + raise SignalValidationError("supplied signal payload does not match signal bytes") manifest = _require_mapping(manifest, "manifest") schema_version = manifest.get("schema_version") if schema_version in ("1", 1): raise SignalValidationError("legacy_untrusted: manifest schema_version 1 cannot satisfy immutable provenance") - if schema_version != 2 or isinstance(schema_version, bool): + if type(schema_version) is not int or schema_version != 2: raise SignalValidationError("manifest schema_version must be 2") + _exact_keys(manifest, MANIFEST_KEYS, "manifest") if manifest.get("manifest_type") != EXPECTED_MANIFEST_TYPE: raise SignalValidationError(f"manifest_type must be {EXPECTED_MANIFEST_TYPE!r}") artifact = _require_mapping(manifest.get("artifact"), "artifact") + _exact_keys(artifact, frozenset({"path", "sha256"}), "artifact") if artifact.get("path") != EXPECTED_ARTIFACT_PATH: raise SignalValidationError(f"artifact.path must be {EXPECTED_ARTIFACT_PATH!r}") artifact_sha256 = _require_string(artifact.get("sha256"), "artifact.sha256") if not SHA256_PATTERN.fullmatch(artifact_sha256): raise SignalValidationError("artifact.sha256 must be a lowercase SHA-256 hex digest") - if artifact_sha256 != hashlib.sha256(signal_path.read_bytes()).hexdigest(): + if artifact_sha256 != hashlib.sha256(raw).hexdigest(): raise SignalValidationError("artifact.sha256 does not match signal bytes") for field in ("as_of", "generated_at", "expires_at", "mode"): @@ -64,6 +109,7 @@ def validate_manifest_v2( raise SignalValidationError(f"{field} must exactly match the signal") producer = _require_mapping(manifest.get("producer"), "producer") + _exact_keys(producer, frozenset({"repository", "commit_sha"}), "producer") if producer.get("repository") != EXPECTED_PRODUCER_REPOSITORY: raise SignalValidationError(f"producer.repository must be {EXPECTED_PRODUCER_REPOSITORY!r}") producer_commit_sha = _require_string(producer.get("commit_sha"), "producer.commit_sha") @@ -75,20 +121,110 @@ def validate_manifest_v2( raise SignalValidationError("input_digest must be sha256:<64 lowercase hex>") policy = _require_mapping(manifest.get("policy"), "policy") + _exact_keys(policy, frozenset({"execution_allowed"}), "policy") if policy.get("execution_allowed") is not False: raise SignalValidationError("policy.execution_allowed must be false") - for field in ("publication_commit", "artifact_commit", "publisher_commit"): - if field in manifest or field in producer: - raise SignalValidationError(f"{field} is not permitted in immutable provenance manifests") + +def validate_pair_bytes(signal_bytes: bytes, manifest_bytes: bytes) -> Mapping[str, Any]: + signal = _parse_json(signal_bytes, "signal") + manifest = _parse_json(manifest_bytes, "manifest") + validate_signal_for_publication(signal) + validate_manifest_v2(Path(EXPECTED_ARTIFACT_PATH), signal, manifest, signal_bytes=signal_bytes) + return manifest + + +def _git(repo: Path, *args: str) -> bytes: + result = subprocess.run(["git", "-C", str(repo), *args], check=False, capture_output=True) + if result.returncode: + raise SignalValidationError(f"Git {args[0]} check failed") + return result.stdout + + +def _require_commit(repo: Path, sha: str) -> None: + if not COMMIT_SHA_PATTERN.fullmatch(sha): + raise SignalValidationError("comparison and producer commits must be lowercase 40-hex SHAs") + _git(repo, "cat-file", "-e", f"{sha}^{{commit}}") + + +def _head_blob(repo: Path, head: str, path: str) -> bytes: + entry = _git(repo, "ls-tree", "-z", head, "--", path).split(b"\0") + if len(entry) != 2 or entry[-1] or b"\t" not in entry[0]: + raise SignalValidationError("both canonical live pair files must exist") + metadata, name = entry[0].split(b"\t", 1) + mode, kind, sha = metadata.split() + if name.decode() != path or kind != b"blob" or mode not in {b"100644", b"100755"}: + raise SignalValidationError("canonical live pair members must be regular Git blobs") + return _git(repo, "cat-file", "blob", sha.decode("ascii")) + + +def _require_worktree_bytes(repo: Path, path: str, expected: bytes) -> None: + current = repo + for part in Path(path).parts: + current = current / part + if current.is_symlink(): + raise SignalValidationError("live pair worktree symlinks are not permitted") + if not current.is_file() or current.read_bytes() != expected: + raise SignalValidationError("live pair worktree bytes do not match committed head") + + +def validate_changed_pair(repo: Path, *, base_sha: str, head_sha: str, event: str) -> bool: + """Validate changed committed pair bytes; no network or repository mutation.""" + repo = repo.resolve() + if event not in {"pull_request", "push"}: + raise SignalValidationError("changed-pair gate requires pull_request or push") + _require_commit(repo, base_sha) + _require_commit(repo, head_sha) + comparison_base = base_sha + if event == "pull_request": + comparison_base = _git(repo, "merge-base", base_sha, head_sha).decode("ascii").strip() + _require_commit(repo, comparison_base) + raw_paths = _git( + repo, "diff", "--name-only", "--no-renames", "-z", comparison_base, head_sha, + "--", *sorted(PAIR_PATHS), + ) + changed = {path.decode("utf-8") for path in raw_paths.split(b"\0") if path} + if not changed: + return False + if changed != PAIR_PATHS: + raise SignalValidationError("live signal and manifest must change together") + blobs = {path: _head_blob(repo, head_sha, path) for path in PAIR_PATHS} + for path, raw in blobs.items(): + _require_worktree_bytes(repo, path, raw) + manifest = validate_pair_bytes(blobs[EXPECTED_ARTIFACT_PATH], blobs[MANIFEST_PATH]) + producer = manifest["producer"]["commit_sha"] + _require_commit(repo, producer) + _git(repo, "merge-base", "--is-ancestor", producer, head_sha) + return True def main() -> int: parser = argparse.ArgumentParser(description="Validate a long-horizon shadow signal artifact.") parser.add_argument("path", nargs="?", default=str(DEFAULT_SIGNAL_PATH), help="Signal JSON path") parser.add_argument("--allow-missing", action="store_true", help="Exit successfully when the path does not exist") + parser.add_argument("--require-manifest", action="store_true", help="Require strict pair structure; does not authenticate a producer run") + parser.add_argument("--changed-pair", action="store_true", help="Gate canonical live pair changes using local committed Git objects") + parser.add_argument("--event", choices=("pull_request", "push")) + parser.add_argument("--base-sha") + parser.add_argument("--head-sha") args = parser.parse_args() + if args.allow_missing and (args.require_manifest or args.changed_pair): + parser.error("--allow-missing cannot bypass strict pair validation") + if args.changed_pair: + if not all((args.event, args.base_sha, args.head_sha)): + parser.error("--changed-pair requires --event, --base-sha, and --head-sha") + if args.path != str(DEFAULT_SIGNAL_PATH) or args.require_manifest: + parser.error("--changed-pair reads only canonical committed paths") + try: + changed = validate_changed_pair(Path.cwd(), base_sha=args.base_sha, head_sha=args.head_sha, event=args.event) + except (SignalValidationError, OSError) as exc: + raise SystemExit(f"invalid signal pair structure: {exc}") from exc + print(STRUCTURE_SUCCESS if changed else "NO_LIVE_PAIR_CHANGE retained_pair_status=DIAGNOSTIC_UNTRUSTED") + return 0 + if any((args.event, args.base_sha, args.head_sha)): + parser.error("Git comparison arguments require --changed-pair") + path = Path(args.path) if not path.exists(): if args.allow_missing: @@ -96,18 +232,25 @@ def main() -> int: return 0 raise SystemExit(f"signal artifact not found: {path}") - payload = json.loads(path.read_text(encoding="utf-8")) try: - validate_signal(payload) + signal_bytes = path.read_bytes() manifest_path = path.with_suffix(".manifest.json") + if args.require_manifest: + if not manifest_path.is_file(): + raise SignalValidationError("strict pair requires a sibling manifest") + validate_pair_bytes(signal_bytes, manifest_path.read_bytes()) + print(STRUCTURE_SUCCESS) + return 0 + payload = json.loads(signal_bytes.decode("utf-8")) + validate_signal(payload) if manifest_path.exists(): manifest = json.loads(manifest_path.read_text(encoding="utf-8")) - validate_manifest_v2(path, payload, manifest) + validate_manifest_v2(path, payload, manifest, signal_bytes=signal_bytes) except SignalValidationError as exc: raise SystemExit(f"invalid or untrusted signal artifact: {exc}") from exc if path.with_suffix(".manifest.json").exists(): - print(f"valid signal artifact with immutable provenance: {path}") + print(f"valid signal schema and manifest structure: {path}; actual context/run provenance UNVERIFIED") else: print(f"valid signal schema only; immutable provenance unavailable: {path}") return 0 diff --git a/src/research_signal_context_pipelines/schema.py b/src/research_signal_context_pipelines/schema.py index 0119d1b..16a67fe 100644 --- a/src/research_signal_context_pipelines/schema.py +++ b/src/research_signal_context_pipelines/schema.py @@ -31,6 +31,12 @@ class SignalValidationError(ValueError): ALLOWED_BIAS_VALUES = frozenset({"positive", "negative", "neutral", "watch", "avoid"}) REQUIRED_SIGNAL_HORIZON = "1-3 years" SUPPORTED_SCHEMA_VERSIONS = frozenset({"1", "2"}) +PUBLICATION_ALLOWED_KEYS = frozenset(REQUIRED_TOP_LEVEL_KEYS) | frozenset( + {"model_version", "scoring_version", "theme_bias", "symbol_bias", "symbol_theme_exposure"} +) +PUBLICATION_BIAS_KEYS = frozenset({"bias", "confidence", "rationale", "horizon", "risk_flags", "linked_themes"}) +PUBLICATION_EVIDENCE_KEYS = frozenset({"sources", "summary", "data_gaps"}) +PUBLICATION_POLICY_KEYS = frozenset({"execution_allowed", "portfolio_allocation_allowed", "downstream_use"}) def _require_mapping(value: Any, name: str) -> Mapping[str, Any]: @@ -153,6 +159,64 @@ def _validate_bias_mapping(mapping: Mapping[str, Any], name: str) -> None: _validate_bias_value(bias, f"{name}[{key!r}]") +def validate_signal_for_publication(payload: Mapping[str, Any]) -> None: + """Check the consumer-compatible structure without authenticating a producer run. + + The default reader remains compatible with retained examples and replay. + Freshness relative to a report cutoff is a separate consumer decision. + """ + payload = _require_mapping(payload, "signal") + if set(payload) - PUBLICATION_ALLOWED_KEYS: + raise SignalValidationError("signal contains unknown publication fields") + validate_signal(payload) + if payload["schema_version"] not in SUPPORTED_SCHEMA_VERSIONS: + raise SignalValidationError("unsupported publication schema_version") + if payload["horizon"] != REQUIRED_SIGNAL_HORIZON or payload["regime"] not in ALLOWED_REGIMES: + raise SignalValidationError("invalid publication horizon or regime") + for field in ("model_version", "scoring_version"): + if field in payload: + _require_string(payload[field], field) + + for field in ("candidate_bias", "theme_bias", "symbol_bias"): + for bias in payload.get(field, {}).values(): + if isinstance(bias, Mapping): + if set(bias) - PUBLICATION_BIAS_KEYS: + raise SignalValidationError("bias contains unknown publication fields") + if bias["bias"] not in ALLOWED_BIAS_VALUES: + raise SignalValidationError("invalid publication bias") + evidence = payload["evidence"] + if set(evidence) - PUBLICATION_EVIDENCE_KEYS: + raise SignalValidationError("evidence contains unknown publication fields") + _require_string_list(evidence.get("data_gaps", []), "evidence.data_gaps", allow_empty=True) + policy = payload["policy"] + if set(policy) - PUBLICATION_POLICY_KEYS or policy.get("portfolio_allocation_allowed", False) is not False: + raise SignalValidationError("invalid publication policy fields") + downstream = policy["downstream_use"].lower() + prohibited = ("allocation", "broker", "execution", "order", "position", "account") + blocking = ("blocked", "not allowed", "do not", "never", "no ") + if ( + not any(term in downstream for term in ("research", "advisory", "shadow")) + or "live" in downstream + or (any(term in downstream for term in prohibited) and not any(term in downstream for term in blocking)) + ): + raise SignalValidationError("publication policy must retain research-only restrictions") + + try: + source_date = dt.date.fromisoformat(payload["as_of"]) + text = payload["generated_at"] + generated = dt.datetime.fromisoformat(text[:-1] + "+00:00" if text.endswith("Z") else text) + expiry_date = dt.date.fromisoformat(payload["expires_at"]) + except (TypeError, ValueError) as exc: + raise SignalValidationError("invalid publication dates") from exc + if generated.tzinfo is None or generated.utcoffset() is None: + raise SignalValidationError("publication generated_at must be timezone-aware") + if generated.date() < source_date: + raise SignalValidationError("publication generated_at precedes as_of") + expiry = dt.datetime.combine(expiry_date, dt.time(23, 59, 59), tzinfo=generated.tzinfo) + if expiry < generated: + raise SignalValidationError("publication expires_at precedes generated_at") + + def _validate_bias_value(value: Any, name: str) -> None: if isinstance(value, str): bias = value diff --git a/tests/test_signal_validation.py b/tests/test_signal_validation.py index 35d5c6b..2d92141 100644 --- a/tests/test_signal_validation.py +++ b/tests/test_signal_validation.py @@ -4,6 +4,8 @@ import hashlib import json from pathlib import Path +import subprocess +import sys import pytest @@ -263,3 +265,307 @@ def test_available_at_present_but_invalid_is_rejected() -> None: with pytest.raises(SignalValidationError, match="available_at"): validate_signal(payload) + + +@pytest.mark.parametrize("location", ["root", "artifact", "producer", "policy"]) +def test_manifest_v2_rejects_unknown_fields(tmp_path: Path, location: str) -> None: + signal_path, manifest = _write_v2_artifact_pair(tmp_path) + target = manifest if location == "root" else manifest[location] + target["unexpected"] = "not part of the contract" + with pytest.raises(SignalValidationError, match="keys|fields"): + signal_validator.validate_manifest_v2(signal_path, load_example(), manifest) + + +@pytest.mark.parametrize("version", [2.0, "2", True]) +def test_manifest_v2_requires_integer_version(tmp_path: Path, version: object) -> None: + signal_path, manifest = _write_v2_artifact_pair(tmp_path) + manifest["schema_version"] = version + with pytest.raises(SignalValidationError): + signal_validator.validate_manifest_v2(signal_path, load_example(), manifest) + + +def _publication_signal() -> dict: + signal = load_example() + signal["policy"] = { + "execution_allowed": False, + "portfolio_allocation_allowed": False, + "downstream_use": "Research-only context; do not route to broker execution.", + } + return signal + + +def _write_publication_pair(directory: Path, producer: str = "a" * 40) -> tuple[Path, Path]: + directory.mkdir(parents=True, exist_ok=True) + signal = _publication_signal() + signal_path = directory / "latest_signal.json" + raw = json.dumps(signal, sort_keys=True).encode() + signal_path.write_bytes(raw) + manifest = { + "manifest_type": "research_signal_context", + "schema_version": 2, + "artifact": {"path": "data/output/latest_signal.json", "sha256": hashlib.sha256(raw).hexdigest()}, + **{key: signal[key] for key in ("as_of", "generated_at", "expires_at", "mode")}, + "producer": {"repository": "QuantStrategyLab/ResearchSignalContextPipelines", "commit_sha": producer}, + "input_digest": "sha256:" + "b" * 64, + "policy": {"execution_allowed": False}, + } + manifest_path = signal_path.with_suffix(".manifest.json") + manifest_path.write_text(json.dumps(manifest, sort_keys=True)) + return signal_path, manifest_path + + +def _validator_cli(*args: str, cwd: Path | None = None) -> subprocess.CompletedProcess: + return subprocess.run( + [sys.executable, str(ROOT / "scripts/validate_latest_signal.py"), *args], + cwd=cwd, text=True, capture_output=True, check=False, + ) + + +def test_strict_pair_reports_structure_only(tmp_path: Path) -> None: + signal_path, _ = _write_publication_pair(tmp_path) + result = _validator_cli(str(signal_path), "--require-manifest") + assert result.returncode == 0, result.stderr + assert "PAIR_STRUCTURE_VALID" in result.stdout + assert "context_provenance=UNVERIFIED" in result.stdout + assert "producer_run_provenance=UNVERIFIED" in result.stdout + + +def test_strict_pair_accepts_signal_v2_metadata(tmp_path: Path) -> None: + signal_path, manifest_path = _write_publication_pair(tmp_path) + signal = json.loads(signal_path.read_text()) + signal.update(schema_version="2", model_version="synthetic-v2", scoring_version="synthetic-rules") + raw = json.dumps(signal, sort_keys=True).encode() + signal_path.write_bytes(raw) + manifest = json.loads(manifest_path.read_text()) + manifest["artifact"]["sha256"] = hashlib.sha256(raw).hexdigest() + manifest_path.write_text(json.dumps(manifest)) + result = _validator_cli(str(signal_path), "--require-manifest") + assert result.returncode == 0, result.stderr + + +def test_strict_pair_cannot_bypass_missing_manifest(tmp_path: Path) -> None: + signal_path, manifest_path = _write_publication_pair(tmp_path) + manifest_path.unlink() + assert _validator_cli(str(signal_path), "--require-manifest").returncode != 0 + assert _validator_cli(str(signal_path), "--require-manifest", "--allow-missing").returncode != 0 + assert _validator_cli(str(signal_path)).returncode == 0 + + +def test_manifest_binds_the_supplied_payload_to_exact_file_bytes(tmp_path: Path) -> None: + signal_path, manifest = _write_v2_artifact_pair(tmp_path) + other = load_example() + other["confidence"] = 0.01 + with pytest.raises(SignalValidationError, match="signal bytes|payload"): + signal_validator.validate_manifest_v2(signal_path, other, manifest) + + +@pytest.mark.parametrize( + ("field", "value"), + [ + ("unknown", True), + ("available_at", "2026-05-28T22:00:00Z"), + ("generated_at", "2026-05-28T12:00:00"), + ("generated_at", "2020-01-01T12:00:00Z"), + ("expires_at", "2020-01-01"), + ("confidence", True), + ("confidence", float("nan")), + ("confidence", float("inf")), + ], +) +def test_strict_publication_rejects_unsafe_shape_or_time(field: str, value: object) -> None: + signal = _publication_signal() + signal[field] = value + with pytest.raises(SignalValidationError): + signal_validator.validate_signal_for_publication(signal) + + +@pytest.mark.parametrize("location", ["evidence", "policy", "bias"]) +def test_strict_publication_rejects_unknown_nested_fields(location: str) -> None: + signal = _publication_signal() + if location == "bias": + signal["candidate_bias"] = {"TEST": {"bias": "watch", "unknown": True}} + else: + signal[location]["unknown"] = True + with pytest.raises(SignalValidationError): + signal_validator.validate_signal_for_publication(signal) + + +def test_strict_publication_rejects_non_string_gaps_and_unsafe_policy() -> None: + signal = _publication_signal() + signal["evidence"]["data_gaps"] = [123] + with pytest.raises(SignalValidationError): + signal_validator.validate_signal_for_publication(signal) + signal = _publication_signal() + signal["policy"]["downstream_use"] = "Live account orders" + with pytest.raises(SignalValidationError): + signal_validator.validate_signal_for_publication(signal) + + +def test_strict_pair_rejects_duplicate_keys(tmp_path: Path) -> None: + signal_path, manifest_path = _write_publication_pair(tmp_path) + raw = signal_path.read_bytes() + duplicated = b'{"confidence":0.01,' + raw[1:] + signal_path.write_bytes(duplicated) + manifest = json.loads(manifest_path.read_text()) + manifest["artifact"]["sha256"] = hashlib.sha256(duplicated).hexdigest() + manifest_path.write_text(json.dumps(manifest)) + assert _validator_cli(str(signal_path), "--require-manifest").returncode != 0 + + +def test_strict_pair_rejects_duplicate_manifest_keys(tmp_path: Path) -> None: + signal_path, manifest_path = _write_publication_pair(tmp_path) + raw = manifest_path.read_bytes() + manifest_path.write_bytes(b'{"schema_version":1,' + raw[1:]) + assert _validator_cli(str(signal_path), "--require-manifest").returncode != 0 + + +def _git(repo: Path, *args: str) -> str: + result = subprocess.run(["git", "-C", str(repo), *args], text=True, capture_output=True, check=True) + return result.stdout.strip() + + +def _commit(repo: Path, message: str) -> str: + _git(repo, "add", "-A") + _git(repo, "commit", "-q", "-m", message) + return _git(repo, "rev-parse", "HEAD") + + +def _pair_repository(tmp_path: Path) -> tuple[Path, str]: + repo = tmp_path / "repo" + repo.mkdir() + _git(repo, "init", "-q") + _git(repo, "config", "user.name", "Synthetic fixture") + _git(repo, "config", "user.email", "fixture@example.invalid") + live = repo / "data/output" + live.mkdir(parents=True) + (live / "latest_signal.json").write_text('{"retained_legacy":true}') + (live / "latest_signal.manifest.json").write_text('{"schema_version":"1"}') + (repo / "README.md").write_text("synthetic repository\n") + return repo, _commit(repo, "synthetic baseline") + + +def _changed_pair(repo: Path, base: str, head: str, event: str = "pull_request") -> subprocess.CompletedProcess: + return _validator_cli("--changed-pair", "--event", event, "--base-sha", base, "--head-sha", head, cwd=repo) + + +@pytest.mark.parametrize("event", ["pull_request", "push"]) +def test_changed_pair_validates_exact_head_objects(tmp_path: Path, event: str) -> None: + repo, base = _pair_repository(tmp_path) + _write_publication_pair(repo / "data/output", base) + head = _commit(repo, "valid synthetic pair") + result = _changed_pair(repo, base, head, event) + assert result.returncode == 0, result.stderr + assert "PAIR_STRUCTURE_VALID" in result.stdout + + +def test_unrelated_change_does_not_relabel_or_block_retained_legacy_pair(tmp_path: Path) -> None: + repo, base = _pair_repository(tmp_path) + (repo / "README.md").write_text("unrelated change\n") + head = _commit(repo, "unrelated") + result = _changed_pair(repo, base, head) + assert result.returncode == 0, result.stderr + assert "NO_LIVE_PAIR_CHANGE" in result.stdout + assert "PAIR_STRUCTURE_VALID" not in result.stdout + + +@pytest.mark.parametrize("action", ["signal_only", "manifest_only", "delete", "rename"]) +def test_changed_pair_rejects_one_side_deletion_or_rename(tmp_path: Path, action: str) -> None: + repo, base = _pair_repository(tmp_path) + live = repo / "data/output" + if action == "signal_only": + (live / "latest_signal.json").write_text(json.dumps(_publication_signal())) + elif action == "manifest_only": + (live / "latest_signal.manifest.json").write_text('{"schema_version":2}') + elif action == "delete": + (live / "latest_signal.json").unlink() + (live / "latest_signal.manifest.json").unlink() + else: + (live / "latest_signal.json").rename(live / "renamed.json") + head = _commit(repo, action) + assert _changed_pair(repo, base, head).returncode != 0 + + +def test_worktree_replacement_cannot_rescue_invalid_committed_pair(tmp_path: Path) -> None: + repo, base = _pair_repository(tmp_path) + signal_path, manifest_path = _write_publication_pair(repo / "data/output", base) + manifest = json.loads(manifest_path.read_text()) + manifest["artifact"]["sha256"] = "0" * 64 + manifest_path.write_text(json.dumps(manifest)) + head = _commit(repo, "invalid committed pair") + _write_publication_pair(repo / "data/output", base) + assert _changed_pair(repo, base, head).returncode != 0 + + +def test_changed_pair_rejects_worktree_mutation_after_valid_commit(tmp_path: Path) -> None: + repo, base = _pair_repository(tmp_path) + signal_path, _ = _write_publication_pair(repo / "data/output", base) + head = _commit(repo, "valid pair") + signal_path.write_bytes(signal_path.read_bytes() + b"\n") + assert _changed_pair(repo, base, head).returncode != 0 + + +def test_pr_uses_merge_base_but_push_uses_exact_before(tmp_path: Path) -> None: + repo, common = _pair_repository(tmp_path) + _git(repo, "checkout", "-q", "-b", "target") + _write_publication_pair(repo / "data/output", common) + target_head = _commit(repo, "target changed its pair") + _git(repo, "checkout", "-q", "-b", "feature", common) + (repo / "README.md").write_text("feature changes no live inputs\n") + feature_head = _commit(repo, "feature documentation") + result = _changed_pair(repo, target_head, feature_head, "pull_request") + assert result.returncode == 0, result.stderr + assert "NO_LIVE_PAIR_CHANGE" in result.stdout + assert _changed_pair(repo, target_head, feature_head, "push").returncode != 0 + + +def test_changed_pair_rejects_regular_file_replaced_by_symlink(tmp_path: Path) -> None: + repo, base = _pair_repository(tmp_path) + signal_path, _ = _write_publication_pair(repo / "data/output", base) + raw = signal_path.read_bytes() + saved = signal_path.with_name("saved.json") + saved.write_bytes(raw) + signal_path.unlink() + signal_path.symlink_to("saved.json") + head = _commit(repo, "symlink pair") + assert _changed_pair(repo, base, head).returncode != 0 + + +def test_changed_pair_rejects_invalid_or_missing_commit(tmp_path: Path) -> None: + repo, base = _pair_repository(tmp_path) + assert _changed_pair(repo, "main", base).returncode != 0 + assert _changed_pair(repo, "f" * 40, base).returncode != 0 + + +def test_changed_pair_rejects_unresolved_producer_lineage(tmp_path: Path) -> None: + repo, base = _pair_repository(tmp_path) + _write_publication_pair(repo / "data/output", "f" * 40) + head = _commit(repo, "unknown producer") + assert _changed_pair(repo, base, head).returncode != 0 + + +def test_changed_pair_rejects_existing_nonancestor_producer(tmp_path: Path) -> None: + repo, base = _pair_repository(tmp_path) + _git(repo, "checkout", "-q", "-b", "producer-side") + (repo / "README.md").write_text("sibling history\n") + producer = _commit(repo, "sibling producer declaration") + _git(repo, "checkout", "-q", "-b", "publication", base) + _write_publication_pair(repo / "data/output", producer) + head = _commit(repo, "not descended from producer") + assert _changed_pair(repo, base, head).returncode != 0 + + +def test_changed_pair_rejects_missing_comparison_arguments(tmp_path: Path) -> None: + repo, _ = _pair_repository(tmp_path) + assert _validator_cli("--changed-pair", cwd=repo).returncode != 0 + + +def test_ci_gates_changed_pair_in_existing_read_only_test_job() -> None: + workflow = (ROOT / ".github/workflows/ci.yml").read_text() + assert "--changed-pair" in workflow + assert "fetch-depth: 0" in workflow + assert "github.event.pull_request.base.sha" in workflow + assert "github.event.pull_request.head.sha" in workflow + assert "github.event.before" in workflow + assert "contents: read" in workflow + assert "pull-requests: write" not in workflow + assert "python -m pytest -q" in workflow