feat: validate immutable signal provenance manifests

scripts/post_shadow_signal_request.py

@@ -4,9 +4,11 @@
 import argparse
 from collections.abc import Mapping
 import datetime as dt
+import hashlib
 import json
 import os
 from pathlib import Path
+import re
 import sys
 import urllib.error
 import urllib.parse
@@ -16,6 +18,7 @@
 
 DEFAULT_API_URL = "https://api.github.com"
 DEFAULT_LABEL = "long-horizon-shadow"
+COMMIT_SHA_PATTERN = re.compile(r"[0-9a-fA-F]{40}")
 
 
 def github_request(method: str, url: str, token: str, payload: dict[str, Any] | None = None) -> Any:
@@ -75,6 +78,15 @@ def load_context_bundle(path: str | None) -> dict[str, Any] | None:
     return json.loads(Path(path).read_text(encoding="utf-8"))
 
 
+def build_immutable_provenance(context_path: Path, producer_commit_sha: str) -> dict[str, str]:
+    if not COMMIT_SHA_PATTERN.fullmatch(producer_commit_sha):
+        raise ValueError("producer commit must be an immutable 40-hex commit")
+    return {
+        "producer_commit_sha": producer_commit_sha,
+        "input_digest": f"sha256:{hashlib.sha256(context_path.read_bytes()).hexdigest()}",
+    }
+
+
 def resolve_as_of_date(raw_as_of_date: str | None, context_bundle: Mapping[str, Any] | None) -> str:
     if raw_as_of_date:
         return raw_as_of_date
