diff --git a/.github/workflows/qar_vnext_d3_daily_preview_clock_binding.yml b/.github/workflows/qar_vnext_d3_daily_preview_clock_binding.yml new file mode 100644 index 0000000..29a73f4 --- /dev/null +++ b/.github/workflows/qar_vnext_d3_daily_preview_clock_binding.yml @@ -0,0 +1,77 @@ +name: QAR vNext D3 Frozen Clock Preview Evidence + +on: + pull_request: + paths: + - ".github/workflows/qar_vnext_d3_daily_preview_clock_binding.yml" + - "scripts/d3_clock_bound_build.py" + - "scripts/d3_clock_bound_verify.py" + - "src/quant_advisor_research/**" + - "examples/political_events.example.csv" + - "examples/political_watchlist.example.csv" + - "pyproject.toml" + - "uv.lock" + - "tests/test_d3_clock_binding.py" + - "tests/test_preview_bundle.py" + workflow_dispatch: + inputs: + as_of: + description: "Representative daily report date" + required: true + default: "2026-06-20" + type: string + +permissions: + contents: read + +jobs: + build-and-verify: + runs-on: ubuntu-latest + timeout-minutes: 15 + steps: + - name: Checkout repository + uses: actions/checkout@v6 + with: + fetch-depth: 0 + - name: Setup Python + uses: actions/setup-python@v6 + with: + python-version: "3.11" + - name: Install advisor package + run: python -m pip install -e . + - name: Build with exact frozen clock binding + env: + AS_OF: ${{ inputs.as_of || '2026-06-20' }} + BASE_SHA: ${{ github.event.pull_request.base.sha || github.sha }} + run: | + set -euo pipefail + python scripts/d3_clock_bound_build.py \ + --as-of "${AS_OF}" \ + --political-events examples/political_events.example.csv \ + --political-watchlist examples/political_watchlist.example.csv \ + --artifact-dir "${RUNNER_TEMP}/qar-daily-preview" \ + --evidence-path "${RUNNER_TEMP}/qar-daily-preview-build-evidence.json" \ + --base-sha "${BASE_SHA}" \ + --frozen-generated-at "2026-07-15T00:00:00Z" + - name: Upload daily preview artifact + uses: actions/upload-artifact@v7 + with: + name: qar-daily-preview-clock-binding + path: ${{ runner.temp }}/qar-daily-preview + if-no-files-found: error + - name: Download daily preview artifact + uses: actions/download-artifact@v7 + with: + name: qar-daily-preview-clock-binding + path: ${{ runner.temp }}/qar-daily-preview-downloaded + - name: Verify exact frozen clock binding + env: + BASE_SHA: ${{ github.event.pull_request.base.sha || github.sha }} + run: | + set -euo pipefail + python scripts/d3_clock_bound_verify.py \ + --artifact-dir "${RUNNER_TEMP}/qar-daily-preview-downloaded" \ + --build-evidence-path "${RUNNER_TEMP}/qar-daily-preview-build-evidence.json" \ + --evidence-path "${RUNNER_TEMP}/qar-daily-preview-download-evidence.json" \ + --base-sha "${BASE_SHA}" \ + --frozen-generated-at "2026-07-15T00:00:00Z" diff --git a/docs/qar_vnext_d3_frozen_clock_binding.md b/docs/qar_vnext_d3_frozen_clock_binding.md new file mode 100644 index 0000000..8a24072 --- /dev/null +++ b/docs/qar_vnext_d3_frozen_clock_binding.md @@ -0,0 +1,3 @@ +# QAR D3 frozen-clock evidence binding + +This isolated representative-fixture slice uses two independent daily producer calls under an explicit harness-frozen `generated_at`. The exact frozen value is checked against both report snapshots, both manifest source records, build evidence, and downloaded verification evidence. The artifact remains exactly `report.json`, `report.html`, and `manifest.json`; no production trust, Pages, publisher, weekly/monthly, legacy, or identity integration is added. diff --git a/scripts/d3_clock_bound_build.py b/scripts/d3_clock_bound_build.py new file mode 100644 index 0000000..2f107b1 --- /dev/null +++ b/scripts/d3_clock_bound_build.py @@ -0,0 +1,156 @@ +#!/usr/bin/env python3 +"""Build D3 representative evidence with an explicit harness-frozen clock.""" +from __future__ import annotations + +import argparse +import hashlib +import json +import re +import shutil +import sys +import tempfile +from pathlib import Path +from unittest.mock import patch + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) + +from quant_advisor_research import advisory_report +from quant_advisor_research.preview_bundle import PreviewBundleError, build_preview_bundle, read_preview_bundle + + +BASE_SHA_RE = re.compile(r"[0-9a-f]{40}") +FILES = ["manifest.json", "report.html", "report.json"] +BUNDLE = "qar.preview_bundle.v1" +SCHEMA = "5" +CONTRACT = "model_recommendations.v5" + + +def error(code: str) -> PreviewBundleError: + return PreviewBundleError(code) + + +def sha256(path: Path) -> str: + return hashlib.sha256(path.read_bytes()).hexdigest() + + +def require_exact_files(artifact_dir: Path) -> None: + if sorted(path.name for path in artifact_dir.iterdir()) != FILES or any( + not (artifact_dir / name).is_file() for name in FILES + ): + raise error("file_set_invalid") + + +def require_external_paths(artifact_dir: Path, *evidence_paths: Path) -> None: + try: + artifact_resolved = artifact_dir.resolve() + for evidence_path in evidence_paths: + evidence_resolved = evidence_path.resolve() + if evidence_resolved == artifact_resolved or artifact_resolved in evidence_resolved.parents: + raise error("evidence_path_inside_artifact") + except PreviewBundleError: + raise + except (OSError, RuntimeError, ValueError): + raise error("evidence_path_invalid") from None + + +def contract_source(report: dict[str, object], manifest: dict[str, object], frozen: str) -> dict[str, object]: + source = manifest.get("source") + expected = { + "schema_version": SCHEMA, + "contract_version": CONTRACT, + "cadence": "daily", + "as_of": report.get("as_of"), + "generated_at": frozen, + } + if ( + report.get("schema_version") != SCHEMA + or report.get("cadence") != "daily" + or report.get("generated_at") != frozen + or manifest.get("bundle_contract") != BUNDLE + or not isinstance(source, dict) + or source != expected + ): + raise error("contract_or_clock_drift") + return source + + +def build(args: argparse.Namespace) -> None: + if BASE_SHA_RE.fullmatch(args.base_sha) is None or not args.frozen_generated_at: + raise error("input_invalid") + events = Path(args.political_events) + watchlist = Path(args.political_watchlist) + output = Path(args.artifact_dir) + evidence_path = Path(args.evidence_path) + require_external_paths(output, evidence_path) + if not events.is_file() or not watchlist.is_file(): + raise error("fixture_missing") + repeat_parent = Path(tempfile.mkdtemp(prefix=f".{output.name}.repeat-", dir=output.parent)) + try: + repeat_output = repeat_parent / "preview" + with patch.object(advisory_report, "utc_now_iso", return_value=args.frozen_generated_at): + first = advisory_report.build_advisory_report( + as_of=args.as_of, cadence="daily", political_events_path=events, political_watchlist_path=watchlist + ) + build_preview_bundle(first, output) + first_readback = read_preview_bundle(output) + second = advisory_report.build_advisory_report( + as_of=args.as_of, cadence="daily", political_events_path=events, political_watchlist_path=watchlist + ) + build_preview_bundle(second, repeat_output) + second_readback = read_preview_bundle(repeat_output) + first_report = dict(first_readback.report) + second_report = dict(second_readback.report) + first_manifest = dict(first_readback.manifest) + second_manifest = dict(second_readback.manifest) + first_source = contract_source(first_report, first_manifest, args.frozen_generated_at) + second_source = contract_source(second_report, second_manifest, args.frozen_generated_at) + if first_source != second_source: + raise error("producer_metadata_mismatch") + if any((output / name).read_bytes() != (repeat_output / name).read_bytes() for name in FILES): + raise error("repeat_build_non_deterministic") + payload = { + "source_kind": "repository_representative_fixture", + "base_sha": args.base_sha, + "bundle_contract": BUNDLE, + "frozen_generated_at": args.frozen_generated_at, + "producer_generated_at_values": [first_report["generated_at"], second_report["generated_at"]], + "report_generated_at": first_report["generated_at"], + "manifest_generated_at_values": [first_source["generated_at"], second_source["generated_at"]], + "source": {key: first_report[key] for key in ("cadence", "as_of", "generated_at", "schema_version")}, + "manifest_source": first_source, + "files": sorted(path.name for path in output.iterdir()), + "sha256": {name: sha256(output / name) for name in FILES}, + "repeat_build_bytes": True, + "deterministic_clock": {"mode": "frozen_harness", "generated_at": args.frozen_generated_at}, + } + finally: + shutil.rmtree(repeat_parent, ignore_errors=True) + require_exact_files(output) + if evidence_path.exists(): + raise error("evidence_exists") + evidence_path.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") + require_exact_files(output) + print(json.dumps(payload, ensure_ascii=False, sort_keys=True)) + + +def main() -> None: + parser = argparse.ArgumentParser() + parser.add_argument("--as-of", required=True) + parser.add_argument("--political-events", required=True) + parser.add_argument("--political-watchlist", required=True) + parser.add_argument("--artifact-dir", required=True) + parser.add_argument("--evidence-path", required=True) + parser.add_argument("--base-sha", required=True) + parser.add_argument("--frozen-generated-at", required=True) + try: + build(parser.parse_args()) + except PreviewBundleError as exc: + print(f"d3_clock_build_failed:{exc.code}", file=sys.stderr) + raise SystemExit(1) from None + except (OSError, TypeError, ValueError, UnicodeError): + print("d3_clock_build_failed", file=sys.stderr) + raise SystemExit(1) from None + + +if __name__ == "__main__": + main() diff --git a/scripts/d3_clock_bound_verify.py b/scripts/d3_clock_bound_verify.py new file mode 100644 index 0000000..1e8eb04 --- /dev/null +++ b/scripts/d3_clock_bound_verify.py @@ -0,0 +1,153 @@ +#!/usr/bin/env python3 +"""Verify downloaded D3 bytes and exact frozen-clock/build-evidence binding.""" +from __future__ import annotations + +import argparse +import hashlib +import json +import re +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) + +from quant_advisor_research.preview_bundle import PreviewBundleError, read_preview_bundle + + +BASE_SHA_RE = re.compile(r"[0-9a-f]{40}") +FILES = ["manifest.json", "report.html", "report.json"] +BUNDLE = "qar.preview_bundle.v1" +SCHEMA = "5" +CONTRACT = "model_recommendations.v5" + + +def error(code: str) -> PreviewBundleError: + return PreviewBundleError(code) + + +def require_exact_files(artifact_dir: Path) -> None: + if sorted(path.name for path in artifact_dir.iterdir()) != FILES or any( + not (artifact_dir / name).is_file() for name in FILES + ): + raise error("file_set_invalid") + + +def require_external_paths(artifact_dir: Path, *evidence_paths: Path) -> None: + try: + artifact_resolved = artifact_dir.resolve() + for evidence_path in evidence_paths: + evidence_resolved = evidence_path.resolve() + if evidence_resolved == artifact_resolved or artifact_resolved in evidence_resolved.parents: + raise error("evidence_path_inside_artifact") + except PreviewBundleError: + raise + except (OSError, RuntimeError, ValueError): + raise error("evidence_path_invalid") from None + + +def canonical(value: object) -> bytes: + try: + return json.dumps( + value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False + ).encode("utf-8") + except (TypeError, ValueError, UnicodeError, RecursionError): + raise error("manifest_canonical_invalid") from None + + +def verify(args: argparse.Namespace) -> None: + if BASE_SHA_RE.fullmatch(args.base_sha) is None or not args.frozen_generated_at: + raise error("input_invalid") + output = Path(args.artifact_dir) + build_evidence_path = Path(args.build_evidence_path) + evidence_path = Path(args.evidence_path) + require_external_paths(output, build_evidence_path, evidence_path) + if sorted(path.name for path in output.iterdir()) != FILES: + raise error("file_set_invalid") + raw_manifest = (output / "manifest.json").read_bytes() + try: + parsed_manifest = json.loads(raw_manifest.decode("utf-8")) + except (TypeError, ValueError, UnicodeError, RecursionError): + raise error("manifest_canonical_invalid") from None + if raw_manifest != canonical(parsed_manifest): + raise error("manifest_noncanonical") + readback = read_preview_bundle(output) + report = dict(readback.report) + manifest = dict(readback.manifest) + source = manifest.get("source") + expected_source = { + "schema_version": SCHEMA, + "contract_version": CONTRACT, + "cadence": "daily", + "as_of": report.get("as_of"), + "generated_at": args.frozen_generated_at, + } + if ( + report.get("schema_version") != SCHEMA + or report.get("cadence") != "daily" + or report.get("generated_at") != args.frozen_generated_at + or manifest.get("bundle_contract") != BUNDLE + or not isinstance(source, dict) + or source != expected_source + ): + raise error("contract_or_clock_drift") + try: + build_evidence = json.loads(build_evidence_path.read_text(encoding="utf-8")) + except (OSError, TypeError, ValueError, UnicodeError, RecursionError): + raise error("build_evidence_invalid") from None + actual_hashes = {name: hashlib.sha256((output / name).read_bytes()).hexdigest() for name in FILES} + expected_build_evidence = { + "source_kind": "repository_representative_fixture", + "base_sha": args.base_sha, + "bundle_contract": BUNDLE, + "frozen_generated_at": args.frozen_generated_at, + "producer_generated_at_values": [args.frozen_generated_at, args.frozen_generated_at], + "report_generated_at": args.frozen_generated_at, + "manifest_generated_at_values": [args.frozen_generated_at, args.frozen_generated_at], + "source": { + "cadence": "daily", + "as_of": report.get("as_of"), + "generated_at": args.frozen_generated_at, + "schema_version": SCHEMA, + }, + "manifest_source": source, + "files": FILES, + "sha256": actual_hashes, + "repeat_build_bytes": True, + "deterministic_clock": {"mode": "frozen_harness", "generated_at": args.frozen_generated_at}, + } + if build_evidence != expected_build_evidence: + raise error("build_evidence_mismatch") + payload = { + "source_kind": "downloaded_repository_representative_fixture", + "base_sha": args.base_sha, + "bundle_contract": BUNDLE, + "frozen_generated_at": args.frozen_generated_at, + "source": source, + "files": FILES, + "sha256": actual_hashes, + "manifest_canonical_bytes": True, + "build_evidence_bound": True, + } + if evidence_path.exists(): + raise error("evidence_exists") + evidence_path.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") + require_exact_files(output) + print(json.dumps(payload, ensure_ascii=False, sort_keys=True)) + + +def main() -> None: + parser = argparse.ArgumentParser() + parser.add_argument("--artifact-dir", required=True) + parser.add_argument("--build-evidence-path", required=True) + parser.add_argument("--evidence-path", required=True) + parser.add_argument("--base-sha", required=True) + parser.add_argument("--frozen-generated-at", required=True) + try: + verify(parser.parse_args()) + except (PreviewBundleError, OSError, TypeError, ValueError, UnicodeError): + print("d3_clock_readback_failed", file=sys.stderr) + raise SystemExit(1) from None + + +if __name__ == "__main__": + main() diff --git a/tests/test_d3_clock_binding.py b/tests/test_d3_clock_binding.py new file mode 100644 index 0000000..f0dddd5 --- /dev/null +++ b/tests/test_d3_clock_binding.py @@ -0,0 +1,137 @@ +from __future__ import annotations + +import json +import subprocess +import sys +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[1] +BUILD = ROOT / "scripts/d3_clock_bound_build.py" +VERIFY = ROOT / "scripts/d3_clock_bound_verify.py" +BASE_SHA = "c5ba801a8696eec63c7ba348f3f125cb52cd06ff" +FROZEN = "2026-07-15T00:00:00Z" + + +def run_script(script: Path, *args: str) -> subprocess.CompletedProcess[str]: + return subprocess.run([sys.executable, str(script), *args], cwd=ROOT, text=True, capture_output=True, check=False) + + +def build_args(output: Path, evidence: Path) -> tuple[str, ...]: + return ( + "--as-of", "2026-06-20", + "--political-events", str(ROOT / "examples/political_events.example.csv"), + "--political-watchlist", str(ROOT / "examples/political_watchlist.example.csv"), + "--artifact-dir", str(output), "--evidence-path", str(evidence), + "--base-sha", BASE_SHA, "--frozen-generated-at", FROZEN, + ) + + +def test_build_binds_cli_frozen_clock_to_both_independent_outputs(tmp_path): + output = tmp_path / "preview" + evidence = tmp_path / "build-evidence.json" + result = run_script(BUILD, *build_args(output, evidence)) + + assert result.returncode == 0, result.stderr + payload = json.loads(evidence.read_text(encoding="utf-8")) + assert payload["frozen_generated_at"] == FROZEN + assert payload["producer_generated_at_values"] == [FROZEN, FROZEN] + assert payload["manifest_generated_at_values"] == [FROZEN, FROZEN] + assert payload["report_generated_at"] == FROZEN + + +def test_verify_requires_exact_frozen_clock_binding(tmp_path): + output = tmp_path / "preview" + evidence = tmp_path / "build-evidence.json" + assert run_script(BUILD, *build_args(output, evidence)).returncode == 0 + payload = json.loads(evidence.read_text(encoding="utf-8")) + payload["frozen_generated_at"] = "2026-07-15T00:00:01Z" + evidence.write_text(json.dumps(payload), encoding="utf-8") + + result = run_script( + VERIFY, "--artifact-dir", str(output), "--build-evidence-path", str(evidence), + "--evidence-path", str(tmp_path / "download-evidence.json"), "--base-sha", BASE_SHA, + "--frozen-generated-at", FROZEN, + ) + + assert result.returncode != 0 + assert "readback_failed" in result.stderr + + +def test_verify_rejects_manifest_clock_drift(tmp_path): + output = tmp_path / "preview" + evidence = tmp_path / "build-evidence.json" + assert run_script(BUILD, *build_args(output, evidence)).returncode == 0 + manifest = json.loads((output / "manifest.json").read_text(encoding="utf-8")) + manifest["source"]["generated_at"] = "2026-07-15T00:00:01Z" + (output / "manifest.json").write_text(json.dumps(manifest), encoding="utf-8") + + result = run_script( + VERIFY, "--artifact-dir", str(output), "--build-evidence-path", str(evidence), + "--evidence-path", str(tmp_path / "download-evidence.json"), "--base-sha", BASE_SHA, + "--frozen-generated-at", FROZEN, + ) + + assert result.returncode != 0 + assert "readback_failed" in result.stderr + + +def test_verify_rejects_every_build_evidence_payload_shape_tamper(tmp_path): + output = tmp_path / "preview" + evidence = tmp_path / "build-evidence.json" + assert run_script(BUILD, *build_args(output, evidence)).returncode == 0 + original = json.loads(evidence.read_text(encoding="utf-8")) + + for key in original: + tampered = dict(original) + tampered.pop(key) + evidence.write_text(json.dumps(tampered), encoding="utf-8") + result = run_script( + VERIFY, "--artifact-dir", str(output), "--build-evidence-path", str(evidence), + "--evidence-path", str(tmp_path / f"download-{key}.json"), "--base-sha", BASE_SHA, + "--frozen-generated-at", FROZEN, + ) + assert result.returncode != 0 + evidence.write_text(json.dumps(original), encoding="utf-8") + + tampered = dict(original) + tampered["unknown"] = "reject" + evidence.write_text(json.dumps(tampered), encoding="utf-8") + result = run_script( + VERIFY, "--artifact-dir", str(output), "--build-evidence-path", str(evidence), + "--evidence-path", str(tmp_path / "download-unknown.json"), "--base-sha", BASE_SHA, + "--frozen-generated-at", FROZEN, + ) + assert result.returncode != 0 + + +def test_build_rejects_evidence_path_inside_artifact_before_writing(tmp_path): + output = tmp_path / "preview" + result = run_script( + BUILD, + *build_args(output, output / "build-evidence.json"), + ) + assert result.returncode != 0 + assert not output.exists() + + +def test_verify_rejects_evidence_paths_inside_or_aliasing_artifact(tmp_path): + output = tmp_path / "preview" + evidence = tmp_path / "build-evidence.json" + assert run_script(BUILD, *build_args(output, evidence)).returncode == 0 + + inside = run_script( + VERIFY, "--artifact-dir", str(output), "--build-evidence-path", str(evidence), + "--evidence-path", str(output / "download.json"), "--base-sha", BASE_SHA, + "--frozen-generated-at", FROZEN, + ) + assert inside.returncode != 0 + + alias = tmp_path / "alias" + alias.symlink_to(output, target_is_directory=True) + aliased = run_script( + VERIFY, "--artifact-dir", str(output), "--build-evidence-path", str(alias / "build.json"), + "--evidence-path", str(tmp_path / "download.json"), "--base-sha", BASE_SHA, + "--frozen-generated-at", FROZEN, + ) + assert aliased.returncode != 0