diff --git a/.github/workflows/runtime-artifact-evidence-gate.yml b/.github/workflows/runtime-artifact-evidence-gate.yml new file mode 100644 index 0000000..88f906a --- /dev/null +++ b/.github/workflows/runtime-artifact-evidence-gate.yml @@ -0,0 +1,134 @@ +name: Runtime Artifact Evidence Gate + +on: + workflow_dispatch: + schedule: + # Verify immutable inputs before the main daily operating window. The + # individual publisher workflows retain responsibility for generating them. + - cron: "10 12 * * 1-5" + +permissions: + contents: read + id-token: write + issues: write + +concurrency: + group: runtime-artifact-evidence-gate + cancel-in-progress: false + +jobs: + verify-required-runtime-artifacts: + runs-on: ubuntu-latest + timeout-minutes: 15 + env: + GCP_PROJECT_ID: longbridgequant + GCP_WORKLOAD_IDENTITY_PROVIDER: projects/252919773759/locations/global/workloadIdentityPools/github-actions/providers/github-main + GCP_WORKLOAD_IDENTITY_SERVICE_ACCOUNT: qsl-artifact-evidence@longbridgequant.iam.gserviceaccount.com + steps: + - uses: actions/checkout@v6 + + - uses: actions/setup-python@v6 + with: + python-version: "3.12" + + - name: Authenticate read-only evidence identity + uses: google-github-actions/auth@7c6bc770dae815cd3e89ee6cdf493a5fab2cc093 # v3 + with: + workload_identity_provider: ${{ env.GCP_WORKLOAD_IDENTITY_PROVIDER }} + service_account: ${{ env.GCP_WORKLOAD_IDENTITY_SERVICE_ACCOUNT }} + + - name: Set up Google Cloud CLI + uses: google-github-actions/setup-gcloud@v3 + with: + project_id: ${{ env.GCP_PROJECT_ID }} + version: ">= 416.0.0" + + - name: Validate control-plane registry + run: | + set -euo pipefail + mkdir -p data/output/runtime-artifact-evidence + python3 python/scripts/build_config.py --check + python3 python/scripts/build_config.py --runtime-artifact-evidence-registry \ + > data/output/runtime-artifact-evidence/registry.json + + - name: Verify remote objects, manifests, digests, and freshness + id: verify + continue-on-error: true + run: | + set -euo pipefail + python3 python/scripts/verify_runtime_artifact_evidence.py \ + --registry data/output/runtime-artifact-evidence/registry.json \ + --output data/output/runtime-artifact-evidence/receipt.json + + - name: Summarize evidence receipt + if: always() + run: | + set -euo pipefail + python3 - <<'PY' >> "$GITHUB_STEP_SUMMARY" + import json + from pathlib import Path + + receipt_path = Path("data/output/runtime-artifact-evidence/receipt.json") + print("## Runtime Artifact Evidence Gate") + print() + if not receipt_path.exists(): + print("- receipt: `missing`") + raise SystemExit(0) + receipt = json.loads(receipt_path.read_text(encoding="utf-8")) + summary = receipt.get("summary", {}) + print(f"- status: `{receipt.get('status')}`") + print(f"- required artifacts: `{summary.get('required_artifact_count')}`") + print(f"- verified artifacts: `{summary.get('verified_artifact_count')}`") + print(f"- failed artifacts: `{summary.get('failed_artifact_count')}`") + print("- boundary: read-only; no publishing, runtime change, or order submission") + PY + + - name: Upload evidence receipt + if: always() + uses: actions/upload-artifact@v7 + with: + name: runtime-artifact-evidence-receipt + path: data/output/runtime-artifact-evidence/ + if-no-files-found: warn + retention-days: 30 + + - name: Create or update parked-evidence issue + if: always() && steps.verify.outcome == 'failure' + env: + GH_TOKEN: ${{ github.token }} + RUN_URL: ${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }} + run: | + set -euo pipefail + gh label create runtime-artifact-evidence \ + --description "Read-only required runtime artifact verification findings" \ + --color d93f0b \ + --force >/dev/null + body="$(mktemp)" + { + echo "## Runtime artifact evidence requires attention" + echo + echo "- Run: ${RUN_URL}" + echo "- This gate is read-only. It did not publish data, alter a runtime target, or submit an order." + echo "- Keep affected routes parked until the owning publisher emits a verified artifact and manifest." + echo + echo '```json' + if [ -s data/output/runtime-artifact-evidence/receipt.json ]; then + python3 -m json.tool data/output/runtime-artifact-evidence/receipt.json + else + echo '{"status":"receipt_missing"}' + fi + echo '```' + } > "$body" + existing="$(gh issue list --state open --label runtime-artifact-evidence --json number --jq '.[0].number // ""')" + if [ -n "$existing" ]; then + gh issue comment "$existing" --body-file "$body" + else + gh issue create \ + --title "Runtime artifact evidence requires attention" \ + --label runtime-artifact-evidence \ + --body-file "$body" + fi + + - name: Fail when external evidence is invalid + if: steps.verify.outcome == 'failure' + run: exit 1 diff --git a/docs/qsl_universal_dry_run_coverage_v1.zh-CN.md b/docs/qsl_universal_dry_run_coverage_v1.zh-CN.md index 1350536..7c1fcd3 100644 --- a/docs/qsl_universal_dry_run_coverage_v1.zh-CN.md +++ b/docs/qsl_universal_dry_run_coverage_v1.zh-CN.md @@ -42,11 +42,17 @@ | 层次 | 含义 | 当前结果 | | --- | --- | --- | | 已声明路径 | 策略域、策略许可和平台 `dry_run` 许可匹配 | 59 条 | -| 默认可构建路径 | 不需要临时人工补充制品，即可生成通过 runtime policy 校验的不下单目标 | 57 条 | +| 默认可构建路径 | 不需要临时人工补充制品，即可生成通过 runtime policy 校验的不下单目标 | 59 条 | 如果策略依赖 `runtime_artifacts.feature_snapshot.required=true`，默认可构建路径还必须配置成对的、`gs://` 开头的快照和 manifest URI。缺失时策略会被标成 **PARKED**，健康检查 `strategy_platform_dry_run_coverage` 以 critical failure 失败，执行保持关闭；这不是允许自动补一个猜测的 URI。 -当前受此规则约束的是 `hk_low_vol_dividend_quality_snapshot` 的 IBKR 与 LongBridge 两条已声明路径。它们需要由 [HkEquitySnapshotPipelines](https://github.com/QuantStrategyLab/HkEquitySnapshotPipelines) 发布并验证的 `feature_snapshot` 与 manifest 后才能成为默认可构建路线。该流水线的制品、来源、回测和人工审阅是独立证据边界；完成后可通过配置默认 URI，或在一次已审阅的不下单运行中成对传入两个 URI。自动修复只能发现、暂停、提示和复验，绝不伪造数据制品或提高生命周期。 +当前受此规则约束的是 `global_etf_rotation`、`russell_top50_leader_rotation` 与 `hk_low_vol_dividend_quality_snapshot`。前两者由 [UsEquitySnapshotPipelines](https://github.com/QuantStrategyLab/UsEquitySnapshotPipelines) 发布，港股由 [HkEquitySnapshotPipelines](https://github.com/QuantStrategyLab/HkEquitySnapshotPipelines) 发布；三者都必须先通过各自来源、回测和人工审阅边界，才可成为默认可构建路线。自动修复只能发现、暂停、提示和复验，绝不伪造数据制品或提高生命周期。 + +### 只读制品证据门 + +`Runtime Artifact Evidence Gate` 从同一份 `platform-config.json` 生成全部必需制品清单，并以专用只读身份逐项验证：对象可读、manifest 的策略身份、SHA-256 与快照文件一致、以及 `snapshot_as_of` 没有超过策略声明的时效预算。当前预算由发布节奏决定：日更全球 ETF 保留 5 个自然日以覆盖周末与一个市场假日；月更 Russell 与港股保留 40 个自然日。 + +该门只会生成回执和待处理事项。它不会发布数据、修改 URI、改变运行目标、提升生命周期或提交订单；异常路线必须继续保持 **PARKED**，直到制品所属流水线重新发布并通过验证。 ## 插件边界 @@ -62,6 +68,8 @@ python3 python/scripts/build_config.py --check python3 python/scripts/runtime_settings.py validate python3 python/scripts/build_config.py --platform-health-report +python3 python/scripts/build_config.py --runtime-artifact-evidence-registry > /tmp/runtime-artifact-registry.json +python3 python/scripts/verify_runtime_artifact_evidence.py --registry /tmp/runtime-artifact-registry.json node tests/strategy_switch_worker_validation.mjs ``` diff --git a/platform-config.json b/platform-config.json index 695636a..c1fa921 100644 --- a/platform-config.json +++ b/platform-config.json @@ -519,7 +519,8 @@ "feature_snapshot": { "required": true, "path": "gs://qsl-runtime-logs-shared/strategy-artifacts/us_equity/global_etf_rotation/global_etf_rotation_feature_snapshot_latest.csv", - "manifest_path": "gs://qsl-runtime-logs-shared/strategy-artifacts/us_equity/global_etf_rotation/global_etf_rotation_feature_snapshot_latest.csv.manifest.json" + "manifest_path": "gs://qsl-runtime-logs-shared/strategy-artifacts/us_equity/global_etf_rotation/global_etf_rotation_feature_snapshot_latest.csv.manifest.json", + "max_age_days": 5 } }, "features": { @@ -565,7 +566,8 @@ "feature_snapshot": { "required": true, "path": "gs://qsl-runtime-logs-shared/strategy-artifacts/us_equity/russell_top50_leader_rotation_staging/russell_top50_leader_rotation_feature_snapshot_latest.csv", - "manifest_path": "gs://qsl-runtime-logs-shared/strategy-artifacts/us_equity/russell_top50_leader_rotation_staging/russell_top50_leader_rotation_feature_snapshot_latest.csv.manifest.json" + "manifest_path": "gs://qsl-runtime-logs-shared/strategy-artifacts/us_equity/russell_top50_leader_rotation_staging/russell_top50_leader_rotation_feature_snapshot_latest.csv.manifest.json", + "max_age_days": 40 } }, "features": { @@ -726,7 +728,8 @@ "feature_snapshot": { "required": true, "path": "gs://qsl-runtime-logs-shared/strategy-artifacts/hk_equity/hk_low_vol_dividend_quality_snapshot_staging/hk_low_vol_dividend_quality_snapshot_factor_snapshot_latest.csv", - "manifest_path": "gs://qsl-runtime-logs-shared/strategy-artifacts/hk_equity/hk_low_vol_dividend_quality_snapshot_staging/hk_low_vol_dividend_quality_snapshot_factor_snapshot_latest.csv.manifest.json" + "manifest_path": "gs://qsl-runtime-logs-shared/strategy-artifacts/hk_equity/hk_low_vol_dividend_quality_snapshot_staging/hk_low_vol_dividend_quality_snapshot_factor_snapshot_latest.csv.manifest.json", + "max_age_days": 40 } }, "features": { diff --git a/python/scripts/build_config.py b/python/scripts/build_config.py index 1a5dfec..04a0472 100644 --- a/python/scripts/build_config.py +++ b/python/scripts/build_config.py @@ -40,7 +40,7 @@ } SCHEDULER_FIELDS = {"timezone", "main_time", "probe_time", "precheck_time"} MARKET_FIELDS = {"market", "market_calendar", "market_timezone"} -FEATURE_SNAPSHOT_FIELDS = {"required", "path", "manifest_path"} +FEATURE_SNAPSHOT_FIELDS = {"required", "path", "manifest_path", "max_age_days"} # These platforms have a runtime variable pair through which an immutable # feature snapshot and its manifest can be supplied to a generated target. # Keep this small, explicit set next to the config validation so coverage does @@ -356,6 +356,17 @@ def validate(config: dict) -> list[str]: errors.append( f"strategy {sid}: live feature snapshot requires path and manifest_path" ) + max_age_days = feature_snapshot.get("max_age_days") + if required is True and not isinstance(max_age_days, int): + errors.append( + f"strategy {sid}: required feature snapshot max_age_days must be an integer" + ) + elif isinstance(max_age_days, bool) or ( + isinstance(max_age_days, int) and max_age_days < 1 + ): + errors.append( + f"strategy {sid}: feature snapshot max_age_days must be at least 1" + ) return errors @@ -479,6 +490,69 @@ def build_strategy_platform_dry_run_coverage(config: dict | None = None) -> dict } +def build_runtime_artifact_evidence_registry(config: dict | None = None) -> dict[str, object]: + """Build the immutable, no-order verification plan for required snapshots. + + This registry deliberately describes only controller-declared artifacts. It + never publishes data, rewrites a URI, or grants a strategy a higher + lifecycle stage. The verifier consumes it using a read-only cloud + identity and reports failures back to the operator. + """ + config = config if config is not None else load_config() + coverage_by_profile = { + str(row["profile"]): row + for row in build_strategy_platform_dry_run_coverage(config)["profiles"] + if isinstance(row, dict) and isinstance(row.get("profile"), str) + } + entries: list[dict[str, object]] = [] + for profile, strategy in sorted(config.get("strategies", {}).items()): + if not isinstance(strategy, dict): + continue + runtime_artifacts = strategy.get("runtime_artifacts") + feature_snapshot = ( + runtime_artifacts.get("feature_snapshot") + if isinstance(runtime_artifacts, dict) + else None + ) + if not isinstance(feature_snapshot, dict) or feature_snapshot.get("required") is not True: + continue + snapshot_path = feature_snapshot.get("path") + manifest_path = feature_snapshot.get("manifest_path") + max_age_days = feature_snapshot.get("max_age_days") + if not ( + isinstance(snapshot_path, str) + and snapshot_path.startswith("gs://") + and isinstance(manifest_path, str) + and manifest_path.startswith("gs://") + and isinstance(max_age_days, int) + and not isinstance(max_age_days, bool) + and max_age_days >= 1 + ): + continue + coverage = coverage_by_profile.get(str(profile), {}) + entries.append( + { + "profile": str(profile), + "domain": str(strategy.get("domain") or ""), + "snapshot_path": snapshot_path, + "manifest_path": manifest_path, + "max_age_days": max_age_days, + "dry_run_platforms": list(coverage.get("buildable_dry_run_platforms", [])), + "boundary": "read_only_evidence_check_no_publish_no_execution", + } + ) + return { + "schema_version": "runtime_artifact_evidence_registry.v1", + "entries": entries, + "summary": {"required_artifact_count": len(entries)}, + "boundary": ( + "Read-only validation plan. A passing entry proves the declared object, " + "manifest digest, and freshness contract; it does not authorize paper, " + "shadow, or live execution." + ), + } + + def validate_notification_references(config: dict, errors: list[str]) -> None: """Keep every Telegram notification route in runtime configuration.""" notifications = config.get("notifications") @@ -1071,6 +1145,11 @@ def main() -> int: parser.add_argument("--live-candidate-queue", action="store_true", help="Print live-candidate queue JSON and exit") parser.add_argument("--platform-health-report", action="store_true", help="Print platform health report JSON and exit") parser.add_argument("--automation-registry", action="store_true", help="Print strategy automation registry JSON and exit") + parser.add_argument( + "--runtime-artifact-evidence-registry", + action="store_true", + help="Print the read-only verification registry for required runtime artifacts and exit", + ) args = parser.parse_args() config = load_config() @@ -1081,6 +1160,16 @@ def main() -> int: if args.automation_registry: print(json.dumps(build_strategy_automation_registry(config), ensure_ascii=False, indent=2, sort_keys=True)) return 0 + if args.runtime_artifact_evidence_registry: + print( + json.dumps( + build_runtime_artifact_evidence_registry(config), + ensure_ascii=False, + indent=2, + sort_keys=True, + ) + ) + return 0 errors = validate(config) if errors: diff --git a/python/scripts/verify_runtime_artifact_evidence.py b/python/scripts/verify_runtime_artifact_evidence.py new file mode 100644 index 0000000..dda5666 --- /dev/null +++ b/python/scripts/verify_runtime_artifact_evidence.py @@ -0,0 +1,190 @@ +#!/usr/bin/env python3 +"""Verify controller-declared runtime snapshots with a read-only GCS identity. + +The control plane owns the list of required artifacts. This tool only reads +the paired snapshot and manifest, verifies the declared digest/profile/date, +and emits a receipt. It must never publish artifacts, change runtime +settings, or invoke an execution path. +""" + +from __future__ import annotations + +import argparse +import datetime as dt +import hashlib +import json +import subprocess +from collections.abc import Callable +from pathlib import Path +from typing import Any + + +BytesFetcher = Callable[[str], bytes] + + +def _read_gcs(uri: str) -> bytes: + """Read a single GCS object through the ambient read-only identity.""" + completed = subprocess.run( + ["gcloud", "storage", "cat", uri], + check=False, + capture_output=True, + ) + if completed.returncode != 0: + detail = completed.stderr.decode("utf-8", errors="replace").strip() + raise RuntimeError(detail or f"gcloud storage cat exited {completed.returncode}") + return completed.stdout + + +def _parse_snapshot_date(value: object) -> dt.date: + if not isinstance(value, str): + raise ValueError("manifest.snapshot_as_of must be an ISO date string") + try: + return dt.date.fromisoformat(value) + except ValueError as exc: + raise ValueError("manifest.snapshot_as_of must be an ISO date string") from exc + + +def verify_entry( + entry: dict[str, Any], + *, + fetch_bytes: BytesFetcher, + today: dt.date, +) -> dict[str, Any]: + """Verify one registry entry without changing remote state.""" + profile = str(entry.get("profile") or "") + result: dict[str, Any] = { + "profile": profile, + "snapshot_path": entry.get("snapshot_path"), + "manifest_path": entry.get("manifest_path"), + "max_age_days": entry.get("max_age_days"), + "status": "failed", + "errors": [], + } + errors: list[str] = result["errors"] + snapshot_path = entry.get("snapshot_path") + manifest_path = entry.get("manifest_path") + max_age_days = entry.get("max_age_days") + if not profile: + errors.append("registry profile is missing") + if not isinstance(snapshot_path, str) or not snapshot_path.startswith("gs://"): + errors.append("registry snapshot_path must be a gs:// URI") + if not isinstance(manifest_path, str) or not manifest_path.startswith("gs://"): + errors.append("registry manifest_path must be a gs:// URI") + if isinstance(max_age_days, bool) or not isinstance(max_age_days, int) or max_age_days < 1: + errors.append("registry max_age_days must be an integer of at least 1") + if errors: + return result + + try: + snapshot_bytes = fetch_bytes(snapshot_path) + except Exception as exc: # noqa: BLE001 - receipt must retain remote read failure. + errors.append(f"snapshot_read_failed: {exc}") + return result + try: + manifest_bytes = fetch_bytes(manifest_path) + manifest = json.loads(manifest_bytes.decode("utf-8")) + except Exception as exc: # noqa: BLE001 - receipt must retain remote parse failure. + errors.append(f"manifest_read_failed: {exc}") + return result + if not isinstance(manifest, dict): + errors.append("manifest must be a JSON object") + return result + + manifest_profile = manifest.get("strategy_profile") + if manifest_profile != profile: + errors.append( + f"manifest.strategy_profile mismatch: expected {profile!r}, got {manifest_profile!r}" + ) + expected_sha = manifest.get("snapshot_sha256") + actual_sha = hashlib.sha256(snapshot_bytes).hexdigest() + if not isinstance(expected_sha, str) or expected_sha != actual_sha: + errors.append("manifest.snapshot_sha256 does not match snapshot bytes") + try: + snapshot_as_of = _parse_snapshot_date(manifest.get("snapshot_as_of")) + age_days = (today - snapshot_as_of).days + result["snapshot_as_of"] = snapshot_as_of.isoformat() + result["age_days"] = age_days + if age_days < 0: + errors.append("manifest.snapshot_as_of is in the future") + elif age_days > max_age_days: + errors.append( + f"snapshot is stale: age {age_days} days exceeds max_age_days {max_age_days}" + ) + except ValueError as exc: + errors.append(str(exc)) + if not errors: + result["status"] = "verified" + return result + + +def verify_registry( + registry: dict[str, Any], + *, + fetch_bytes: BytesFetcher, + today: dt.date, +) -> dict[str, Any]: + entries = registry.get("entries") + if not isinstance(entries, list): + raise ValueError("registry.entries must be a list") + results = [ + verify_entry(entry, fetch_bytes=fetch_bytes, today=today) + if isinstance(entry, dict) + else { + "profile": "", + "status": "failed", + "errors": ["registry entry must be an object"], + } + for entry in entries + ] + failures = [item for item in results if item["status"] != "verified"] + return { + "schema_version": "runtime_artifact_evidence_receipt.v1", + "checked_at": today.isoformat(), + "status": "verified" if not failures else "failed", + "summary": { + "required_artifact_count": len(results), + "verified_artifact_count": len(results) - len(failures), + "failed_artifact_count": len(failures), + }, + "entries": results, + "boundary": ( + "Read-only evidence receipt. It does not publish data, change a runtime " + "target, or authorize paper, shadow, or live execution." + ), + } + + +def _load_registry(path: Path) -> dict[str, Any]: + payload = json.loads(path.read_text(encoding="utf-8")) + if not isinstance(payload, dict): + raise ValueError("registry must be a JSON object") + return payload + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser( + description="Read-only verification of required runtime snapshot artifacts." + ) + parser.add_argument("--registry", required=True, type=Path) + parser.add_argument("--output", type=Path) + parser.add_argument( + "--as-of", + type=dt.date.fromisoformat, + help="Override the verification date (ISO date; intended for deterministic tests).", + ) + args = parser.parse_args(argv) + receipt = verify_registry( + _load_registry(args.registry), + fetch_bytes=_read_gcs, + today=args.as_of or dt.datetime.now(tz=dt.UTC).date(), + ) + payload = json.dumps(receipt, ensure_ascii=False, indent=2, sort_keys=True) + if args.output: + args.output.parent.mkdir(parents=True, exist_ok=True) + args.output.write_text(payload + "\n", encoding="utf-8") + print(payload) + return 0 if receipt["status"] == "verified" else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/python/tests/test_runtime_artifact_evidence.py b/python/tests/test_runtime_artifact_evidence.py new file mode 100644 index 0000000..02753c8 --- /dev/null +++ b/python/tests/test_runtime_artifact_evidence.py @@ -0,0 +1,103 @@ +from __future__ import annotations + +import datetime as dt +import hashlib +import importlib.util +import json +import sys +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[2] +MODULE_PATH = ROOT / "python" / "scripts" / "verify_runtime_artifact_evidence.py" +SPEC = importlib.util.spec_from_file_location("verify_runtime_artifact_evidence", MODULE_PATH) +artifact_evidence = importlib.util.module_from_spec(SPEC) +assert SPEC.loader is not None +sys.modules[SPEC.name] = artifact_evidence +SPEC.loader.exec_module(artifact_evidence) + + +def _registry(*, max_age_days: int = 5) -> tuple[dict, dict[str, bytes]]: + snapshot = b"symbol,score\nQQQ,1\n" + manifest = { + "strategy_profile": "example_profile", + "snapshot_sha256": hashlib.sha256(snapshot).hexdigest(), + "snapshot_as_of": "2026-08-27", + } + objects = { + "gs://bucket/example.csv": snapshot, + "gs://bucket/example.csv.manifest.json": json.dumps(manifest).encode("utf-8"), + } + return ( + { + "schema_version": "runtime_artifact_evidence_registry.v1", + "entries": [ + { + "profile": "example_profile", + "snapshot_path": "gs://bucket/example.csv", + "manifest_path": "gs://bucket/example.csv.manifest.json", + "max_age_days": max_age_days, + } + ], + }, + objects, + ) + + +def test_verify_registry_accepts_fresh_matching_manifest(): + registry, objects = _registry() + + receipt = artifact_evidence.verify_registry( + registry, + fetch_bytes=objects.__getitem__, + today=dt.date(2026, 8, 28), + ) + + assert receipt["status"] == "verified" + assert receipt["summary"] == { + "required_artifact_count": 1, + "verified_artifact_count": 1, + "failed_artifact_count": 0, + } + assert receipt["entries"][0]["age_days"] == 1 + + +def test_verify_registry_fails_closed_on_digest_mismatch(): + registry, objects = _registry() + manifest = json.loads(objects["gs://bucket/example.csv.manifest.json"]) + manifest["snapshot_sha256"] = "0" * 64 + objects["gs://bucket/example.csv.manifest.json"] = json.dumps(manifest).encode("utf-8") + + receipt = artifact_evidence.verify_registry( + registry, + fetch_bytes=objects.__getitem__, + today=dt.date(2026, 8, 28), + ) + + assert receipt["status"] == "failed" + assert receipt["entries"][0]["errors"] == [ + "manifest.snapshot_sha256 does not match snapshot bytes" + ] + + +def test_verify_registry_fails_closed_on_stale_or_unreadable_snapshot(): + registry, objects = _registry(max_age_days=1) + + stale = artifact_evidence.verify_registry( + registry, + fetch_bytes=objects.__getitem__, + today=dt.date(2026, 8, 29), + ) + assert stale["status"] == "failed" + assert "snapshot is stale" in stale["entries"][0]["errors"][0] + + def unavailable(_: str) -> bytes: + raise RuntimeError("forbidden") + + unavailable_receipt = artifact_evidence.verify_registry( + registry, + fetch_bytes=unavailable, + today=dt.date(2026, 8, 28), + ) + assert unavailable_receipt["status"] == "failed" + assert unavailable_receipt["entries"][0]["errors"] == ["snapshot_read_failed: forbidden"] diff --git a/python/tests/test_runtime_settings.py b/python/tests/test_runtime_settings.py index c84da18..feef1df 100644 --- a/python/tests/test_runtime_settings.py +++ b/python/tests/test_runtime_settings.py @@ -107,6 +107,20 @@ def test_platform_health_monitor_workflow_creates_codex_ready_issue(self): self.assertIn("codex-repair-ready", workflow) self.assertIn("Do not enable live switching", workflow) + def test_runtime_artifact_evidence_gate_is_read_only_and_uses_registry(self): + workflow = (ROOT / ".github/workflows/runtime-artifact-evidence-gate.yml").read_text( + encoding="utf-8" + ) + + self.assertIn("id-token: write", workflow) + self.assertIn("--runtime-artifact-evidence-registry", workflow) + self.assertIn("verify_runtime_artifact_evidence.py", workflow) + self.assertIn("qsl-artifact-evidence@", workflow) + self.assertIn("no publishing, runtime change, or order submission", workflow) + self.assertIn("Keep affected routes parked", workflow) + self.assertNotIn("gcloud storage cp", workflow) + self.assertNotIn("Manual Strategy Switch", workflow) + def test_manual_switch_platform_choices_cover_supported_platforms(self): workflow = (ROOT / ".github/workflows/manual-strategy-switch.yml").read_text(encoding="utf-8") platform_choices: list[str] = [] @@ -2439,6 +2453,34 @@ def test_build_config_rejects_non_boolean_snapshot_requirement(self): build_config.validate(config), ) + def test_runtime_artifact_evidence_registry_covers_every_required_snapshot(self): + config = build_config.load_config() + + registry = build_config.build_runtime_artifact_evidence_registry(config) + + self.assertEqual(registry["schema_version"], "runtime_artifact_evidence_registry.v1") + self.assertEqual(registry["summary"]["required_artifact_count"], 3) + self.assertEqual( + {entry["profile"] for entry in registry["entries"]}, + { + "global_etf_rotation", + "russell_top50_leader_rotation", + "hk_low_vol_dividend_quality_snapshot", + }, + ) + self.assertTrue(all(entry["max_age_days"] >= 1 for entry in registry["entries"])) + + def test_build_config_requires_required_snapshot_freshness_budget(self): + config = build_config.load_config() + config["strategies"]["global_etf_rotation"]["runtime_artifacts"][ + "feature_snapshot" + ].pop("max_age_days") + + self.assertIn( + "strategy global_etf_rotation: required feature snapshot max_age_days must be an integer", + build_config.validate(config), + ) + def test_build_config_reports_malformed_scheduler_timezone(self): config = build_config.load_config() config["scheduling"]["profiles"]["us_daily"]["timezone"] = "../UTC" diff --git a/web/strategy-switch-console/config.js b/web/strategy-switch-console/config.js index 3956f35..4c6b07f 100644 --- a/web/strategy-switch-console/config.js +++ b/web/strategy-switch-console/config.js @@ -109,7 +109,7 @@ export const RUNTIME_CATALOG_PROJECTION = { "data_status": "catalog_only", "source": { "path": "platform-config.json", - "content_sha256": "sha256:61ca2ad4dc29d22fca901d3a598d83ae5c2916493333b24a1f8ac5dd9ebfe4fe", + "content_sha256": "sha256:dbfbd4a029c8ed3761c2b5db8d97ff1668166bd55d94bb03ed5806c357aa0a39", "catalog_as_of": "2026-08-19" }, "policy": { diff --git a/web/strategy-switch-console/runtime-catalog-projection.json b/web/strategy-switch-console/runtime-catalog-projection.json index 40a61aa..eb556f8 100644 --- a/web/strategy-switch-console/runtime-catalog-projection.json +++ b/web/strategy-switch-console/runtime-catalog-projection.json @@ -22,7 +22,7 @@ "schema_version": "qsl.runtime_catalog_projection.v1", "source": { "catalog_as_of": "2026-08-19", - "content_sha256": "sha256:61ca2ad4dc29d22fca901d3a598d83ae5c2916493333b24a1f8ac5dd9ebfe4fe", + "content_sha256": "sha256:dbfbd4a029c8ed3761c2b5db8d97ff1668166bd55d94bb03ed5806c357aa0a39", "path": "platform-config.json" }, "strategies": [