@@ -114,6 +126,7 @@ def build_issue_body(
     provider: str,
     bridge_repository: str,
     context_bundle: Mapping[str, Any] | None = None,
+    immutable_provenance: Mapping[str, str] | None = None,
 ) -> str:
     sections = [
         [
@@ -141,6 +154,41 @@ def build_issue_body(
             "",
         ],
         [context_markdown(context_bundle)],
+        [
+            "",
+            "## Immutable Provenance Contract",
+            "",
+            "Any new signal must update `latest_signal.json` and its sibling manifest in the same PR.",
+            "The manifest must use this v2 contract; v1 is legacy_untrusted and cannot authorize downstream use.",
+            "",
+            "```json",
+            json.dumps(
+                {
+                    "manifest_type": "research_signal_context",
+                    "schema_version": 2,
+                    "artifact": {
+                        "path": "data/output/latest_signal.json",
+                        "sha256": "sha256 of exact latest_signal.json bytes",
+                    },
+                    "as_of": "copy exactly from signal",
+                    "generated_at": "copy exactly from signal",
+                    "expires_at": "copy exactly from signal",
+                    "mode": "copy exactly from signal",
+                    "producer": {
+                        "repository": "QuantStrategyLab/ResearchSignalContextPipelines",
+                        "commit_sha": (immutable_provenance or {}).get("producer_commit_sha", "required 40-hex commit"),
+                    },
+                    "input_digest": (immutable_provenance or {}).get("input_digest", "required sha256 digest"),
+                    "policy": {"execution_allowed": False},
+                },
+                ensure_ascii=True,
+                indent=2,
+                sort_keys=True,
+            ),
+            "```",
+            "",
+            "Do not add a publisher or publication self-commit field. Do not upgrade existing v1 artifacts by hand.",
+        ],
         [
             "",
             "Do not infer historical AI signals that were not generated point-in-time.",
@@ -201,16 +249,28 @@ def main() -> int:
     if not token:
         print("GITHUB_TOKEN is required", file=sys.stderr)
         return 1
+    if not args.context_file:
+        print("immutable provenance requires --context-file", file=sys.stderr)
+        return 1
 
     context_bundle = load_context_bundle(args.context_file)
     as_of_date = resolve_as_of_date(args.as_of_date, context_bundle)
     title = build_issue_title(as_of_date)
+    try:
+        immutable_provenance = build_immutable_provenance(
+            Path(args.context_file),
+            os.environ.get("GITHUB_SHA", args.source_ref),
+        )
+    except (OSError, ValueError) as exc:
+        print(f"immutable provenance unavailable: {exc}", file=sys.stderr)
+        return 1
     body = build_issue_body(
         as_of_date=as_of_date,
         source_ref=args.source_ref,
         provider=args.provider,
         bridge_repository=args.bridge_repository,
         context_bundle=context_bundle,
+        immutable_provenance=immutable_provenance,
     )
     try:
         action, issue_number, issue_url = upsert_issue(

scripts/validate_latest_signal.py

@@ -2,9 +2,13 @@
 from __future__ import annotations
 
 import argparse
+from collections.abc import Mapping
+import hashlib
 import json
+import re
 import sys
 from pathlib import Path
+from typing import Any
 
 ROOT = Path(__file__).resolve().parents[1]
 sys.path.insert(0, str(ROOT / "src"))
@@ -13,6 +17,70 @@
 
 
 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}")
+SHA256_PATTERN = re.compile(r"[0-9a-f]{64}")
+
+
+def _require_mapping(value: Any, name: str) -> Mapping[str, Any]:
+    if not isinstance(value, Mapping):
+        raise SignalValidationError(f"{name} must be an object")
+    return value
+
+
+def _require_string(value: Any, name: str) -> str:
+    if not isinstance(value, str) or not value:
+        raise SignalValidationError(f"{name} must be a non-empty string")
+    return value
+
+
+def validate_manifest_v2(
+    signal_path: Path,
+    signal: Mapping[str, Any],
+    manifest: Mapping[str, Any],
+) -> None:
+    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):
+        raise SignalValidationError("manifest schema_version must be 2")
+    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")
+    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():
+        raise SignalValidationError("artifact.sha256 does not match signal bytes")
+
+    for field in ("as_of", "generated_at", "expires_at", "mode"):
+        if manifest.get(field) != signal.get(field):
+            raise SignalValidationError(f"{field} must exactly match the signal")
+
+    producer = _require_mapping(manifest.get("producer"), "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")
+    if not COMMIT_SHA_PATTERN.fullmatch(producer_commit_sha):
+        raise SignalValidationError("producer.commit_sha must be an immutable 40-hex commit")
+
+    input_digest = _require_string(manifest.get("input_digest"), "input_digest")
+    if not input_digest.startswith("sha256:") or not SHA256_PATTERN.fullmatch(input_digest.removeprefix("sha256:")):
+        raise SignalValidationError("input_digest must be sha256:<64 lowercase hex>")
+
+    policy = _require_mapping(manifest.get("policy"), "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 main() -> int:
@@ -31,10 +99,17 @@ def main() -> int:
     payload = json.loads(path.read_text(encoding="utf-8"))
     try:
         validate_signal(payload)
+        manifest_path = path.with_suffix(".manifest.json")
+        if manifest_path.exists():
+            manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
+            validate_manifest_v2(path, payload, manifest)
     except SignalValidationError as exc:
-        raise SystemExit(f"invalid signal artifact: {exc}") from exc
+        raise SystemExit(f"invalid or untrusted signal artifact: {exc}") from exc
 
-    print(f"valid signal artifact: {path}")
+    if path.with_suffix(".manifest.json").exists():
+        print(f"valid signal artifact with immutable provenance: {path}")
+    else:
+        print(f"valid signal schema only; immutable provenance unavailable: {path}")
     return 0
 
 

tests/test_signal_validation.py

@@ -1,11 +1,14 @@
 from __future__ import annotations
 
+import hashlib
 import json
 from pathlib import Path
 
 import pytest
 
 from research_signal_context_pipelines import SignalValidationError, validate_signal
+from scripts import post_shadow_signal_request as shadow_issue
+from scripts import validate_latest_signal as signal_validator
 
 
 ROOT = Path(__file__).resolve().parents[1]
@@ -137,3 +140,72 @@ def test_committed_latest_signal_covers_advisor_long_context() -> None:
     assert payload.get("symbol_theme_exposure")
     covered_symbols = set(payload.get("symbol_bias", {})) | set(payload.get("symbol_theme_exposure", {}))
     assert {"MU", "INTC", "AMD", "VRT", "DELL"} <= covered_symbols
+
+
+def _write_v2_artifact_pair(tmp_path: Path) -> tuple[Path, dict]:
+    signal_path = tmp_path / "latest_signal.json"
+    signal = load_example()
+    signal_bytes = json.dumps(signal, sort_keys=True).encode("utf-8")
+    signal_path.write_bytes(signal_bytes)
+    manifest = {
+        "manifest_type": "research_signal_context",
+        "schema_version": 2,
+        "artifact": {
+            "path": "data/output/latest_signal.json",
+            "sha256": hashlib.sha256(signal_bytes).hexdigest(),
+        },
+        "as_of": signal["as_of"],
+        "generated_at": signal["generated_at"],
+        "expires_at": signal["expires_at"],
+        "mode": signal["mode"],
+        "producer": {
+            "repository": "QuantStrategyLab/ResearchSignalContextPipelines",
+            "commit_sha": "a" * 40,
+        },
+        "input_digest": "sha256:" + "b" * 64,
+        "policy": {"execution_allowed": False},
+    }
+    return signal_path, manifest
+
+
+def test_manifest_v2_rejects_stale_signal_digest(tmp_path: Path) -> None:
+    signal_path, manifest = _write_v2_artifact_pair(tmp_path)
+    manifest["artifact"]["sha256"] = "0" * 64
+
+    with pytest.raises(SignalValidationError, match="artifact.sha256"):
+        signal_validator.validate_manifest_v2(signal_path, load_example(), manifest)
+
+
+def test_manifest_v2_rejects_missing_required_input_digest(tmp_path: Path) -> None:
+    signal_path, manifest = _write_v2_artifact_pair(tmp_path)
+    del manifest["input_digest"]
+
+    with pytest.raises(SignalValidationError, match="input_digest"):
+        signal_validator.validate_manifest_v2(signal_path, load_example(), manifest)
+
+
+def test_manifest_v2_rejects_mutable_producer_ref(tmp_path: Path) -> None:
+    signal_path, manifest = _write_v2_artifact_pair(tmp_path)
+    manifest["producer"]["commit_sha"] = "main"
+
+    with pytest.raises(SignalValidationError, match="producer.commit_sha"):
+        signal_validator.validate_manifest_v2(signal_path, load_example(), manifest)
+
+
+def test_manifest_v1_is_explicitly_legacy_untrusted(tmp_path: Path) -> None:
+    signal_path, _ = _write_v2_artifact_pair(tmp_path)
+
+    with pytest.raises(SignalValidationError, match="legacy_untrusted"):
+        signal_validator.validate_manifest_v2(signal_path, load_example(), {"schema_version": "1"})
+
+
+def test_shadow_request_binds_context_digest_to_immutable_commit(tmp_path: Path) -> None:
+    context_path = tmp_path / "context.json"
+    context_path.write_bytes(b'{"as_of":"2026-05-29"}')
+
+    provenance = shadow_issue.build_immutable_provenance(context_path, "c" * 40)
+
+    assert provenance == {
+        "producer_commit_sha": "c" * 40,
+        "input_digest": "sha256:" + hashlib.sha256(context_path.read_bytes()).hexdigest(),
+    }