From 5e3a2dba9cf3bef3708709150c8e6d95fbb9f1d3 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Wed, 7 Oct 2026 03:58:10 +0800 Subject: [PATCH] ci: wire disabled single-file theme draft publisher --- .github/workflows/theme_momentum_snapshot.yml | 343 +++++- README.md | 60 + README.zh-CN.md | 49 + docs/architecture.md | 13 +- scripts/publish_theme_snapshot_pr.py | 617 ++++++++++ tests/test_publish_theme_snapshot_pr.py | 1076 +++++++++++++++++ .../test_theme_momentum_snapshot_workflow.py | 465 ++++++- 7 files changed, 2568 insertions(+), 55 deletions(-) create mode 100644 scripts/publish_theme_snapshot_pr.py create mode 100644 tests/test_publish_theme_snapshot_pr.py diff --git a/.github/workflows/theme_momentum_snapshot.yml b/.github/workflows/theme_momentum_snapshot.yml index cf5d0f6..dc7e113 100644 --- a/.github/workflows/theme_momentum_snapshot.yml +++ b/.github/workflows/theme_momentum_snapshot.yml @@ -8,6 +8,11 @@ on: required: false default: false type: boolean + publish_draft_pr: + description: "Request a same-repository draft PR; source content approval is also required. Manual main only." + required: false + default: false + type: boolean as_of: description: "Optional YYYY-MM-DD cutoff. Empty uses latest available/downloaded price date." required: false @@ -26,7 +31,7 @@ on: - "false" - "true" commit_outputs: - description: "Commit generated theme_momentum_snapshot.json back to data/output." + description: "Deprecated, ignored: direct commits are disabled; this never authorizes a draft PR." required: false default: "false" type: choice @@ -34,12 +39,11 @@ on: - "false" - "true" schedule: - # Refresh before the advisory publication workflow. The script records data - # source and hash metadata; downstream use remains research-only. + # Schedule is artifact-only. It never opts into draft PR publication. - cron: "45 11 * * 6" permissions: - contents: write + contents: read concurrency: group: ${{ github.workflow }}-${{ github.ref_name }} @@ -49,13 +53,19 @@ jobs: build-theme-momentum: runs-on: ubuntu-latest timeout-minutes: 30 + outputs: + artifact_id: ${{ steps.publisher_package_upload.outputs.artifact-id }} + artifact_digest: ${{ steps.publisher_package_upload.outputs.artifact-digest }} + candidate_sha256: ${{ steps.publisher_package.outputs.candidate_sha256 }} + receipt_sha256: ${{ steps.publisher_package.outputs.receipt_sha256 }} steps: - # Manual diagnostic only. Normal generation and publication behavior is unchanged. + # Manual diagnostic only. Conflicting flags fail before either App mint. - name: Check trusted publisher identity context id: publisher_identity_preflight if: ${{ github.event_name == 'workflow_dispatch' && inputs.identity_only }} env: RESEARCH_PUBLISHER_APP_ID: ${{ vars.RESEARCH_PUBLISHER_APP_ID }} + PUBLISH_DRAFT_PR: ${{ inputs.publish_draft_pr }} run: | python - <<'PY' import os @@ -67,6 +77,7 @@ jobs: repository_id = os.environ.get("GITHUB_REPOSITORY_ID", "") context_matches = ( os.environ.get("GITHUB_EVENT_NAME") == "workflow_dispatch" + and os.environ.get("PUBLISH_DRAFT_PR") != "true" and os.environ.get("GITHUB_REF") == "refs/heads/main" and os.environ.get("GITHUB_REPOSITORY") == expected_repository and os.environ.get("GITHUB_WORKFLOW_REF") == expected_workflow_ref @@ -174,9 +185,18 @@ jobs: with Path(os.environ["GITHUB_STEP_SUMMARY"]).open("a", encoding="utf-8") as output: output.write("Publisher identity-only result (no data published):\n" + serialized + "\n") PY - - uses: actions/checkout@v6 + - name: Check out generation source + id: generation_checkout + uses: actions/checkout@d23441a48e516b6c34aea4fa41551a30e30af803 # v6 if: ${{ !inputs.identity_only }} - - uses: actions/setup-python@v6 + with: + ref: ${{ github.sha }} + persist-credentials: false + - name: Check publication content admission before collection + id: publisher_admission + if: ${{ !inputs.identity_only && inputs.publish_draft_pr }} + run: python scripts/publish_theme_snapshot_pr.py admission + - uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # v6 if: ${{ !inputs.identity_only }} with: python-version: "3.11" @@ -184,6 +204,7 @@ jobs: if: ${{ !inputs.identity_only }} run: python -m pip install -e . - name: Build theme momentum snapshot + id: generation if: ${{ !inputs.identity_only }} env: INPUT_AS_OF: ${{ github.event.inputs.as_of || '' }} @@ -191,6 +212,7 @@ jobs: STRICT_DOWNLOADS: ${{ github.event.inputs.strict_downloads || 'false' }} run: | set -euo pipefail + echo "started_at=$(date -u +'%Y-%m-%dT%H:%M:%S.%6NZ')" >> "$GITHUB_OUTPUT" mkdir -p data/output ARGS=(--output data/output/theme_momentum_snapshot.json) if [ -n "${INPUT_AS_OF}" ]; then @@ -203,6 +225,7 @@ jobs: ARGS+=(--strict-downloads) fi python scripts/build_theme_momentum_snapshot.py "${ARGS[@]}" | tee /tmp/theme_momentum_summary.json + echo "completed_at=$(date -u +'%Y-%m-%dT%H:%M:%S.%6NZ')" >> "$GITHUB_OUTPUT" { echo "## Theme momentum snapshot" echo "" @@ -210,42 +233,280 @@ jobs: cat /tmp/theme_momentum_summary.json echo '```' } >> "$GITHUB_STEP_SUMMARY" - - name: Commit snapshot - if: ${{ !inputs.identity_only }} + - name: Prepare same-run publication package + id: publisher_package + if: ${{ !inputs.identity_only && inputs.publish_draft_pr }} env: - COMMIT_OUTPUTS: ${{ github.event_name == 'schedule' && 'true' || github.event.inputs.commit_outputs || 'false' }} + PACKAGE_DIR: ${{ runner.temp }}/theme-publication-package + GENERATION_STARTED_AT: ${{ steps.generation.outputs.started_at }} + GENERATION_COMPLETED_AT: ${{ steps.generation.outputs.completed_at }} run: | - set -euo pipefail - if [ "${COMMIT_OUTPUTS}" != "true" ]; then - echo "Snapshot commit disabled." - exit 0 - fi - git config user.name "github-actions[bot]" - git config user.email "41898282+github-actions[bot]@users.noreply.github.com" - git add data/output/theme_momentum_snapshot.json - if git diff --cached --quiet; then - echo "No theme momentum changes to commit." - else - git commit -m "Update theme momentum snapshot [skip ci]" - push_log="$(mktemp)" - if git push 2>"${push_log}"; then - rm -f "${push_log}" - else - if grep -Fq 'GH013: Repository rule violations found for refs/heads/main' "${push_log}" && grep -Eqi 'required status check' "${push_log}"; then - echo "Push blocked by repository rules; keeping artifact-only output." - cat "${push_log}" - rm -f "${push_log}" - else - cat "${push_log}" >&2 - rm -f "${push_log}" - exit 1 - fi - fi - fi - - name: Upload theme momentum artifact - if: ${{ !inputs.identity_only }} - uses: actions/upload-artifact@v7 + python scripts/publish_theme_snapshot_pr.py prepare \ + --candidate data/output/theme_momentum_snapshot.json \ + --package-dir "$PACKAGE_DIR" \ + --started-at "$GENERATION_STARTED_AT" \ + --completed-at "$GENERATION_COMPLETED_AT" + - name: Upload same-run publication package + id: publisher_package_upload + if: ${{ !inputs.identity_only && inputs.publish_draft_pr }} + uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7 + with: + name: theme-momentum-publication-${{ github.run_id }}-${{ github.run_attempt }} + path: | + ${{ runner.temp }}/theme-publication-package/theme_momentum_snapshot.json + ${{ runner.temp }}/theme-publication-package/publication-package.json + if-no-files-found: error + - name: Upload theme momentum artifact or failed-package fallback + id: snapshot_upload + if: ${{ always() && !inputs.identity_only && steps.generation.outcome == 'success' && (!inputs.publish_draft_pr || steps.publisher_package_upload.outcome != 'success') }} + uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7 with: - name: theme-momentum-snapshot + name: theme-momentum-snapshot-${{ github.run_id }}-${{ github.run_attempt }} path: data/output/theme_momentum_snapshot.json if-no-files-found: error + - name: Report generation and publication request + id: generation_summary + if: ${{ always() && !inputs.identity_only }} + env: + PUBLICATION_REQUESTED: ${{ inputs.publish_draft_pr }} + LEGACY_COMMIT_OUTPUTS: ${{ inputs.commit_outputs }} + GENERATION_OUTCOME: ${{ steps.generation.outcome }} + ADMISSION_OUTCOME: ${{ steps.publisher_admission.outcome }} + PACKAGE_OUTCOME: ${{ steps.publisher_package.outcome }} + PACKAGE_UPLOAD_OUTCOME: ${{ steps.publisher_package_upload.outcome }} + SNAPSHOT_UPLOAD_OUTCOME: ${{ steps.snapshot_upload.outcome }} + run: | + python - <<'PY' + import json + import os + from pathlib import Path + + requested = os.environ.get("PUBLICATION_REQUESTED") == "true" + generated = os.environ.get("GENERATION_OUTCOME") == "success" + if requested and os.environ.get("ADMISSION_OUTCOME") != "success": + status = "admission_rejected" + elif not generated: + status = "generation_failed" + elif requested and os.environ.get("PACKAGE_OUTCOME") != "success": + status = "package_rejected" + elif requested and os.environ.get("PACKAGE_UPLOAD_OUTCOME") == "success": + status = "package_ready" + elif not requested and os.environ.get("SNAPSHOT_UPLOAD_OUTCOME") == "success": + status = "artifact_only" + else: + status = "artifact_upload_failed" + result = { + "generated": generated, + "publication_requested": requested, + "published": False, + "status": status, + "legacy_commit_outputs_ignored": os.environ.get("LEGACY_COMMIT_OUTPUTS") == "true", + } + serialized = json.dumps(result, sort_keys=True) + print(serialized) + with Path(os.environ["GITHUB_STEP_SUMMARY"]).open("a", encoding="utf-8") as output: + output.write("Theme generation result (no data published):\n" + serialized + "\n") + PY + + publish-theme-draft-pr: + needs: build-theme-momentum + if: >- + ${{ github.event_name == 'workflow_dispatch' && + github.repository == 'QuantStrategyLab/ResearchSignalContextPipelines' && + github.ref == 'refs/heads/main' && + github.workflow_ref == 'QuantStrategyLab/ResearchSignalContextPipelines/.github/workflows/theme_momentum_snapshot.yml@refs/heads/main' && + inputs.publish_draft_pr && !inputs.identity_only }} + runs-on: ubuntu-latest + timeout-minutes: 15 + permissions: + contents: read + steps: + - name: Check out reviewed publisher source + id: publisher_checkout + uses: actions/checkout@d23441a48e516b6c34aea4fa41551a30e30af803 # v6 + with: + ref: ${{ github.sha }} + persist-credentials: false + - name: Download the exact current-run package + id: publisher_download + uses: actions/download-artifact@3e5f45b2cfb9172054b4087a40e8e0b5a5461e7c # v8 + with: + artifact-ids: ${{ needs.build-theme-momentum.outputs.artifact_id }} + path: ${{ runner.temp }}/theme-publication-package + merge-multiple: true + digest-mismatch: error + - name: Validate package and unchanged main before mint + id: publisher_validate + env: + GITHUB_TOKEN: ${{ github.token }} + PACKAGE_DIR: ${{ runner.temp }}/theme-publication-package + CANDIDATE_SHA256: ${{ needs.build-theme-momentum.outputs.candidate_sha256 }} + RECEIPT_SHA256: ${{ needs.build-theme-momentum.outputs.receipt_sha256 }} + ARTIFACT_ID: ${{ needs.build-theme-momentum.outputs.artifact_id }} + ARTIFACT_DIGEST: ${{ needs.build-theme-momentum.outputs.artifact_digest }} + run: | + python scripts/publish_theme_snapshot_pr.py validate \ + --package-dir "$PACKAGE_DIR" \ + --candidate-sha256 "$CANDIDATE_SHA256" \ + --receipt-sha256 "$RECEIPT_SHA256" \ + --artifact-id "$ARTIFACT_ID" \ + --artifact-digest "$ARTIFACT_DIGEST" + - name: Create single-repository draft publisher token + id: draft_publisher_token + uses: actions/create-github-app-token@bcd2ba49218906704ab6c1aa796996da409d3eb1 # v3.2.0 + with: + app-id: "5214420" + private-key: ${{ secrets.RESEARCH_PUBLISHER_APP_PRIVATE_KEY }} + owner: QuantStrategyLab + repositories: ResearchSignalContextPipelines + permission-contents: write + permission-pull-requests: write + - name: Create or reuse the exact draft PR + id: draft_publish + env: + RESEARCH_PUBLISHER_TOKEN: ${{ steps.draft_publisher_token.outputs.token }} + RESEARCH_PUBLISHER_INSTALLATION_ID: ${{ steps.draft_publisher_token.outputs.installation-id }} + RESEARCH_PUBLISHER_APP_SLUG: ${{ steps.draft_publisher_token.outputs.app-slug }} + PACKAGE_DIR: ${{ runner.temp }}/theme-publication-package + CANDIDATE_SHA256: ${{ needs.build-theme-momentum.outputs.candidate_sha256 }} + RECEIPT_SHA256: ${{ needs.build-theme-momentum.outputs.receipt_sha256 }} + ARTIFACT_ID: ${{ needs.build-theme-momentum.outputs.artifact_id }} + ARTIFACT_DIGEST: ${{ needs.build-theme-momentum.outputs.artifact_digest }} + RESULT_PATH: ${{ runner.temp }}/theme-publication-result/publication-result.json + run: | + python scripts/publish_theme_snapshot_pr.py publish \ + --package-dir "$PACKAGE_DIR" \ + --candidate-sha256 "$CANDIDATE_SHA256" \ + --receipt-sha256 "$RECEIPT_SHA256" \ + --artifact-id "$ARTIFACT_ID" \ + --artifact-digest "$ARTIFACT_DIGEST" \ + --result "$RESULT_PATH" + - name: Report sanitized draft publisher result + id: draft_publisher_summary + if: ${{ always() }} + env: + RESULT_PATH: ${{ runner.temp }}/theme-publication-result/publication-result.json + CHECKOUT_OUTCOME: ${{ steps.publisher_checkout.outcome }} + DOWNLOAD_OUTCOME: ${{ steps.publisher_download.outcome }} + VALIDATE_OUTCOME: ${{ steps.publisher_validate.outcome }} + MINT_OUTCOME: ${{ steps.draft_publisher_token.outcome }} + PUBLISH_OUTCOME: ${{ steps.draft_publish.outcome }} + ARTIFACT_ID: ${{ needs.build-theme-momentum.outputs.artifact_id }} + ARTIFACT_DIGEST: ${{ needs.build-theme-momentum.outputs.artifact_digest }} + CANDIDATE_SHA256: ${{ needs.build-theme-momentum.outputs.candidate_sha256 }} + RECEIPT_SHA256: ${{ needs.build-theme-momentum.outputs.receipt_sha256 }} + run: | + python - <<'PY' + import json + import os + import re + from pathlib import Path + + path = Path(os.environ["RESULT_PATH"]) + result = {"published": False, "review_required": True, "status": "publisher_failed"} + for outcome, failure in ( + ("CHECKOUT_OUTCOME", "checkout_failed"), + ("DOWNLOAD_OUTCOME", "download_failed"), + ("VALIDATE_OUTCOME", "validation_rejected"), + ("MINT_OUTCOME", "mint_failed"), + ): + if os.environ.get(outcome) != "success": + result["status"] = failure + break + else: + try: + raw = path.read_bytes() + value = json.loads(raw) if len(raw) <= 16384 else None + if type(value) is dict: + for key in ("branch_created", "pr_created"): + if key in value and (value[key] is None or type(value[key]) is bool): + result[key] = value[key] + if type(value.get("main_contains_candidate")) is bool: + result["main_contains_candidate"] = value["main_contains_candidate"] + if value.get("status") in {"failed", "already_on_base", "draft_pr_created", "draft_pr_reused"}: + result["status"] = value["status"] + # Only known public reason codes, never arbitrary exception text. + reasons = set(""" + API_ALREADY_EXISTS API_METHOD_FORBIDDEN API_PAGINATION_UNEXPECTED API_RESPONSE_OVERSIZED + API_WRITE_FORBIDDEN REF_WRITE_FORBIDDEN TREE_WRITE_FORBIDDEN PR_WRITE_FORBIDDEN + RECEIPT_NONCANONICAL PUBLICATION_FAILED + API_TARGET_FORBIDDEN API_TRANSPORT_FAILED APP_IDENTITY_MISMATCH + APP_SCOPE_MISMATCH ARGUMENTS_REQUIRED ARTIFACT_BINDING_INVALID + BASE_INVALID BASE_MOVED BASE_PATH_INVALID BRANCH_CREATION_UNCONFIRMED + BRANCH_MODIFIED CANDIDATE_EXPIRED CANDIDATE_PATH_INVALID + CHECKOUT_SHA_MISMATCH CONFIG_HASH_MISMATCH CONTENT_APPROVAL_REQUIRED + COUNT_INVALID COUNT_MISMATCH COVERAGE_MISMATCH DATE_INVALID + FILE_NOT_REGULAR FILE_OVERSIZED GENERATION_TIME_INVALID JSON_DUPLICATE_KEY + JSON_INVALID JSON_NONFINITE NUMBER_INVALID NUMBER_RANGE + PACKAGE_CANDIDATE_MISMATCH PACKAGE_CONTEXT_MISMATCH PACKAGE_HASH_INVALID + PACKAGE_HASH_MISMATCH PACKAGE_NOT_EMPTY PACKAGE_PATHS_INVALID + PACKAGE_PATH_INVALID PACKAGE_SCHEMA_INVALID PRICE_SOURCE_NOT_ADMITTED + PR_CREATION_UNCONFIRMED PR_HUMAN_STATE_CHANGED PR_IDENTITY_AMBIGUOUS + PR_IDENTITY_MISMATCH REMOTE_BLOB_MISMATCH REMOTE_COMMIT_INVALID + REMOTE_REF_INVALID REMOTE_TREE_INVALID RESEARCH_CONTRACT_MISMATCH + RESULT_REQUIRED RUN_INVALID SCHEMA_KEYS SCOPE_INVALID SOURCE_SHA_INVALID + SUMMARY_MISMATCH SYMBOL_SCOPE_INVALID THEME_SCHEMA_INVALID THEME_SCOPE_INVALID + TIME_INVALID TOKEN_UNAVAILABLE UNTRUSTED_CONTEXT VALIDATION_OR_API_FAILED + """.split()) + reasons.update(f"API_HTTP_{code}" for code in range(100, 600)) + reason = value.get("reason_code") + if type(reason) is str and reason in reasons: + result["reason_code"] = reason + elif reason is not None: + result["reason_code"] = "VALIDATION_OR_API_FAILED" + for key, pattern in ( + ("branch", r"automation/theme-[1-9][0-9]*-[1-9][0-9]*-[0-9a-f]{16}"), + ("package_key", r"[0-9a-f]{64}"), + ("pr_url", r"https://github\.com/QuantStrategyLab/ResearchSignalContextPipelines/pull/[1-9][0-9]*"), + ("head_sha", r"[0-9a-f]{40}"), + ("candidate_sha256", r"[0-9a-f]{64}"), + ): + field = value.get(key) + if type(field) is str and re.fullmatch(pattern, field): + result[key] = field + except Exception: + pass + # Bind the sanitized result to the exact same-run transport outputs. + for key, pattern in ( + ("artifact_id", r"[1-9][0-9]*"), + ("artifact_digest", r"[0-9a-f]{64}"), + ("candidate_sha256", r"[0-9a-f]{64}"), + ("receipt_sha256", r"[0-9a-f]{64}"), + ): + value = os.environ.get(key.upper(), "") + if re.fullmatch(pattern, value): + result[key] = value + # Producer provenance comes only from this job's trusted runtime, never the artifact. + expected_repository = "QuantStrategyLab/ResearchSignalContextPipelines" + expected_workflow = expected_repository + "/.github/workflows/theme_momentum_snapshot.yml@refs/heads/main" + if ( + os.environ.get("GITHUB_EVENT_NAME") == "workflow_dispatch" + and os.environ.get("GITHUB_REF") == "refs/heads/main" + and os.environ.get("GITHUB_REPOSITORY") == expected_repository + and os.environ.get("GITHUB_REPOSITORY_ID") == "1252248184" + and os.environ.get("GITHUB_WORKFLOW_REF") == expected_workflow + ): + result.update(repository=expected_repository, repository_id="1252248184", workflow_ref=expected_workflow) + source_sha = os.environ.get("GITHUB_SHA", "") + if re.fullmatch(r"[0-9a-f]{40}", source_sha): + result.update(source_sha=source_sha, expected_base_sha=source_sha) + for key in ("run_id", "run_attempt"): + value = os.environ.get("GITHUB_" + key.upper(), "") + if re.fullmatch(r"[1-9][0-9]*", value): + result[key] = value + result["published"] = False + result["review_required"] = True + serialized = json.dumps(result, sort_keys=True) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(serialized + "\n", encoding="utf-8") + print(serialized) + with Path(os.environ["GITHUB_STEP_SUMMARY"]).open("a", encoding="utf-8") as output: + output.write("Draft PR result (human review and merge still required):\n" + serialized + "\n") + PY + - name: Upload sanitized publisher result + if: ${{ always() }} + uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7 + with: + name: theme-momentum-publication-result-${{ github.run_id }}-${{ github.run_attempt }} + path: ${{ runner.temp }}/theme-publication-result/publication-result.json + if-no-files-found: error diff --git a/README.md b/README.md index 4fc26c7..5accb1c 100644 --- a/README.md +++ b/README.md @@ -41,6 +41,66 @@ python -m pip install -e . python -m pytest -q ``` +## Theme snapshot workflow and draft PR boundary + +`.github/workflows/theme_momentum_snapshot.yml` normally builds +`data/output/theme_momentum_snapshot.json` and retains it as an Actions artifact. +The weekly schedule is artifact-only. The legacy `commit_outputs` input is +**deprecated and ignored**; it never pushes to `main` or authorizes a PR. + +The source includes a restricted, **production-disabled** draft publisher: + +- `identity_only=true` is the existing manual-main identity diagnostic. It does + not collect, generate, or publish data. Combining it with `publish_draft_pr=true` + fails before any token mint or generation. +- `publish_draft_pr` is a manual boolean, default `false`. Even `true` is not + content authorization: `PUBLICATION_CONTENT_APPROVED = False` in + `scripts/publish_theme_snapshot_pr.py` blocks publication requests immediately + after checkout, before installation, collection, generation, or production mint. +- Enabling a future run requires explicit approval of the source/provider and + derived-data publication scope, a reviewed source-policy change, and the + explicit manual-main flag. Publicly readable input, repository MIT licensing, + and an App installation do not establish redistribution rights. No scheduled + publication is authorized by this implementation. + +Normal generation preserves `as_of`, optional `prices_path`, and +`strict_downloads`. Publication policy separately constrains eligible inputs; +operator-supplied CSVs do not receive implied publication approval. The only +future data-PR path is `data/output/theme_momentum_snapshot.json`; raw prices, +AI signal/manifest pairs, history, configuration, and all other files are excluded. + +The build job records the actual generation interval, prepares an exact JSON plus +`publication-package.json` receipt, and passes artifact ID/digest and both raw +SHA-256 hashes through same-run job outputs. The isolated publisher checks out +the fixed run SHA with credentials persistence disabled, downloads only that +current-run artifact ID into runner temporary storage, and validates the package, +source/config/run provenance, content policy, and unchanged expected `main` before +minting a production token. It does not install or execute artifact content. + +`GITHUB_TOKEN` has only `contents: read`; there is no Actions read permission or +cross-run retrieval. The dedicated App token requests only Contents and Pull +requests write for this repository and is revoked by the official action's +normal post step. The diagnostic token remains isolated to the identity-only +branch; the production token is passed only to the restricted publish helper. +All workflow action dependencies are pinned to full commit SHAs. + +Future writes use a run-bound immutable branch, a single-parent/single-file +commit, and a draft PR. Exact existing drafts may be reused; moved `main`, changed +branches, and closed/merged/non-draft PRs fail closed. There is no force-push, +`main` write, automatic merge, protection bypass, old App fallback, or silent +success after remote failure. Sanitized receipts distinguish artifact generation, +package readiness, draft creation, and failure. A created draft still reports +`published=false`: human review and merge are still required. + +Local source/mock validation (no provider calls, App mint, dispatch, or data PR): + +```bash +python -m pip install -e '.[test]' +python -m pytest -q tests/test_theme_momentum_snapshot_workflow.py tests/test_publish_theme_snapshot_pr.py +python -m pytest -q +git diff --check +``` + ## Useful docs - [`docs/architecture.md`](docs/architecture.md) diff --git a/README.zh-CN.md b/README.zh-CN.md index 8be8e86..32d4340 100644 --- a/README.zh-CN.md +++ b/README.zh-CN.md @@ -41,6 +41,55 @@ python -m pip install -e . python -m pytest -q ``` +## 主题快照 workflow 与草稿 PR 边界 + +`.github/workflows/theme_momentum_snapshot.yml` 默认生成 +`data/output/theme_momentum_snapshot.json` 并保留为 Actions artifact。 +每周定时任务仅生成 artifact。旧 `commit_outputs` 输入已**弃用且忽略**， +不会向 `main` 推送，也不代表 PR 发布授权。 + +源码包含受限且**尚未启用生产发布**的草稿 publisher： + +- `identity_only=true` 保留现有的手动 main 身份诊断，不采集、生成或发布数据。 + 若同时设置 `publish_draft_pr=true`，在任何 token mint 或生成前拒绝。 +- `publish_draft_pr` 是默认 `false` 的手动布尔输入。设为 `true` 仍不构成内容授权： + `scripts/publish_theme_snapshot_pr.py` 中的 `PUBLICATION_CONTENT_APPROVED = False` + 会在 checkout 后立即阻止发布请求，早于安装、采集、生成和生产 token mint。 +- 未来真实运行需要明确批准来源/provider 及派生数据的公开范围，经源码 review + 修改内容政策，并显式开启手动 main 发布开关。公开可读取、仓库 MIT 许可证和 App + 安装均不证明再分发权利。本实现不授权定时发布。 + +普通生成保留 `as_of`、可选 `prices_path` 和 `strict_downloads` 的原有行为。 +发布政策另行约束允许的输入；用户提供 CSV 不代表获得公开许可。未来数据 PR +唯一允许的路径为 `data/output/theme_momentum_snapshot.json`，不包括原始价格、 +AI signal/manifest pair、history、配置或任何其他文件。 + +build job 记录真实生成起止时间，准备精确 JSON 与 `publication-package.json` +回执，通过同 run 的 job outputs 传递 artifact ID/digest 和两个原始文件 SHA-256。 +隔离 publisher checkout 当前 run 的固定 SHA，关闭凭据持久化，仅按该 run 的 +artifact ID 下载到 runner 临时目录；在生产 mint 前验证 package、源码/配置/run +绑定、内容政策及 `main` 是否仍是预期基线。不会安装或执行 artifact 中的代码。 + +`GITHUB_TOKEN` 仅有 `contents: read`，没有 Actions read 权限或跨 run 检索。 +专用 App token 仅申请本仓 Contents 与 Pull requests write，并由官方 action +默认 post step 撤销。诊断 token 只用于 identity-only 分支，生产 token 只传给 +受限 publish helper。所有 workflow action 依赖均固定为完整 commit SHA。 + +未来写入使用绑定 run 的不可变分支、单 parent/单文件 commit 和草稿 PR。 +完全匹配的现有草稿可复用；`main` 前进、分支被修改、PR 已关闭/合并/转为非草稿时 +拒绝继续。没有 force-push、直接写 `main`、自动 merge、绕过保护、旧 App 回退， +也不会把远端失败静默当作成功。脱敏回执分别记录 artifact 生成、package 就绪、 +草稿创建或失败；`published=false` 表明仍需人工审阅和合并。 + +本地源码/mock 验证，不调用 provider、不 mint、不 dispatch、不创建数据 PR： + +```bash +python -m pip install -e '.[test]' +python -m pytest -q tests/test_theme_momentum_snapshot_workflow.py tests/test_publish_theme_snapshot_pr.py +python -m pytest -q +git diff --check +``` + ## 延伸文档 - [`docs/architecture.md`](docs/architecture.md) diff --git a/docs/architecture.md b/docs/architecture.md index 8fde2c9..e24ba86 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -8,8 +8,17 @@ pipeline without changing that production boundary. This repository is deliberately narrower than `AIAuditBridge`. It owns research inputs, validation, saved artifacts, and replay harnesses. It does not -own model provider routing, API keys, GitHub App write orchestration, live -notifications, or execution behavior. +own model provider routing, API keys, cross-repository GitHub App write +orchestration, live notifications, or execution behavior. + +A limited same-repository exception is the source-only, production-disabled +publisher for `data/output/theme_momentum_snapshot.json`. It can only transport +that exact file through a run-bound draft PR using the dedicated single-repository +App; it never writes `main`, merges, bypasses protection, or handles AI signal +pairs or other files. Content/source rights require explicit approval and a +reviewed source-policy change before a manual-main publication request can pass. +The schedule remains artifact-only. See the [English workflow boundary](../README.md#theme-snapshot-workflow-and-draft-pr-boundary) +and [中文说明](../README.zh-CN.md#主题快照-workflow-与草稿-pr-边界). ## Main Design Pressure diff --git a/scripts/publish_theme_snapshot_pr.py b/scripts/publish_theme_snapshot_pr.py new file mode 100644 index 0000000..52b08fa --- /dev/null +++ b/scripts/publish_theme_snapshot_pr.py @@ -0,0 +1,617 @@ +#!/usr/bin/env python3 +"""Prepare one same-run theme JSON for a normal draft PR; publication is disabled.""" +from __future__ import annotations + +import argparse +import base64 +import datetime as dt +import hashlib +import json +import math +import os +from pathlib import Path +import re +import stat +import subprocess +import sys +from typing import Any +import urllib.error +import urllib.parse +import urllib.request + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT / "src")) +from research_signal_context_pipelines.theme_momentum import ( # noqa: E402 + DEFAULT_TOP_SYMBOLS, + build_theme_momentum_snapshot, + validate_theme_momentum_snapshot, +) +from research_signal_context_pipelines.theme_universe import ( # noqa: E402 + load_symbol_theme_exposure, + load_theme_taxonomy, +) + +REPOSITORY = "QuantStrategyLab/ResearchSignalContextPipelines" +REPOSITORY_ID = "1252248184" +WORKFLOW_REF = f"{REPOSITORY}/.github/workflows/theme_momentum_snapshot.yml@refs/heads/main" +DATA_PATH = "data/output/theme_momentum_snapshot.json" +CANDIDATE_NAME = "theme_momentum_snapshot.json" +RECEIPT_NAME = "publication-package.json" +CONFIG_PATHS = ("config/theme_taxonomy.csv", "config/symbol_theme_exposure.csv") +MAX_BYTES = 128 * 1024 +# Source-reviewed rights/content approval is required before changing this gate. +# A workflow input, environment variable, App grant or earlier one-off import is not approval. +PUBLICATION_CONTENT_APPROVED = False +PUBLIC_PRICE_SOURCE = "yahoo_chart_download" +SHA = re.compile(r"[0-9a-f]{40}\Z") +SHA256 = re.compile(r"[0-9a-f]{64}\Z") +POSITIVE = re.compile(r"[1-9][0-9]*\Z") + + +ERROR_CODES = frozenset(""" + API_ALREADY_EXISTS API_METHOD_FORBIDDEN API_PAGINATION_UNEXPECTED API_RESPONSE_OVERSIZED API_TARGET_FORBIDDEN + API_TRANSPORT_FAILED API_WRITE_FORBIDDEN APP_IDENTITY_MISMATCH APP_SCOPE_MISMATCH + ARGUMENTS_REQUIRED ARTIFACT_BINDING_INVALID BASE_INVALID BASE_MOVED + BASE_PATH_INVALID BRANCH_CREATION_UNCONFIRMED BRANCH_MODIFIED CANDIDATE_EXPIRED + CANDIDATE_PATH_INVALID CHECKOUT_SHA_MISMATCH CONFIG_HASH_MISMATCH CONTENT_APPROVAL_REQUIRED + COUNT_INVALID COUNT_MISMATCH COVERAGE_MISMATCH DATE_INVALID + FILE_NOT_REGULAR FILE_OVERSIZED GENERATION_TIME_INVALID JSON_DUPLICATE_KEY + JSON_INVALID JSON_NONFINITE NUMBER_INVALID NUMBER_RANGE + PACKAGE_CANDIDATE_MISMATCH PACKAGE_CONTEXT_MISMATCH PACKAGE_HASH_INVALID PACKAGE_HASH_MISMATCH + PACKAGE_NOT_EMPTY PACKAGE_PATHS_INVALID PACKAGE_PATH_INVALID PACKAGE_SCHEMA_INVALID + PRICE_SOURCE_NOT_ADMITTED PR_CREATION_UNCONFIRMED PR_HUMAN_STATE_CHANGED PR_IDENTITY_AMBIGUOUS + PR_IDENTITY_MISMATCH PR_WRITE_FORBIDDEN PUBLICATION_FAILED RECEIPT_NONCANONICAL + REF_WRITE_FORBIDDEN REMOTE_BLOB_MISMATCH REMOTE_COMMIT_INVALID REMOTE_REF_INVALID + REMOTE_TREE_INVALID RESEARCH_CONTRACT_MISMATCH RESULT_REQUIRED RUN_INVALID + SCHEMA_KEYS SCOPE_INVALID SOURCE_SHA_INVALID SUMMARY_MISMATCH + SYMBOL_SCOPE_INVALID THEME_SCHEMA_INVALID THEME_SCOPE_INVALID TIME_INVALID + TOKEN_UNAVAILABLE TREE_WRITE_FORBIDDEN UNTRUSTED_CONTEXT VALIDATION_OR_API_FAILED +""".split()) + + +class PublicationError(RuntimeError): + """Only fixed, non-sensitive error codes reach logs or receipts.""" + + def __init__(self, code: str): + self.code = code if code in ERROR_CODES or re.fullmatch(r"API_HTTP_[1-5][0-9]{2}", code) else "PUBLICATION_FAILED" + super().__init__(self.code) + + +def require(condition: bool, code: str) -> None: + if not condition: + raise PublicationError(code) + + +def sha256(raw: bytes) -> str: + return hashlib.sha256(raw).hexdigest() + + +def git_blob_sha(raw: bytes) -> str: + return hashlib.sha1(b"blob " + str(len(raw)).encode("ascii") + b"\0" + raw).hexdigest() + + +def json_bytes(value: object) -> bytes: + return (json.dumps(value, indent=2, sort_keys=True, allow_nan=False) + "\n").encode("utf-8") + + +def parse_json(raw: bytes) -> Any: + def pairs(items): + result = {} + for key, value in items: + require(key not in result, "JSON_DUPLICATE_KEY") + result[key] = value + return result + + def invalid(_): + raise PublicationError("JSON_NONFINITE") + + try: + return json.loads(raw, object_pairs_hook=pairs, parse_constant=invalid) + except PublicationError: + raise + except (ValueError, UnicodeError, RecursionError): + raise PublicationError("JSON_INVALID") from None + + +def regular_bytes(path: Path, *, limit: int = MAX_BYTES) -> bytes: + require(path.exists() and not path.is_symlink() and stat.S_ISREG(path.stat().st_mode), "FILE_NOT_REGULAR") + require(path.stat().st_size <= limit, "FILE_OVERSIZED") + raw = path.read_bytes() + require(len(raw) <= limit, "FILE_OVERSIZED") + return raw + + +def keys(value: object, expected: str) -> dict: + require(type(value) is dict and set(value) == set(expected.split()), "SCHEMA_KEYS") + return value + + +def same_typed(actual: object, expected: object) -> bool: + if type(actual) is not type(expected): + return False + if isinstance(expected, dict): + return set(actual) == set(expected) and all(same_typed(actual[k], v) for k, v in expected.items()) + if isinstance(expected, list): + return len(actual) == len(expected) and all(same_typed(a, b) for a, b in zip(actual, expected)) + return actual == expected + + +def date(value: object) -> dt.date: + try: + require(type(value) is str, "DATE_INVALID") + parsed = dt.date.fromisoformat(value) + require(parsed.isoformat() == value, "DATE_INVALID") + return parsed + except (ValueError, TypeError): + raise PublicationError("DATE_INVALID") from None + + +def instant(value: object) -> dt.datetime: + try: + require(type(value) is str, "TIME_INVALID") + parsed = dt.datetime.fromisoformat(value.replace("Z", "+00:00")) + require(parsed.tzinfo is not None and parsed.utcoffset() == dt.timedelta(0), "TIME_INVALID") + return parsed + except (ValueError, TypeError): + raise PublicationError("TIME_INVALID") from None + + +def number(value: object, low: float | None = None, high: float | None = None, *, nullable: bool = True) -> None: + if value is None and nullable: + return + require(type(value) in (int, float) and math.isfinite(value), "NUMBER_INVALID") + require((low is None or value >= low) and (high is None or value <= high), "NUMBER_RANGE") + + +def count(value: object, expected: int | None = None) -> None: + require(type(value) is int and value >= 0, "COUNT_INVALID") + require(expected is None or value == expected, "COUNT_MISMATCH") + + +def string_set(value: object, allowed: set[str]) -> set[str]: + require(type(value) is list and all(type(x) is str for x in value), "SCOPE_INVALID") + require(value == sorted(set(value)) and set(value) <= allowed, "SCOPE_INVALID") + return set(value) + + +def context_from_env(env: dict[str, str] | None = None) -> dict[str, str]: + env = os.environ if env is None else env + require(env.get("GITHUB_EVENT_NAME") == "workflow_dispatch", "UNTRUSTED_CONTEXT") + require(env.get("GITHUB_REF") == "refs/heads/main", "UNTRUSTED_CONTEXT") + require(env.get("GITHUB_REPOSITORY") == REPOSITORY, "UNTRUSTED_CONTEXT") + require(env.get("GITHUB_REPOSITORY_ID") == REPOSITORY_ID, "UNTRUSTED_CONTEXT") + require(env.get("GITHUB_WORKFLOW_REF") == WORKFLOW_REF, "UNTRUSTED_CONTEXT") + source_sha = env.get("GITHUB_SHA", "") + require(bool(SHA.fullmatch(source_sha)), "SOURCE_SHA_INVALID") + for name in ("GITHUB_RUN_ID", "GITHUB_RUN_ATTEMPT"): + require(bool(POSITIVE.fullmatch(env.get(name, ""))), "RUN_INVALID") + return { + "repository": REPOSITORY, "repository_id": REPOSITORY_ID, + "source_sha": source_sha, "expected_base_sha": source_sha, + "workflow_ref": WORKFLOW_REF, + "run_id": env["GITHUB_RUN_ID"], "run_attempt": env["GITHUB_RUN_ATTEMPT"], + } + + +def admission(context: dict[str, str]) -> None: + keys(context, "repository repository_id source_sha expected_base_sha workflow_ref run_id run_attempt") + require(all(type(value) is str for value in context.values()), "UNTRUSTED_CONTEXT") + require(context["repository_id"] == REPOSITORY_ID, "UNTRUSTED_CONTEXT") + require(bool(SHA.fullmatch(context["source_sha"])) and context["source_sha"] == context["expected_base_sha"], "SOURCE_SHA_INVALID") + require(bool(POSITIVE.fullmatch(context["run_id"])) and bool(POSITIVE.fullmatch(context["run_attempt"])), "RUN_INVALID") + # This is deliberately not overridable by CLI inputs or environment variables. + require(PUBLICATION_CONTENT_APPROVED is True, "CONTENT_APPROVAL_REQUIRED") + require(context["repository"] == REPOSITORY and context["workflow_ref"] == WORKFLOW_REF, "UNTRUSTED_CONTEXT") + + +def verify_checkout(context: dict[str, str], root: Path = ROOT) -> None: + result = subprocess.run(["git", "rev-parse", "HEAD"], cwd=root, capture_output=True, text=True, check=False) + require(result.returncode == 0 and result.stdout.strip() == context["source_sha"], "CHECKOUT_SHA_MISMATCH") + + +def config_hashes(root: Path) -> dict[str, str]: + return {path: sha256(regular_bytes(root / path)) for path in CONFIG_PATHS} + + +def validate_candidate(raw: bytes, root: Path, started_at: str, completed_at: str, *, now: dt.datetime | None = None) -> dict: + require(len(raw) <= MAX_BYTES, "FILE_OVERSIZED") + candidate = keys(parse_json(raw), "schema_version as_of generated_at expires_at model_version scoring_version mode artifact_type horizon horizon_window horizon_window_label taxonomy_version methodology summary theme_ranks data_quality policy source_artifacts") + try: + validate_theme_momentum_snapshot(candidate) + except (ValueError, TypeError, KeyError): + raise PublicationError("THEME_SCHEMA_INVALID") from None + as_of = date(candidate["as_of"]) + generated = instant(candidate["generated_at"]) + started, completed = instant(started_at), instant(completed_at) + now = now or dt.datetime.now(dt.UTC) + require(started <= generated <= completed <= now and as_of <= generated.date(), "GENERATION_TIME_INVALID") + require(now.date() <= date(candidate["expires_at"]), "CANDIDATE_EXPIRED") + themes = load_theme_taxonomy(root / CONFIG_PATHS[0]) + exposures = load_symbol_theme_exposure(root / CONFIG_PATHS[1], known_theme_ids=themes) + template = build_theme_momentum_snapshot([], themes=themes, exposures=exposures, as_of=as_of, generated_at=generated) + for field in ("schema_version", "expires_at", "model_version", "scoring_version", "mode", "artifact_type", "horizon", "horizon_window", "horizon_window_label", "taxonomy_version", "methodology", "policy"): + require(same_typed(candidate[field], template[field]), "RESEARCH_CONTRACT_MISMATCH") + require(same_typed(candidate["source_artifacts"], { + "prices": PUBLIC_PRICE_SOURCE, "theme_taxonomy": CONFIG_PATHS[0], "theme_exposures": CONFIG_PATHS[1], + }), "PRICE_SOURCE_NOT_ADMITTED") + quality = keys(candidate["data_quality"], "coverage missing_price_symbols insufficient_history_symbols unranked_themes") + universe = set(exposures) + missing = string_set(quality["missing_price_symbols"], universe) + priced = universe - missing + insufficient = string_set(quality["insufficient_history_symbols"], priced) + unranked = string_set(quality["unranked_themes"], set(themes)) + coverage = keys(quality["coverage"], "configured_symbol_count priced_symbol_count price_coverage_ratio insufficient_history_symbol_count") + count(coverage["configured_symbol_count"], len(universe)) + count(coverage["priced_symbol_count"], len(priced)) + count(coverage["insufficient_history_symbol_count"], len(insufficient)) + number(coverage["price_coverage_ratio"], 0, 1, nullable=False) + require(coverage["price_coverage_ratio"] == round(len(priced) / len(universe), 6), "COVERAGE_MISMATCH") + ranks = candidate["theme_ranks"] + require(type(ranks) is list and len(ranks) <= len(themes), "THEME_SCOPE_INVALID") + seen = set() + for index, rank in enumerate(ranks, 1): + keys(rank, "theme_id theme_name sector horizon rank momentum_score breadth_3m component_count priced_symbol_count returns risk top_symbols source_policy") + theme_id = rank["theme_id"] + require(type(theme_id) is str and theme_id in themes and theme_id not in seen, "THEME_SCOPE_INVALID") + seen.add(theme_id) + theme = themes[theme_id] + for field in ("theme_name", "sector", "horizon", "source_policy"): + require(type(rank[field]) is str and rank[field] == getattr(theme, field), "THEME_SCOPE_INVALID") + members = {symbol for symbol, exposure in exposures.items() if theme_id in exposure.theme_ids} + priced_members = members & priced + require(bool(priced_members), "THEME_SCOPE_INVALID") + count(rank["rank"], index) + count(rank["component_count"], len(members)) + count(rank["priced_symbol_count"], len(priced_members)) + number(rank["momentum_score"]) + number(rank["breadth_3m"], 0, 1) + for value in keys(rank["returns"], "3m 6_1m 12_1m").values(): + number(value, -1) + risk = keys(rank["risk"], "realized_vol_63d drawdown_126d") + number(risk["realized_vol_63d"], 0) + number(risk["drawdown_126d"], -1, 0) + top = rank["top_symbols"] + require(type(top) is list and len(top) == min(DEFAULT_TOP_SYMBOLS, len(priced_members)), "SYMBOL_SCOPE_INVALID") + symbols = set() + for symbol in top: + keys(symbol, "symbol momentum_score return_3m return_6_1m return_12_1m") + name = symbol["symbol"] + require(type(name) is str and name in priced_members and name not in symbols, "SYMBOL_SCOPE_INVALID") + symbols.add(name) + number(symbol["momentum_score"]) + for field in ("return_3m", "return_6_1m", "return_12_1m"): + number(symbol[field], -1) + expected_ranked = {theme_id for theme_id in themes if any(theme_id in exposures[symbol].theme_ids for symbol in priced)} + require(seen == expected_ranked and unranked == set(themes) - seen, "THEME_SCOPE_INVALID") + summary = keys(candidate["summary"], "ranked_theme_count priced_symbol_count top_theme_ids") + count(summary["ranked_theme_count"], len(ranks)) + count(summary["priced_symbol_count"], len(priced)) + require(summary["top_theme_ids"] == [rank["theme_id"] for rank in ranks[:5]], "SUMMARY_MISMATCH") + return candidate + + +def prepare_package(candidate: Path, directory: Path, context: dict, started_at: str, completed_at: str, *, root: Path = ROOT, now=None) -> dict: + admission(context) + candidate = candidate.absolute() + expected = root.absolute() / DATA_PATH + require(candidate == expected, "CANDIDATE_PATH_INVALID") + require(not any(path.is_symlink() for path in (candidate, candidate.parent, candidate.parent.parent)), "FILE_NOT_REGULAR") + raw = regular_bytes(candidate) + validate_candidate(raw, root, started_at, completed_at, now=now) + require(not directory.is_symlink(), "PACKAGE_PATH_INVALID") + directory.mkdir(parents=True, exist_ok=True) + require(not any(directory.iterdir()), "PACKAGE_NOT_EMPTY") + receipt = { + "schema_version": 1, "kind": "rscp_theme_publication", **context, + "generation_started_at": started_at, "generation_completed_at": completed_at, + "candidate": {"path": DATA_PATH, "sha256": sha256(raw), "size_bytes": len(raw)}, + "config_sha256": config_hashes(root), + } + receipt_raw = json_bytes(receipt) + (directory / CANDIDATE_NAME).write_bytes(raw) + (directory / RECEIPT_NAME).write_bytes(receipt_raw) + return {"candidate_sha256": sha256(raw), "receipt_sha256": sha256(receipt_raw)} + + +def validate_package(directory: Path, context: dict, candidate_sha256: str, receipt_sha256: str, artifact_id: str, artifact_digest: str, *, root: Path = ROOT, now=None) -> tuple[bytes, dict]: + admission(context) + require(bool(SHA256.fullmatch(candidate_sha256)) and bool(SHA256.fullmatch(receipt_sha256)), "PACKAGE_HASH_INVALID") + require(bool(POSITIVE.fullmatch(artifact_id)) and bool(SHA256.fullmatch(artifact_digest)), "ARTIFACT_BINDING_INVALID") + require(directory.is_dir() and not directory.is_symlink(), "PACKAGE_PATH_INVALID") + require({path.name for path in directory.iterdir()} == {CANDIDATE_NAME, RECEIPT_NAME}, "PACKAGE_PATHS_INVALID") + raw, receipt_raw = regular_bytes(directory / CANDIDATE_NAME), regular_bytes(directory / RECEIPT_NAME) + require(sha256(raw) == candidate_sha256 and sha256(receipt_raw) == receipt_sha256, "PACKAGE_HASH_MISMATCH") + receipt = parse_json(receipt_raw) + require(receipt_raw == json_bytes(receipt), "RECEIPT_NONCANONICAL") + validate_receipt(raw, receipt, context, root=root, now=now) + return raw, receipt + + +def validate_receipt(raw: bytes, receipt: dict, context: dict, *, root: Path = ROOT, now=None) -> None: + keys(receipt, "schema_version kind repository repository_id source_sha expected_base_sha workflow_ref run_id run_attempt generation_started_at generation_completed_at candidate config_sha256") + require(type(receipt["schema_version"]) is int and receipt["schema_version"] == 1 and receipt["kind"] == "rscp_theme_publication", "PACKAGE_SCHEMA_INVALID") + require(all(type(receipt.get(key)) is str and receipt[key] == value for key, value in context.items()), "PACKAGE_CONTEXT_MISMATCH") + require(same_typed(receipt["candidate"], {"path": DATA_PATH, "sha256": sha256(raw), "size_bytes": len(raw)}), "PACKAGE_CANDIDATE_MISMATCH") + require(same_typed(receipt["config_sha256"], config_hashes(root)), "CONFIG_HASH_MISMATCH") + validate_candidate(raw, root, receipt["generation_started_at"], receipt["generation_completed_at"], now=now) + + +class NoRedirect(urllib.request.HTTPRedirectHandler): + def redirect_request(self, req, fp, code, msg, headers, newurl): + return None + + +def already_exists_response(error: urllib.error.HTTPError, method: str, path: str, payload: object) -> bool: + """Recognize only bounded, exact conflicts for this attempted ref/PR.""" + if error.code != 422 or method != "POST" or type(payload) is not dict: + return False + if path not in {endpoint("git/refs"), endpoint("pulls")}: + return False + try: + raw = error.read(16384 + 1) + if len(raw) > 16384: + return False + value = parse_json(raw) + if type(value) is not dict: + return False + if path == endpoint("git/refs"): + return value.get("message") == "Reference already exists" + errors = value.get("errors") + expected = f"A pull request already exists for QuantStrategyLab:{payload.get('head')}." + return ( + value.get("message") == "Validation Failed" + and type(errors) is list and len(errors) == 1 and type(errors[0]) is dict + and errors[0].get("resource") == "PullRequest" + and errors[0].get("code") == "custom" + and errors[0].get("message") == expected + ) + except Exception: + return False + finally: + error.close() + + +def may_reconcile(error: PublicationError) -> bool: + return error.code in {"API_TRANSPORT_FAILED", "API_ALREADY_EXISTS"} or bool(re.fullmatch(r"API_HTTP_5[0-9]{2}", error.code)) + + +class GitHub: + """Fixed GitHub host; no credential persistence, redirects, raw errors or retries.""" + + def __init__(self, token: str, *, read_only: bool = False): + require(bool(token), "TOKEN_UNAVAILABLE") + self.token, self.read_only = token, read_only + + def request(self, method: str, path: str, payload=None, *, missing_ok=False, single_page=False): + require(method in {"GET", "POST"} and (not self.read_only or method == "GET"), "API_METHOD_FORBIDDEN") + require(path.startswith(f"/repos/{REPOSITORY}/") or path == "/installation/repositories?per_page=100", "API_TARGET_FORBIDDEN") + if method == "POST": + allowed = {endpoint(name) for name in ("git/blobs", "git/trees", "git/commits", "git/refs", "pulls")} + require(path in allowed and type(payload) is dict, "API_WRITE_FORBIDDEN") + if path == endpoint("git/refs"): + require(type(payload.get("ref")) is str and bool(re.fullmatch(r"refs/heads/automation/theme-[1-9][0-9]*-[1-9][0-9]*-[0-9a-f]{16}", payload["ref"])), "REF_WRITE_FORBIDDEN") + elif path == endpoint("git/trees"): + entries = payload.get("tree") + require(type(entries) is list and len(entries) == 1, "TREE_WRITE_FORBIDDEN") + entry = keys(entries[0], "path mode type sha") + require(entry["path"] == DATA_PATH and entry["mode"] == "100644" and entry["type"] == "blob", "TREE_WRITE_FORBIDDEN") + elif path == endpoint("pulls"): + require(payload.get("draft") is True and payload.get("base") == "main" and bool(re.fullmatch(r"automation/theme-[1-9][0-9]*-[1-9][0-9]*-[0-9a-f]{16}", str(payload.get("head", "")))), "PR_WRITE_FORBIDDEN") + request = urllib.request.Request( + "https://api.github.com" + path, + data=json.dumps(payload, allow_nan=False).encode() if payload is not None else None, + headers={"Authorization": f"Bearer {self.token}", "Accept": "application/vnd.github+json", "Content-Type": "application/json", "X-GitHub-Api-Version": "2022-11-28"}, + method=method, + ) + try: + with urllib.request.build_opener(NoRedirect()).open(request, timeout=20) as response: + require(not single_page or not response.headers.get("Link"), "API_PAGINATION_UNEXPECTED") + raw = response.read(1024 * 1024 + 1) + require(len(raw) <= 1024 * 1024, "API_RESPONSE_OVERSIZED") + return parse_json(raw) + except urllib.error.HTTPError as error: + if missing_ok and error.code == 404: + return None + if already_exists_response(error, method, path, payload): + raise PublicationError("API_ALREADY_EXISTS") from None + raise PublicationError(f"API_HTTP_{error.code}") from None + except (urllib.error.URLError, OSError, TimeoutError): + raise PublicationError("API_TRANSPORT_FAILED") from None + + +def endpoint(suffix: str) -> str: + return f"/repos/{REPOSITORY}/{suffix}" + + +def assert_base(api: GitHub, context: dict) -> None: + ref = api.request("GET", endpoint("git/ref/heads/main")) + require(type(ref) is dict and ref.get("object", {}).get("sha") == context["expected_base_sha"], "BASE_MOVED") + + +def verify_app_scope(api: GitHub, installation_id: str, app_slug: str) -> None: + require(installation_id == "168603414" and bool(re.fullmatch(r"[a-z0-9-]+", app_slug)), "APP_IDENTITY_MISMATCH") + scope = api.request("GET", "/installation/repositories?per_page=100", single_page=True) + require(type(scope) is dict and type(scope.get("total_count")) is int and scope["total_count"] == 1, "APP_SCOPE_MISMATCH") + repos = scope.get("repositories") + require(type(repos) is list and len(repos) == 1 and type(repos[0]) is dict, "APP_SCOPE_MISMATCH") + require(repos[0].get("full_name") == REPOSITORY and type(repos[0].get("id")) is int and repos[0]["id"] == int(REPOSITORY_ID), "APP_SCOPE_MISMATCH") + + +def leaf_tree(api: GitHub, tree_sha: str) -> dict: + require(bool(SHA.fullmatch(tree_sha)), "REMOTE_TREE_INVALID") + response = api.request("GET", endpoint(f"git/trees/{tree_sha}?recursive=1")) + require(response.get("truncated") is False and type(response.get("tree")) is list, "REMOTE_TREE_INVALID") + leaves = [entry for entry in response["tree"] if entry.get("type") != "tree"] + require(len({entry.get("path") for entry in leaves}) == len(leaves), "REMOTE_TREE_INVALID") + return {entry["path"]: (entry.get("mode"), entry.get("type"), entry.get("sha")) for entry in leaves} + + +def assert_candidate_commit(api: GitHub, head: str, base: str, expected_tree: dict, message: str) -> None: + commit = api.request("GET", endpoint(f"git/commits/{head}")) + require(commit.get("sha") == head and [parent.get("sha") for parent in commit.get("parents", [])] == [base], "BRANCH_MODIFIED") + require(commit.get("message") == message, "BRANCH_MODIFIED") + require(leaf_tree(api, commit.get("tree", {}).get("sha", "")) == expected_tree, "BRANCH_MODIFIED") + + +def find_pr(api: GitHub, branch: str, head: str, marker: str, app_slug: str): + query = urllib.parse.urlencode({"state": "all", "head": f"QuantStrategyLab:{branch}", "base": "main", "per_page": "100"}) + results = api.request("GET", endpoint("pulls?" + query), single_page=True) + require(type(results) is list and len(results) <= 1, "PR_IDENTITY_AMBIGUOUS") + if not results: + return None + return validate_pr(results[0], branch, head, marker, app_slug) + + +def validate_pr(pr: dict, branch: str, head: str, marker: str, app_slug: str) -> dict: + require(pr.get("state") == "open" and pr.get("draft") is True and not pr.get("merged_at"), "PR_HUMAN_STATE_CHANGED") + require(pr.get("head", {}).get("ref") == branch and pr["head"].get("sha") == head and pr.get("base", {}).get("ref") == "main", "PR_IDENTITY_MISMATCH") + for side in ("head", "base"): + require(pr[side].get("repo", {}).get("full_name") == REPOSITORY and pr[side]["repo"].get("id") == int(REPOSITORY_ID), "PR_IDENTITY_MISMATCH") + require(pr.get("user", {}).get("login") == app_slug + "[bot]" and pr["user"].get("type") == "Bot", "PR_IDENTITY_MISMATCH") + require(type(pr.get("body")) is str and pr["body"].count(marker) == 1, "PR_IDENTITY_MISMATCH") + require(pr["body"].count(f"") == 1, "PR_IDENTITY_MISMATCH") + require(type(pr.get("number")) is int and pr["number"] > 0, "PR_IDENTITY_MISMATCH") + require(pr.get("html_url") == f"https://github.com/{REPOSITORY}/pull/{pr['number']}", "PR_IDENTITY_MISMATCH") + return pr + + +def publish_candidate(api: GitHub, raw: bytes, receipt: dict, context: dict, artifact_id: str, artifact_digest: str, app_slug: str, installation_id: str, result: dict, *, root: Path = ROOT, now=None, receipt_sha256: str | None = None) -> dict: + admission(context) + validate_receipt(raw, receipt, context, root=root, now=now) + require(bool(POSITIVE.fullmatch(artifact_id)) and bool(SHA256.fullmatch(artifact_digest)), "ARTIFACT_BINDING_INVALID") + canonical_receipt_sha256 = sha256(json_bytes(receipt)) + require(receipt_sha256 is None or receipt_sha256 == canonical_receipt_sha256, "PACKAGE_HASH_MISMATCH") + receipt_sha256 = canonical_receipt_sha256 if receipt_sha256 is None else receipt_sha256 + result.update(candidate_sha256=sha256(raw), receipt_sha256=receipt_sha256, artifact_id=artifact_id, artifact_digest=artifact_digest, **context) + verify_app_scope(api, installation_id, app_slug) + assert_base(api, context) + base = context["expected_base_sha"] + base_commit = api.request("GET", endpoint(f"git/commits/{base}")) + require(base_commit.get("sha") == base, "BASE_INVALID") + base_tree_sha = base_commit.get("tree", {}).get("sha", "") + expected_tree = leaf_tree(api, base_tree_sha) + require(DATA_PATH in expected_tree and expected_tree[DATA_PATH][:2] == ("100644", "blob"), "BASE_PATH_INVALID") + blob = git_blob_sha(raw) + if expected_tree[DATA_PATH][2] == blob: + result.update(status="already_on_base", main_contains_candidate=True) + return result + expected_tree[DATA_PATH] = ("100644", "blob", blob) + package_key = sha256(json_bytes({**context, "candidate_sha256": sha256(raw)})) + branch = f"automation/theme-{context['run_id']}-{context['run_attempt']}-{package_key[:16]}" + marker = f"" + commit_message = f"Update reviewed research theme snapshot\n\nResearch-Package: {package_key}" + expected_new_head = None + result.update(branch=branch, package_key=package_key, candidate_sha256=sha256(raw)) + ref_path = endpoint("git/ref/heads/" + branch) + ref = api.request("GET", ref_path, missing_ok=True) + if ref is None: + created_blob = api.request("POST", endpoint("git/blobs"), {"encoding": "base64", "content": base64.b64encode(raw).decode("ascii")}) + require(created_blob.get("sha") == blob, "REMOTE_BLOB_MISMATCH") + tree = api.request("POST", endpoint("git/trees"), {"base_tree": base_tree_sha, "tree": [{"path": DATA_PATH, "mode": "100644", "type": "blob", "sha": blob}]}) + require(bool(SHA.fullmatch(tree.get("sha", ""))), "REMOTE_TREE_INVALID") + commit = api.request("POST", endpoint("git/commits"), {"message": commit_message, "tree": tree["sha"], "parents": [base]}) + head = commit.get("sha", "") + require(bool(SHA.fullmatch(head)), "REMOTE_COMMIT_INVALID") + expected_new_head = head + assert_candidate_commit(api, head, base, expected_tree, commit_message) + assert_base(api, context) + result["branch_created"] = None + try: + api.request("POST", endpoint("git/refs"), {"ref": "refs/heads/" + branch, "sha": head}) + except PublicationError as error: + # Only unknown outcomes / exact already-exists conflicts permit one read. + if not may_reconcile(error): + if re.fullmatch(r"API_HTTP_4[0-9]{2}", error.code): + result["branch_created"] = False + raise + ref = api.request("GET", ref_path, missing_ok=True) + if ref is None: + result["branch_created"] = False + raise PublicationError("BRANCH_CREATION_UNCONFIRMED") + head = ref.get("object", {}).get("sha", "") + require(bool(SHA.fullmatch(head)), "REMOTE_REF_INVALID") + require(expected_new_head is None or head == expected_new_head, "BRANCH_MODIFIED") + assert_candidate_commit(api, head, base, expected_tree, commit_message) + result.update(branch_created=True, head_sha=head) + assert_base(api, context) + pr = find_pr(api, branch, head, marker, app_slug) + reused = pr is not None + if pr is None: + body = "\n".join([ + marker, f"", + "Research-only theme context; human review and required CI remain mandatory.", + f"Source/base: {base}", f"Run: https://github.com/{REPOSITORY}/actions/runs/{context['run_id']}", + f"Run attempt: {context['run_attempt']}", f"Candidate raw SHA256: {sha256(raw)}", + f"Receipt raw SHA256: {receipt_sha256}", + f"Same-run artifact ID: {artifact_id}; archive SHA256: {artifact_digest}", + "State: draft PR prepared; published=false until separately reviewed and merged.", + ]) + result["pr_created"] = None + try: + response = api.request("POST", endpoint("pulls"), {"title": "Update reviewed research theme snapshot", "head": branch, "base": "main", "body": body, "draft": True}) + except PublicationError as error: + if not may_reconcile(error): + if re.fullmatch(r"API_HTTP_4[0-9]{2}", error.code): + result["pr_created"] = False + raise + pr = find_pr(api, branch, head, marker, app_slug) + else: + # A local response-validation failure is not an unknown write outcome. + pr = validate_pr(response, branch, head, marker, app_slug) + require(pr is not None, "PR_CREATION_UNCONFIRMED") + result.update(status="draft_pr_reused" if reused else "draft_pr_created", pr_created=True, pr_url=pr["html_url"], review_required=True) + assert_base(api, context) + return result + + +def initial_result() -> dict: + return {"status": "not_run", "reason_code": None, "published": False, "branch_created": False, "pr_created": False, "review_required": True} + + +def main(argv=None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("mode", choices=("admission", "prepare", "validate", "publish")) + parser.add_argument("--candidate", type=Path) + parser.add_argument("--package-dir", type=Path) + parser.add_argument("--started-at") + parser.add_argument("--completed-at") + parser.add_argument("--candidate-sha256") + parser.add_argument("--receipt-sha256") + parser.add_argument("--artifact-id") + parser.add_argument("--artifact-digest") + parser.add_argument("--result", type=Path) + args = parser.parse_args(argv) + result = initial_result() + try: + context = context_from_env() + admission(context) + verify_checkout(context) + if args.mode == "admission": + result["status"] = "admitted" + elif args.mode == "prepare": + require(all((args.candidate, args.package_dir, args.started_at, args.completed_at)), "ARGUMENTS_REQUIRED") + outputs = prepare_package(args.candidate, args.package_dir, context, args.started_at, args.completed_at) + with Path(os.environ["GITHUB_OUTPUT"]).open("a", encoding="utf-8") as output: + for key, value in outputs.items(): + output.write(f"{key}={value}\n") + result.update(status="package_prepared", **outputs) + else: + require(all((args.package_dir, args.candidate_sha256, args.receipt_sha256, args.artifact_id, args.artifact_digest)), "ARGUMENTS_REQUIRED") + raw, receipt = validate_package(args.package_dir, context, args.candidate_sha256, args.receipt_sha256, args.artifact_id, args.artifact_digest) + if args.mode == "validate": + assert_base(GitHub(os.environ.get("GITHUB_TOKEN", ""), read_only=True), context) + result["status"] = "package_validated" + else: + require(args.result is not None, "RESULT_REQUIRED") + publish_candidate(GitHub(os.environ.get("RESEARCH_PUBLISHER_TOKEN", "")), raw, receipt, context, args.artifact_id, args.artifact_digest, os.environ.get("RESEARCH_PUBLISHER_APP_SLUG", ""), os.environ.get("RESEARCH_PUBLISHER_INSTALLATION_ID", ""), result, receipt_sha256=args.receipt_sha256) + except Exception as error: + result.update(status="failed", reason_code=error.code if isinstance(error, PublicationError) else "VALIDATION_OR_API_FAILED") + if args.result is not None: + args.result.parent.mkdir(parents=True, exist_ok=True) + args.result.write_bytes(json_bytes(result)) + print(json.dumps(result, sort_keys=True)) + return 1 if result["status"] == "failed" else 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_publish_theme_snapshot_pr.py b/tests/test_publish_theme_snapshot_pr.py new file mode 100644 index 0000000..4668fcb --- /dev/null +++ b/tests/test_publish_theme_snapshot_pr.py @@ -0,0 +1,1076 @@ +"""Source-only tests: synthetic prices and in-memory GitHub, never a live write.""" +from __future__ import annotations + +import base64 +import copy +import datetime as dt +import io +import json +from pathlib import Path +from types import SimpleNamespace +import urllib.error + +import pytest + +from research_signal_context_pipelines.price_history import PriceRow +from research_signal_context_pipelines import theme_momentum +from scripts import publish_theme_snapshot_pr as publisher + + +BASE = "a" * 40 +BASE_TREE = "b" * 40 +NEW_TREE = "c" * 40 +HEAD = "d" * 40 +ARTIFACT_ID = "9876" +ARTIFACT_DIGEST = "f" * 64 +APP_SLUG = "synthetic-research-publisher" +INSTALLATION_ID = "168603414" +NOW = dt.datetime(2026, 10, 6, 12, tzinfo=dt.UTC) +STARTED = "2026-10-06T10:00:00Z" +COMPLETED = "2026-10-06T11:00:00Z" +TOKEN = "synthetic-test-token-never-real" + + +def trusted_env(): + return { + "GITHUB_EVENT_NAME": "workflow_dispatch", + "GITHUB_REF": "refs/heads/main", + "GITHUB_REPOSITORY": publisher.REPOSITORY, + "GITHUB_REPOSITORY_ID": publisher.REPOSITORY_ID, + "GITHUB_WORKFLOW_REF": publisher.WORKFLOW_REF, + "GITHUB_SHA": BASE, + "GITHUB_RUN_ID": "12345", + "GITHUB_RUN_ATTEMPT": "2", + } + + +@pytest.fixture(autouse=True) +def prohibit_network(monkeypatch): + def forbidden(*args, **kwargs): + raise AssertionError("Tests must not contact a network service") + + monkeypatch.setattr(publisher.urllib.request, "build_opener", forbidden) + monkeypatch.setattr(publisher.urllib.request, "urlopen", forbidden) + + +@pytest.fixture +def approved(monkeypatch): + # This is test-local only. Source and workflow remain fail-closed. + monkeypatch.setattr(publisher, "PUBLICATION_CONTENT_APPROVED", True) + + +@pytest.fixture +def source(tmp_path): + root = tmp_path / "source" + (root / "config").mkdir(parents=True) + (root / publisher.CONFIG_PATHS[0]).write_text( + "taxonomy_version,theme_id,theme_name,sector,horizon,description,source_policy\n" + "synthetic-v1,alpha,Alpha,technology,6-24 months,test,primary evidence required\n" + "synthetic-v1,beta,Beta,energy,6-24 months,test,primary evidence required\n" + ) + (root / publisher.CONFIG_PATHS[1]).write_text( + "symbol,theme_ids,exposure_confidence,rationale\n" + "AAA,alpha,high,synthetic\nBBB,alpha,high,synthetic\nCCC,beta,high,synthetic\n" + ) + themes = publisher.load_theme_taxonomy(root / publisher.CONFIG_PATHS[0]) + exposures = publisher.load_symbol_theme_exposure(root / publisher.CONFIG_PATHS[1], known_theme_ids=themes) + rows = [ + PriceRow(date=NOW.date() - dt.timedelta(days=279 - index), symbol=symbol, close=100 + step * index) + for symbol, step in (("AAA", 0.2), ("BBB", 0.1)) + for index in range(280) + ] + snapshot = publisher.build_theme_momentum_snapshot( + rows, themes=themes, exposures=exposures, as_of=NOW.date(), + generated_at=dt.datetime(2026, 10, 6, 10, 30, tzinfo=dt.UTC), + ) + snapshot["source_artifacts"] = { + "prices": publisher.PUBLIC_PRICE_SOURCE, + "theme_taxonomy": publisher.CONFIG_PATHS[0], + "theme_exposures": publisher.CONFIG_PATHS[1], + } + raw = publisher.json_bytes(snapshot) + candidate = root / publisher.DATA_PATH + candidate.parent.mkdir(parents=True) + candidate.write_bytes(raw) + return SimpleNamespace( + root=root, snapshot=snapshot, raw=raw, candidate=candidate, + context=publisher.context_from_env(trusted_env()), directory=tmp_path / "package", + ) + + +@pytest.fixture +def package(source, approved): + source.hashes = publisher.prepare_package( + source.candidate, source.directory, source.context, STARTED, COMPLETED, root=source.root, now=NOW, + ) + source.receipt = json.loads((source.directory / publisher.RECEIPT_NAME).read_bytes()) + return source + + +def validate(package, **overrides): + arguments = dict( + directory=package.directory, context=package.context, **package.hashes, + artifact_id=ARTIFACT_ID, artifact_digest=ARTIFACT_DIGEST, root=package.root, now=NOW, + ) + arguments.update(overrides) + return publisher.validate_package(**arguments) + + +def change(value, path, replacement): + for key in path[:-1]: + value = value[key] + value[path[-1]] = replacement + + +def rewrite_receipt(package, receipt): + raw = publisher.json_bytes(receipt) + (package.directory / publisher.RECEIPT_NAME).write_bytes(raw) + package.hashes["receipt_sha256"] = publisher.sha256(raw) + + +def test_default_gate_is_literal_false_and_env_cannot_approve(monkeypatch, tmp_path, capsys): + assert publisher.PUBLICATION_CONTENT_APPROVED is False + for key, value in {**trusted_env(), "PUBLICATION_CONTENT_APPROVED": "true", "PUBLISH_DRAFT_PR": "true"}.items(): + monkeypatch.setenv(key, value) + monkeypatch.setattr(publisher, "verify_checkout", lambda *_: pytest.fail("admission must run first")) + result_path = tmp_path / "result.json" + assert publisher.main(["publish", "--result", str(result_path)]) == 1 + result = json.loads(result_path.read_bytes()) + assert result["reason_code"] == "CONTENT_APPROVAL_REQUIRED" + assert result["published"] is False + assert result["branch_created"] is False and result["pr_created"] is False + assert "CONTENT_APPROVAL_REQUIRED" in capsys.readouterr().out + + +def test_cli_has_no_content_approval_override(): + with pytest.raises(SystemExit) as error: + publisher.main(["admission", "--publication-content-approved"]) + assert error.value.code == 2 + + +@pytest.mark.parametrize("operation", ["prepare", "validate", "publish"]) +def test_all_publication_entrypoints_gate_before_io(source, operation): + with pytest.raises(publisher.PublicationError, match="^CONTENT_APPROVAL_REQUIRED$"): + if operation == "prepare": + publisher.prepare_package(source.candidate, source.directory, source.context, STARTED, COMPLETED) + elif operation == "validate": + publisher.validate_package(source.directory, source.context, "", "", "", "") + else: + publisher.publish_candidate(None, b"", {}, source.context, "", "", "", "", {}) + assert not source.directory.exists() + + +@pytest.mark.parametrize("field,value,code", [ + ("GITHUB_EVENT_NAME", "pull_request", "UNTRUSTED_CONTEXT"), + ("GITHUB_REF", "refs/heads/topic", "UNTRUSTED_CONTEXT"), + ("GITHUB_REPOSITORY", "attacker/fork", "UNTRUSTED_CONTEXT"), + ("GITHUB_REPOSITORY_ID", "1", "UNTRUSTED_CONTEXT"), + ("GITHUB_WORKFLOW_REF", publisher.WORKFLOW_REF.replace("main", "topic"), "UNTRUSTED_CONTEXT"), + ("GITHUB_SHA", "A" * 40, "SOURCE_SHA_INVALID"), + ("GITHUB_SHA", "abc123", "SOURCE_SHA_INVALID"), + ("GITHUB_RUN_ID", "0", "RUN_INVALID"), + ("GITHUB_RUN_ID", "1\n", "RUN_INVALID"), + ("GITHUB_RUN_ATTEMPT", "01", "RUN_INVALID"), + ("GITHUB_RUN_ATTEMPT", "", "RUN_INVALID"), +]) +def test_context_rejects_untrusted_event_identity_and_run(field, value, code): + env = {**trusted_env(), field: value} + with pytest.raises(publisher.PublicationError, match=f"^{code}$"): + publisher.context_from_env(env) + + +def test_checkout_must_equal_exact_source_commit(source, monkeypatch): + calls = [] + + def fake_run(command, **kwargs): + calls.append((command, kwargs)) + return SimpleNamespace(returncode=0, stdout=HEAD + "\n") + + monkeypatch.setattr(publisher.subprocess, "run", fake_run) + with pytest.raises(publisher.PublicationError, match="CHECKOUT_SHA_MISMATCH"): + publisher.verify_checkout(source.context, root=source.root) + assert calls[0][0] == ["git", "rev-parse", "HEAD"] + assert calls[0][1]["cwd"] == source.root + + +@pytest.mark.parametrize("field,value", [("repository_id", "1"), ("expected_base_sha", HEAD), ("run_id", "01"), ("run_attempt", 2)]) +def test_direct_admission_also_rejects_invalid_context(source, approved, field, value): + context = {**source.context, field: value} + with pytest.raises(publisher.PublicationError): + publisher.admission(context) + + +def test_builder_candidate_package_roundtrip_preserves_raw_bytes(package): + raw, receipt = validate(package) + assert raw == package.raw == (package.directory / publisher.CANDIDATE_NAME).read_bytes() + assert receipt["source_sha"] == receipt["expected_base_sha"] == BASE + assert receipt["run_id"] == "12345" and receipt["run_attempt"] == "2" + assert receipt["candidate"] == {"path": publisher.DATA_PATH, "sha256": publisher.sha256(raw), "size_bytes": len(raw)} + assert receipt["config_sha256"] == publisher.config_hashes(package.root) + assert package.snapshot["data_quality"]["missing_price_symbols"] == ["CCC"] + assert package.snapshot["data_quality"]["unranked_themes"] == ["beta"] + + +@pytest.mark.parametrize("filename", [publisher.CANDIDATE_NAME, publisher.RECEIPT_NAME]) +def test_package_hash_binds_raw_bytes_not_equivalent_json(package, filename): + path = package.directory / filename + path.write_bytes(path.read_bytes() + b" \n") + with pytest.raises(publisher.PublicationError, match="PACKAGE_HASH_MISMATCH"): + validate(package) + + +def test_receipt_must_be_canonical_even_when_its_updated_hash_matches(package): + raw = json.dumps(package.receipt, separators=(",", ":")).encode() + (package.directory / publisher.RECEIPT_NAME).write_bytes(raw) + package.hashes["receipt_sha256"] = publisher.sha256(raw) + with pytest.raises(publisher.PublicationError, match="RECEIPT_NONCANONICAL"): + validate(package) + + +@pytest.mark.parametrize("field", ["repository", "repository_id", "workflow_ref", "source_sha", "expected_base_sha", "run_id", "run_attempt"]) +def test_receipt_is_bound_to_same_trusted_source_run_and_base(package, field): + receipt = copy.deepcopy(package.receipt) + receipt[field] += "changed" + rewrite_receipt(package, receipt) + with pytest.raises(publisher.PublicationError, match="PACKAGE_CONTEXT_MISMATCH"): + validate(package) + + +@pytest.mark.parametrize("path,value,code", [ + (("schema_version",), True, "PACKAGE_SCHEMA_INVALID"), + (("kind",), "other_package", "PACKAGE_SCHEMA_INVALID"), + (("candidate", "path"), "data/output/other.json", "PACKAGE_CANDIDATE_MISMATCH"), + (("candidate", "size_bytes"), 1, "PACKAGE_CANDIDATE_MISMATCH"), + (("candidate", "sha256"), "0" * 64, "PACKAGE_CANDIDATE_MISMATCH"), + (("config_sha256", publisher.CONFIG_PATHS[0]), "0" * 64, "CONFIG_HASH_MISMATCH"), +]) +def test_receipt_has_strict_candidate_and_config_bindings(package, path, value, code): + receipt = copy.deepcopy(package.receipt) + change(receipt, path, value) + rewrite_receipt(package, receipt) + with pytest.raises(publisher.PublicationError, match=code): + validate(package) + + +def test_receipt_rejects_additional_fields(package): + receipt = {**package.receipt, "override": True} + rewrite_receipt(package, receipt) + with pytest.raises(publisher.PublicationError, match="SCHEMA_KEYS"): + validate(package) + + +def test_config_change_after_preparation_is_rejected(package): + path = package.root / publisher.CONFIG_PATHS[1] + path.write_bytes(path.read_bytes() + b"DDD,beta,high,new synthetic member\n") + with pytest.raises(publisher.PublicationError, match="CONFIG_HASH_MISMATCH"): + validate(package) + + +@pytest.mark.parametrize("overrides,code", [ + ({"candidate_sha256": "A" * 64}, "PACKAGE_HASH_INVALID"), + ({"receipt_sha256": "bad"}, "PACKAGE_HASH_INVALID"), + ({"artifact_id": "0"}, "ARTIFACT_BINDING_INVALID"), + ({"artifact_id": "12\n"}, "ARTIFACT_BINDING_INVALID"), + ({"artifact_digest": "sha256:" + ARTIFACT_DIGEST}, "ARTIFACT_BINDING_INVALID"), +]) +def test_package_rejects_invalid_hash_and_artifact_binding_shapes(package, overrides, code): + with pytest.raises(publisher.PublicationError, match=code): + validate(package, **overrides) + + +@pytest.mark.parametrize("kind", ["extra_file", "extra_directory", "candidate_symlink", "receipt_symlink", "package_symlink"]) +def test_package_rejects_extra_paths_and_symlinks(package, tmp_path, kind): + if kind == "extra_file": + (package.directory / "extra.json").write_text("{}") + elif kind == "extra_directory": + (package.directory / "nested").mkdir() + elif kind == "package_symlink": + link = tmp_path / "package-link" + link.symlink_to(package.directory, target_is_directory=True) + package.directory = link + else: + name = publisher.CANDIDATE_NAME if kind == "candidate_symlink" else publisher.RECEIPT_NAME + path = package.directory / name + target = tmp_path / name + target.write_bytes(path.read_bytes()) + path.unlink() + path.symlink_to(target) + with pytest.raises(publisher.PublicationError, match="PACKAGE_PATHS_INVALID|PACKAGE_PATH_INVALID|FILE_NOT_REGULAR"): + validate(package) + + +def test_prepare_requires_canonical_candidate_and_empty_package(source, approved): + wrong = source.root / "other.json" + wrong.write_bytes(source.raw) + with pytest.raises(publisher.PublicationError, match="CANDIDATE_PATH_INVALID"): + publisher.prepare_package(wrong, source.directory, source.context, STARTED, COMPLETED, root=source.root, now=NOW) + source.directory.mkdir() + (source.directory / "old").write_text("keep") + with pytest.raises(publisher.PublicationError, match="PACKAGE_NOT_EMPTY"): + publisher.prepare_package(source.candidate, source.directory, source.context, STARTED, COMPLETED, root=source.root, now=NOW) + assert (source.directory / "old").read_text() == "keep" + + +def test_prepare_accepts_canonical_workflow_relative_path(source, approved, monkeypatch): + monkeypatch.chdir(source.root) + hashes = publisher.prepare_package( + Path(publisher.DATA_PATH), source.directory, source.context, STARTED, COMPLETED, root=source.root, now=NOW, + ) + assert hashes["candidate_sha256"] == publisher.sha256(source.raw) + + +@pytest.mark.parametrize("relative", ["data", "data/output"]) +def test_prepare_rejects_symlinked_candidate_parent(source, approved, tmp_path, relative): + original = source.root / relative + moved = tmp_path / "moved-data" + original.rename(moved) + original.symlink_to(moved, target_is_directory=True) + with pytest.raises(publisher.PublicationError, match="FILE_NOT_REGULAR"): + publisher.prepare_package(source.candidate, source.directory, source.context, STARTED, COMPLETED, root=source.root, now=NOW) + + +def test_config_symlink_is_rejected_even_when_bytes_match(package, tmp_path): + config = package.root / publisher.CONFIG_PATHS[0] + target = tmp_path / "same-config.csv" + target.write_bytes(config.read_bytes()) + config.unlink() + config.symlink_to(target) + with pytest.raises(publisher.PublicationError, match="FILE_NOT_REGULAR"): + validate(package) + + +def test_regular_bytes_rejects_oversized_candidate(source): + source.candidate.write_bytes(b" " * (publisher.MAX_BYTES + 1)) + with pytest.raises(publisher.PublicationError, match="FILE_OVERSIZED"): + publisher.regular_bytes(source.candidate) + + +@pytest.mark.parametrize("raw,code", [ + (b'{"key":1,"key":2}', "JSON_DUPLICATE_KEY"), + (b'{"nested":{"key":1,"key":2}}', "JSON_DUPLICATE_KEY"), + (b'{"n":NaN}', "JSON_NONFINITE"), + (b'{"n":Infinity}', "JSON_NONFINITE"), + (b'{"n":-Infinity}', "JSON_NONFINITE"), + (b'not JSON', "JSON_INVALID"), + (b'\xff', "JSON_INVALID"), +]) +def test_json_parser_rejects_ambiguous_or_nonfinite_data(raw, code): + with pytest.raises(publisher.PublicationError, match=code): + publisher.parse_json(raw) + + +@pytest.mark.parametrize("path,value", [ + (("schema_version",), 2), + (("policy", "execution_allowed"), True), + (("policy", "portfolio_allocation_allowed"), 0), + (("methodology", "volatility_penalty_weight"), 0), + (("source_artifacts", "prices"), "local-private-prices.csv"), + (("source_artifacts", "theme_taxonomy"), "other.csv"), + (("as_of",), "2026-10-07"), + (("generated_at",), "2026-10-06T10:30:00"), + (("generated_at",), "2026-10-06T10:30:00+08:00"), + (("generated_at",), "2026-10-06T09:59:59Z"), + (("generated_at",), "2026-10-06T11:00:01Z"), + (("expires_at",), "2026-10-05"), + (("expires_at",), "2027-01-01"), + (("summary", "ranked_theme_count"), True), + (("summary", "top_theme_ids"), ["beta"]), + (("data_quality", "coverage", "configured_symbol_count"), 999), + (("data_quality", "coverage", "priced_symbol_count"), True), + (("data_quality", "coverage", "price_coverage_ratio"), True), + (("data_quality", "coverage", "price_coverage_ratio"), 0.5), + (("data_quality", "missing_price_symbols"), ["OUTSIDE"]), + (("data_quality", "missing_price_symbols"), ["CCC", "CCC"]), + (("data_quality", "insufficient_history_symbols"), ["CCC"]), + (("data_quality", "unranked_themes"), []), + (("theme_ranks", 0, "theme_id"), "outside"), + (("theme_ranks", 0, "theme_name"), "changed"), + (("theme_ranks", 0, "source_policy"), "unrestricted"), + (("theme_ranks", 0, "rank"), True), + (("theme_ranks", 0, "component_count"), 1), + (("theme_ranks", 0, "momentum_score"), "0.5"), + (("theme_ranks", 0, "momentum_score"), True), + (("theme_ranks", 0, "breadth_3m"), 1.01), + (("theme_ranks", 0, "returns", "3m"), -1.01), + (("theme_ranks", 0, "risk", "realized_vol_63d"), -0.1), + (("theme_ranks", 0, "risk", "drawdown_126d"), 0.1), + (("theme_ranks", 0, "top_symbols", 0, "symbol"), "CCC"), + (("theme_ranks", 0, "top_symbols", 0, "return_3m"), False), + (("theme_ranks", 0, "top_symbols"), []), +]) +def test_candidate_rejects_changed_contract_numbers_dates_and_universe(source, path, value): + candidate = copy.deepcopy(source.snapshot) + change(candidate, path, value) + with pytest.raises(publisher.PublicationError): + publisher.validate_candidate(publisher.json_bytes(candidate), source.root, STARTED, COMPLETED, now=NOW) + + +@pytest.mark.parametrize("path", [(), ("summary",), ("policy",), ("data_quality",), ("theme_ranks", 0), ("theme_ranks", 0, "returns"), ("theme_ranks", 0, "top_symbols", 0)]) +def test_candidate_rejects_extra_keys_at_each_contract_level(source, path): + candidate = copy.deepcopy(source.snapshot) + value = candidate + for key in path: + value = value[key] + value["unexpected"] = "must not publish" + with pytest.raises(publisher.PublicationError): + publisher.validate_candidate(publisher.json_bytes(candidate), source.root, STARTED, COMPLETED, now=NOW) + + +def test_candidate_rejects_expired_and_future_completed_generation(source): + for now, completed, code in [ + (NOW + dt.timedelta(days=85), COMPLETED, "CANDIDATE_EXPIRED"), + (NOW, "2026-10-06T12:00:01Z", "GENERATION_TIME_INVALID"), + ]: + with pytest.raises(publisher.PublicationError, match=code): + publisher.validate_candidate(source.raw, source.root, STARTED, completed, now=now) + + +def test_builder_insufficient_history_remains_valid_with_nullable_metrics(source): + themes = publisher.load_theme_taxonomy(source.root / publisher.CONFIG_PATHS[0]) + exposures = publisher.load_symbol_theme_exposure(source.root / publisher.CONFIG_PATHS[1], known_theme_ids=themes) + rows = [PriceRow(date=NOW.date(), symbol=symbol, close=100) for symbol in ("AAA", "BBB")] + candidate = publisher.build_theme_momentum_snapshot( + rows, themes=themes, exposures=exposures, as_of=NOW.date(), generated_at=NOW.replace(hour=10, minute=30), + ) + candidate["source_artifacts"] = source.snapshot["source_artifacts"] + validated = publisher.validate_candidate(publisher.json_bytes(candidate), source.root, STARTED, COMPLETED, now=NOW) + assert validated["data_quality"]["insufficient_history_symbols"] == ["AAA", "BBB"] + assert validated["theme_ranks"][0]["momentum_score"] is None + + +def test_builder_default_now_preserves_microseconds_within_same_second_window(source, monkeypatch): + class FakeDatetime(dt.datetime): + @classmethod + def now(cls, tz=None): + return cls(2026, 10, 6, 10, 30, 0, 123456, tzinfo=tz) + + monkeypatch.setattr(theme_momentum, "dt", SimpleNamespace( + datetime=FakeDatetime, UTC=dt.UTC, date=dt.date, timedelta=dt.timedelta, + )) + themes = publisher.load_theme_taxonomy(source.root / publisher.CONFIG_PATHS[0]) + exposures = publisher.load_symbol_theme_exposure(source.root / publisher.CONFIG_PATHS[1], known_theme_ids=themes) + candidate = theme_momentum.build_theme_momentum_snapshot( + [PriceRow(date=NOW.date(), symbol="AAA", close=100)], + themes=themes, exposures=exposures, as_of=NOW.date(), + ) + candidate["source_artifacts"] = source.snapshot["source_artifacts"] + assert candidate["generated_at"] == "2026-10-06T10:30:00.123456Z" + validated = publisher.validate_candidate( + publisher.json_bytes(candidate), source.root, + "2026-10-06T10:30:00.123455Z", "2026-10-06T10:30:00.123457Z", now=NOW, + ) + assert validated["generated_at"] == candidate["generated_at"] + print("Real builder default-now generated_at=" + candidate["generated_at"]) + print("Exact same-second start=.123455Z < generated=.123456Z < completion=.123457Z: validated") + + +@pytest.mark.parametrize("duplicate", ["theme", "symbol"]) +def test_candidate_rejects_duplicate_ranked_themes_and_symbols(source, duplicate): + candidate = copy.deepcopy(source.snapshot) + rank = candidate["theme_ranks"][0] + if duplicate == "theme": + candidate["theme_ranks"].append(copy.deepcopy(rank)) + else: + rank["top_symbols"][1] = copy.deepcopy(rank["top_symbols"][0]) + with pytest.raises(publisher.PublicationError, match="THEME_SCOPE_INVALID|SYMBOL_SCOPE_INVALID"): + publisher.validate_candidate(publisher.json_bytes(candidate), source.root, STARTED, COMPLETED, now=NOW) + + +def test_candidate_rejects_nonfinite_number_created_by_json_exponent(source): + candidate = copy.deepcopy(source.snapshot) + candidate["theme_ranks"][0]["momentum_score"] = "overflow-number" + raw = publisher.json_bytes(candidate).replace(b'"overflow-number"', b"1e309") + with pytest.raises(publisher.PublicationError, match="NUMBER_INVALID"): + publisher.validate_candidate(raw, source.root, STARTED, COMPLETED, now=NOW) + + +class FakeGitHub: + """The one-repository Git Data / draft PR protocol, entirely in memory.""" + + def __init__(self, raw): + self.raw = raw + self.calls = [] + self.refs = {} + self.prs = [] + self.base_reads = 0 + self.move_base_on_read = None + self.fail_write = None + self.fail_code = "API_HTTP_403" + self.uncertain_write = None + self.uncertain_code = "API_TRANSPORT_FAILED" + self.scope = {"total_count": 1, "repositories": [{"full_name": publisher.REPOSITORY, "id": int(publisher.REPOSITORY_ID)}]} + self.trees = {BASE_TREE: [ + {"path": publisher.DATA_PATH, "mode": "100644", "type": "blob", "sha": "e" * 40}, + {"path": "README.md", "mode": "100644", "type": "blob", "sha": "1" * 40}, + ]} + self.commits = {BASE: {"sha": BASE, "tree": {"sha": BASE_TREE}, "parents": []}} + + def request(self, method, path, payload=None, **kwargs): + self.calls.append((method, path, copy.deepcopy(payload))) + assert method in {"GET", "POST"}, "No PATCH, DELETE, merge, or force update is allowed" + assert "/merge" not in path and "/dispatches" not in path + if method == "POST": + assert path in {publisher.endpoint("git/" + name) for name in ("blobs", "trees", "commits", "refs")} | {publisher.endpoint("pulls")} + if path.endswith("/refs"): + assert payload["ref"].startswith("refs/heads/automation/theme-") + assert payload["ref"] != "refs/heads/main" + if self.fail_write == path: + raise publisher.PublicationError(self.fail_code) + if path == "/installation/repositories?per_page=100": + assert method == "GET" and kwargs["single_page"] is True + return copy.deepcopy(self.scope) + prefix = publisher.endpoint("") + assert path.startswith(prefix) + route = path[len(prefix):] + if method == "GET": + if route == "git/ref/heads/main": + self.base_reads += 1 + sha = HEAD if self.move_base_on_read and self.base_reads >= self.move_base_on_read else BASE + return {"object": {"sha": sha}} + if route.startswith("git/ref/heads/"): + branch = route.removeprefix("git/ref/heads/") + return {"object": {"sha": self.refs[branch]}} if branch in self.refs else None + if route.startswith("git/commits/"): + return copy.deepcopy(self.commits[route.removeprefix("git/commits/")]) + if route.startswith("git/trees/"): + sha = route.removeprefix("git/trees/").removesuffix("?recursive=1") + return {"truncated": False, "tree": copy.deepcopy(self.trees[sha])} + if route.startswith("pulls?"): + assert "state=all" in route and kwargs["single_page"] is True + return copy.deepcopy(self.prs) + if method == "POST" and route == "git/blobs": + assert payload["encoding"] == "base64" + assert base64.b64decode(payload["content"], validate=True) == self.raw + return {"sha": publisher.git_blob_sha(self.raw)} + if method == "POST" and route == "git/trees": + assert payload["base_tree"] == BASE_TREE + assert payload["tree"] == [{"path": publisher.DATA_PATH, "mode": "100644", "type": "blob", "sha": publisher.git_blob_sha(self.raw)}] + self.trees[NEW_TREE] = copy.deepcopy( + [entry for entry in self.trees[BASE_TREE] if entry["path"] != publisher.DATA_PATH] + payload["tree"] + ) + return {"sha": NEW_TREE} + if method == "POST" and route == "git/commits": + assert payload["parents"] == [BASE] and payload["tree"] == NEW_TREE + self.commits[HEAD] = {"sha": HEAD, "tree": {"sha": NEW_TREE}, "parents": [{"sha": BASE}], "message": payload["message"]} + return {"sha": HEAD} + if method == "POST" and route == "git/refs": + branch = payload["ref"].removeprefix("refs/heads/") + assert branch not in self.refs, "Never repeat a create-ref write" + self.refs[branch] = payload["sha"] + response = {"ref": payload["ref"], "object": {"sha": payload["sha"]}} + elif method == "POST" and route == "pulls": + assert not self.prs, "Never duplicate a PR creation" + assert payload["draft"] is True and payload["base"] == "main" + repo = {"full_name": publisher.REPOSITORY, "id": int(publisher.REPOSITORY_ID)} + response = { + "number": 42, "html_url": f"https://github.com/{publisher.REPOSITORY}/pull/42", + "state": "open", "draft": True, "merged_at": None, "body": payload["body"], + "user": {"login": APP_SLUG + "[bot]", "type": "Bot"}, + "head": {"ref": payload["head"], "sha": self.refs[payload["head"]], "repo": repo}, + "base": {"ref": "main", "repo": repo}, + } + self.prs.append(response) + else: + raise AssertionError(f"Unexpected fake request: {method} {path}") + if self.uncertain_write == path: + raise publisher.PublicationError(self.uncertain_code) + return copy.deepcopy(response) + + +def publish(package, api, result=None): + return publisher.publish_candidate( + api, package.raw, package.receipt, package.context, ARTIFACT_ID, ARTIFACT_DIGEST, + APP_SLUG, INSTALLATION_ID, result if result is not None else publisher.initial_result(), + root=package.root, now=NOW, + ) + + +def test_full_dormant_prepare_validate_and_draft_pr_flow(package): + raw, receipt = validate(package) + api = FakeGitHub(raw) + result = publish(package, api) + assert result["status"] == "draft_pr_created" + assert result["published"] is False and result["review_required"] is True + assert result["branch_created"] is True and result["pr_created"] is True + assert result["pr_url"] == f"https://github.com/{publisher.REPOSITORY}/pull/42" + body = api.prs[0]["body"] + for binding in (BASE, "12345", ARTIFACT_ID, ARTIFACT_DIGEST, publisher.sha256(raw), publisher.sha256(publisher.json_bytes(receipt))): + assert binding in body + assert "published=false" in body + assert [path for method, path, _ in api.calls if method == "POST"] == [ + publisher.endpoint("git/" + name) for name in ("blobs", "trees", "commits", "refs") + ] + [publisher.endpoint("pulls")] + + +def test_repeat_identical_package_reuses_draft_pr_without_any_write(package): + api = FakeGitHub(package.raw) + first = publish(package, api) + count = len(api.calls) + second = publish(package, api) + assert second["status"] == "draft_pr_reused" + assert second["pr_url"] == first["pr_url"] and second["branch"] == first["branch"] + assert second["published"] is False + assert all(method == "GET" for method, _, _ in api.calls[count:]) + + +def test_candidate_already_on_main_is_noop_and_does_not_claim_publication(package): + api = FakeGitHub(package.raw) + api.trees[BASE_TREE][0]["sha"] = publisher.git_blob_sha(package.raw) + result = publish(package, api) + assert result["status"] == "already_on_base" and result["main_contains_candidate"] is True + assert result["published"] is False + assert all(method == "GET" for method, _, _ in api.calls) + + +@pytest.mark.parametrize("when", [1, 2, 3]) +def test_moved_base_stops_before_branch_or_pr_write(package, when): + api = FakeGitHub(package.raw) + api.move_base_on_read = when + result = publisher.initial_result() + with pytest.raises(publisher.PublicationError, match="BASE_MOVED"): + publish(package, api, result) + assert not api.prs and result["published"] is False + if when <= 2: + assert not api.refs + + +def test_base_moving_during_pr_creation_retains_pr_evidence_without_claiming_success(package): + api = FakeGitHub(package.raw) + api.move_base_on_read = 4 + result = publisher.initial_result() + with pytest.raises(publisher.PublicationError, match="BASE_MOVED"): + publish(package, api, result) + assert len(api.prs) == 1 and result["pr_created"] is True + assert result["pr_url"] == api.prs[0]["html_url"] + assert result["published"] is False + + +@pytest.mark.parametrize("binding", ["raw", "receipt", "context", "config"]) +def test_direct_publish_revalidates_local_bindings_before_contacting_github(package, binding): + api = FakeGitHub(package.raw) + if binding == "raw": + package.raw += b"\n" + elif binding == "receipt": + package.receipt["source_sha"] = HEAD + elif binding == "context": + package.context["run_attempt"] = "99" + else: + path = package.root / publisher.CONFIG_PATHS[0] + path.write_bytes(path.read_bytes() + b"\n") + with pytest.raises(publisher.PublicationError): + publish(package, api) + assert not api.calls + + +def test_publish_rejects_mismatching_explicit_receipt_digest_before_api_calls(package): + api = FakeGitHub(package.raw) + with pytest.raises(publisher.PublicationError, match="PACKAGE_HASH_MISMATCH"): + publisher.publish_candidate( + api, package.raw, package.receipt, package.context, ARTIFACT_ID, ARTIFACT_DIGEST, + APP_SLUG, INSTALLATION_ID, publisher.initial_result(), root=package.root, now=NOW, + receipt_sha256="0" * 64, + ) + assert not api.calls + + +@pytest.mark.parametrize("edit", ["parent", "message", "other_file", "candidate_file", "extra_file"]) +def test_existing_branch_human_edits_are_never_overwritten(package, edit): + api = FakeGitHub(package.raw) + publish(package, api) + if edit == "parent": + api.commits[HEAD]["parents"] = [{"sha": "9" * 40}] + elif edit == "message": + api.commits[HEAD]["message"] = "Human rewrite with identical file bytes" + elif edit == "extra_file": + api.trees[NEW_TREE].append({"path": "human.txt", "type": "blob", "mode": "100644", "sha": "9" * 40}) + else: + path = "README.md" if edit == "other_file" else publisher.DATA_PATH + next(entry for entry in api.trees[NEW_TREE] if entry["path"] == path)["sha"] = "9" * 40 + count = len(api.calls) + with pytest.raises(publisher.PublicationError, match="BRANCH_MODIFIED"): + publish(package, api) + assert all(method == "GET" for method, _, _ in api.calls[count:]) + + +def test_same_tree_manual_head_replacement_is_not_reused(package): + api = FakeGitHub(package.raw) + first = publish(package, api) + replacement = "9" * 40 + api.commits[replacement] = {**copy.deepcopy(api.commits[HEAD]), "sha": replacement} + api.refs[first["branch"]] = replacement + api.prs[0]["head"]["sha"] = replacement + count = len(api.calls) + with pytest.raises(publisher.PublicationError, match="PR_IDENTITY_MISMATCH"): + publish(package, api) + assert all(method == "GET" for method, _, _ in api.calls[count:]) + + +@pytest.mark.parametrize("path,value,code", [ + (("state",), "closed", "PR_HUMAN_STATE_CHANGED"), + (("draft",), False, "PR_HUMAN_STATE_CHANGED"), + (("merged_at",), "2026-10-06T12:00:00Z", "PR_HUMAN_STATE_CHANGED"), + (("user", "login"), "human", "PR_IDENTITY_MISMATCH"), + (("user", "type"), "User", "PR_IDENTITY_MISMATCH"), + (("head", "sha"), "9" * 40, "PR_IDENTITY_MISMATCH"), + (("head", "repo", "full_name"), "attacker/fork", "PR_IDENTITY_MISMATCH"), + (("base", "ref"), "other", "PR_IDENTITY_MISMATCH"), + (("body",), "human replacement", "PR_IDENTITY_MISMATCH"), + (("html_url",), "https://attacker.invalid/pull/42", "PR_IDENTITY_MISMATCH"), +]) +def test_existing_pr_human_state_and_identity_changes_are_respected(package, path, value, code): + api = FakeGitHub(package.raw) + publish(package, api) + change(api.prs[0], path, value) + count = len(api.calls) + with pytest.raises(publisher.PublicationError, match=code): + publish(package, api) + assert all(method == "GET" for method, _, _ in api.calls[count:]) + + +def test_multiple_prs_fail_closed(package): + api = FakeGitHub(package.raw) + publish(package, api) + api.prs.append(copy.deepcopy(api.prs[0])) + with pytest.raises(publisher.PublicationError, match="PR_IDENTITY_AMBIGUOUS"): + publish(package, api) + + +@pytest.mark.parametrize("route", ["git/refs", "pulls"]) +def test_uncertain_create_response_is_reconciled_without_duplicate_post(package, route): + api = FakeGitHub(package.raw) + api.uncertain_write = publisher.endpoint(route) + result = publish(package, api) + assert result["status"] == "draft_pr_created" and result["published"] is False + assert len(api.refs) == len(api.prs) == 1 + assert sum(method == "POST" and path == publisher.endpoint(route) for method, path, _ in api.calls) == 1 + writes = [index for index, (method, path, _) in enumerate(api.calls) if method == "POST" and path == publisher.endpoint(route)] + assert api.calls[writes[0] + 1][0] == "GET" + + +def test_create_ref_race_with_different_same_tree_head_is_not_adopted(package, monkeypatch): + api = FakeGitHub(package.raw) + request = api.request + + def race(method, path, payload=None, **kwargs): + response = request(method, path, payload, **kwargs) + if method == "POST" and path == publisher.endpoint("git/refs"): + replacement = "9" * 40 + api.commits[replacement] = {**copy.deepcopy(api.commits[HEAD]), "sha": replacement} + api.refs[payload["ref"].removeprefix("refs/heads/")] = replacement + raise publisher.PublicationError("API_ALREADY_EXISTS") + return response + + monkeypatch.setattr(api, "request", race) + result = publisher.initial_result() + with pytest.raises(publisher.PublicationError, match="BRANCH_MODIFIED"): + publish(package, api, result) + assert not api.prs and result["published"] is False + assert sum(method == "POST" and path == publisher.endpoint("git/refs") for method, path, _ in api.calls) == 1 + + +@pytest.mark.parametrize("route,code", [ + ("git/blobs", "API_HTTP_403"), ("git/trees", "API_HTTP_403"), ("git/commits", "API_HTTP_403"), + ("git/refs", "API_HTTP_403"), ("pulls", "API_HTTP_403"), +]) +def test_write_refusal_never_claims_publication_or_retries(package, route, code): + api = FakeGitHub(package.raw) + api.fail_write = publisher.endpoint(route) + result = publisher.initial_result() + with pytest.raises(publisher.PublicationError, match=code): + publish(package, api, result) + assert result["published"] is False and not api.prs + assert sum(method == "POST" and path == publisher.endpoint(route) for method, path, _ in api.calls) == 1 + + +@pytest.mark.parametrize("route", ["git/refs", "pulls"]) +@pytest.mark.parametrize("code", ["API_HTTP_401", "API_HTTP_403", "API_HTTP_404", "API_HTTP_422", "API_HTTP_429"]) +def test_definite_create_rejections_propagate_exact_cause_without_reconcile(package, route, code): + api = FakeGitHub(package.raw) + api.fail_write = publisher.endpoint(route) + api.fail_code = code + result = publisher.initial_result() + with pytest.raises(publisher.PublicationError) as raised: + publish(package, api, result) + assert str(raised.value) == code + assert result["published"] is False + writes = [index for index, (method, path, _) in enumerate(api.calls) if method == "POST" and path == publisher.endpoint(route)] + assert len(writes) == 1 + assert api.calls[writes[0] + 1:] == [], "Definite refusal must not trigger reconciliation or another API read" + + +@pytest.mark.parametrize("route", ["git/refs", "pulls"]) +@pytest.mark.parametrize("code", ["API_TRANSPORT_FAILED", "API_HTTP_500", "API_HTTP_503", "API_ALREADY_EXISTS"]) +@pytest.mark.parametrize("committed", [False, True]) +def test_only_uncertain_or_classified_existing_create_has_one_read_reconcile(package, route, code, committed): + api = FakeGitHub(package.raw) + if committed: + api.uncertain_write = publisher.endpoint(route) + api.uncertain_code = code + else: + api.fail_write = publisher.endpoint(route) + api.fail_code = code + result = publisher.initial_result() + if committed: + assert publish(package, api, result)["status"] == "draft_pr_created" + else: + expected = "BRANCH_CREATION_UNCONFIRMED" if route == "git/refs" else "PR_CREATION_UNCONFIRMED" + with pytest.raises(publisher.PublicationError, match=f"^{expected}$"): + publish(package, api, result) + assert result["published"] is False + writes = [index for index, (method, path, _) in enumerate(api.calls) if method == "POST" and path == publisher.endpoint(route)] + assert len(writes) == 1 + after = api.calls[writes[0] + 1:] + assert after and after[0][0] == "GET" + reconciliation_prefix = publisher.endpoint("git/ref/heads/automation/theme-") if route == "git/refs" else publisher.endpoint("pulls?") + assert sum(method == "GET" and path.startswith(reconciliation_prefix) for method, path, _ in after) == 1 + if not committed: + assert len(after) == 1 + + +@pytest.mark.parametrize("field,value,code", [("state", "closed", "PR_HUMAN_STATE_CHANGED"), ("body", "wrong identity", "PR_IDENTITY_MISMATCH")]) +def test_local_create_pr_response_validation_failure_is_not_reconciled(package, monkeypatch, field, value, code): + api = FakeGitHub(package.raw) + request = api.request + + def invalid_response(method, path, payload=None, **kwargs): + response = request(method, path, payload, **kwargs) + if method == "POST" and path == publisher.endpoint("pulls"): + response[field] = value + return response + + monkeypatch.setattr(api, "request", invalid_response) + result = publisher.initial_result() + with pytest.raises(publisher.PublicationError, match=f"^{code}$"): + publish(package, api, result) + assert result["published"] is False + assert api.calls[-1][0:2] == ("POST", publisher.endpoint("pulls")) + assert sum(method == "POST" and path == publisher.endpoint("pulls") for method, path, _ in api.calls) == 1 + + +@pytest.mark.parametrize("scope", [ + {"total_count": 2, "repositories": []}, + {"total_count": True, "repositories": []}, + {"total_count": 1, "repositories": [{"full_name": "attacker/fork", "id": int(publisher.REPOSITORY_ID)}]}, + {"total_count": 1, "repositories": [{"full_name": publisher.REPOSITORY, "id": 1}]}, +]) +def test_wrong_app_scope_is_rejected_before_any_write(package, scope): + api = FakeGitHub(package.raw) + api.scope = scope + with pytest.raises(publisher.PublicationError, match="APP_SCOPE_MISMATCH"): + publish(package, api) + assert all(method == "GET" for method, _, _ in api.calls) + + +def mock_http(monkeypatch, *, raw=b"{}", headers=None, error=None): + requests = [] + + class Response(io.BytesIO): + def __init__(self): + super().__init__(raw) + self.headers = headers or {} + + class Opener: + def open(self, request, timeout): + requests.append(request) + assert timeout == 20 + if error: + raise error + return Response() + + def build_opener(handler): + assert isinstance(handler, publisher.NoRedirect) + return Opener() + + monkeypatch.setattr(publisher.urllib.request, "build_opener", build_opener) + return requests + + +def create_request_payload(route): + branch = "automation/theme-12345-2-0123456789abcdef" + if route == "git/refs": + return {"ref": "refs/heads/" + branch, "sha": HEAD} + return {"head": branch, "base": "main", "draft": True, "title": "Synthetic", "body": "Synthetic"} + + +@pytest.mark.parametrize("route,body", [ + ("git/refs", {"message": "Reference already exists"}), + ("pulls", {"message": "Validation Failed", "errors": [{ + "resource": "PullRequest", "code": "custom", + "message": "A pull request already exists for QuantStrategyLab:automation/theme-12345-2-0123456789abcdef.", + }]}), +]) +def test_github_422_classifies_only_exact_requested_existing_reference_or_pr(monkeypatch, route, body): + error = urllib.error.HTTPError("https://api.github.com", 422, "Unprocessable Entity", {}, io.BytesIO(publisher.json_bytes(body))) + requests = mock_http(monkeypatch, error=error) + with pytest.raises(publisher.PublicationError) as raised: + publisher.GitHub(TOKEN).request("POST", publisher.endpoint(route), create_request_payload(route)) + assert str(raised.value) == "API_ALREADY_EXISTS" + assert len(requests) == 1 + + +@pytest.mark.parametrize("route,raw", [ + ("git/refs", b'{"message":"Validation Failed"}'), + ("git/refs", b'{"message":"Reference already exists for a different target"}'), + ("git/refs", b'{"message":"Reference already exists","message":"other"}'), + ("git/refs", b'not json'), + ("pulls", b'{"message":"Reference already exists"}'), + ("pulls", b'{"message":"Validation Failed","errors":[{"resource":"PullRequest","code":"custom","message":"A pull request already exists for QuantStrategyLab:other-branch."}]}'), + ("pulls", b'{"message":"Validation Failed","errors":[{"resource":"Issue","code":"custom","message":"A pull request already exists for QuantStrategyLab:automation/theme-12345-2-0123456789abcdef."}]}'), +]) +def test_github_422_unknown_or_wrong_target_response_remains_definite_rejection(monkeypatch, route, raw): + error = urllib.error.HTTPError("https://api.github.com", 422, "Unprocessable Entity", {}, io.BytesIO(raw)) + requests = mock_http(monkeypatch, error=error) + with pytest.raises(publisher.PublicationError, match="^API_HTTP_422$"): + publisher.GitHub(TOKEN).request("POST", publisher.endpoint(route), create_request_payload(route)) + assert len(requests) == 1 + + +def test_github_422_classifier_reads_a_bounded_body_and_rejects_oversized_response(monkeypatch): + reads = [] + + class BoundedBody(io.BytesIO): + def read(self, size=-1): + assert 0 < size <= 1024 * 1024 + 1, "Error bodies must have a finite read bound" + reads.append(size) + return super().read(size) + + raw = b'{"message":"Reference already exists"}' + b" " * (1024 * 1024 + 1) + error = urllib.error.HTTPError("https://api.github.com", 422, "Unprocessable Entity", {}, BoundedBody(raw)) + mock_http(monkeypatch, error=error) + with pytest.raises(publisher.PublicationError, match="^API_HTTP_422$"): + publisher.GitHub(TOKEN).request("POST", publisher.endpoint("git/refs"), create_request_payload("git/refs")) + assert len(reads) == 1 + + +@pytest.mark.parametrize("method,path,read_only", [ + ("PATCH", publisher.endpoint("git/refs/heads/main"), False), + ("DELETE", publisher.endpoint("git/refs/heads/main"), False), + ("POST", publisher.endpoint("pulls"), True), + ("GET", "/repos/attacker/other/pulls", False), + ("GET", "https://attacker.invalid/", False), +]) +def test_github_client_rejects_mutation_methods_readonly_writes_and_other_targets(method, path, read_only): + with pytest.raises(publisher.PublicationError, match="API_METHOD_FORBIDDEN|API_TARGET_FORBIDDEN"): + publisher.GitHub(TOKEN, read_only=read_only).request(method, path) + + +@pytest.mark.parametrize("route,payload", [ + ("issues", {"title": "unrelated"}), + ("git/refs", {"ref": "refs/heads/main", "sha": HEAD}), + ("git/refs", {"ref": "refs/heads/other", "sha": HEAD}), + ("pulls", {"head": "automation/theme-12345-2-0123456789abcdef", "base": "main", "draft": False}), + ("pulls", {"head": "automation/theme-12345-2-0123456789abcdef", "base": "other", "draft": True}), + ("git/trees", {"base_tree": BASE_TREE, "tree": [{"path": "README.md", "mode": "100644", "type": "blob", "sha": HEAD}]}), + ("git/trees", {"base_tree": BASE_TREE, "tree": []}), +]) +def test_github_post_allowlist_blocks_other_endpoints_main_refs_and_nondraft_prs(route, payload): + with pytest.raises(publisher.PublicationError): + publisher.GitHub(TOKEN).request("POST", publisher.endpoint(route), payload) + + +def test_github_client_uses_fixed_host_and_no_redirect_handler(monkeypatch): + requests = mock_http(monkeypatch, raw=b'{"ok": true}') + assert publisher.GitHub(TOKEN, read_only=True).request("GET", publisher.endpoint("git/ref/heads/main")) == {"ok": True} + assert len(requests) == 1 + assert requests[0].full_url == "https://api.github.com" + publisher.endpoint("git/ref/heads/main") + assert requests[0].get_header("Authorization") == "Bearer " + TOKEN + assert requests[0].get_header("X-github-api-version") == "2022-11-28" + assert publisher.NoRedirect().redirect_request(None, None, 302, "redirect", {}, "https://attacker.invalid") is None + + +@pytest.mark.parametrize("error,code", [ + (urllib.error.HTTPError("https://api.github.com/" + TOKEN, 302, TOKEN, {}, None), "API_HTTP_302"), + (urllib.error.HTTPError("https://api.github.com/" + TOKEN, 403, TOKEN, {}, None), "API_HTTP_403"), + (urllib.error.URLError(TOKEN), "API_TRANSPORT_FAILED"), + (TimeoutError(TOKEN), "API_TRANSPORT_FAILED"), +]) +def test_github_failures_do_not_retry_redirect_or_expose_secrets(monkeypatch, error, code): + requests = mock_http(monkeypatch, error=error) + with pytest.raises(publisher.PublicationError) as raised: + publisher.GitHub(TOKEN).request("GET", publisher.endpoint("pulls")) + assert str(raised.value) == code + assert TOKEN not in str(raised.value) and raised.value.__suppress_context__ is True + assert len(requests) == 1 + + +def test_unknown_error_text_is_never_a_public_reason_code(): + error = publisher.PublicationError(TOKEN) + assert error.code == str(error) == "PUBLICATION_FAILED" + + +def test_github_missing_ref_404_is_optional_only_when_explicit(monkeypatch): + requests = mock_http(monkeypatch, error=urllib.error.HTTPError("https://api.github.com", 404, "missing", {}, None)) + api = publisher.GitHub(TOKEN) + assert api.request("GET", publisher.endpoint("git/ref/heads/example"), missing_ok=True) is None + with pytest.raises(publisher.PublicationError, match="API_HTTP_404"): + api.request("GET", publisher.endpoint("git/ref/heads/example")) + assert len(requests) == 2 + + +@pytest.mark.parametrize("raw,headers,code", [ + (b"{}", {"Link": "; rel=next"}, "API_PAGINATION_UNEXPECTED"), + (b" " * (1024 * 1024 + 1), {}, "API_RESPONSE_OVERSIZED"), + (b'{"duplicate":1,"duplicate":2}', {}, "JSON_DUPLICATE_KEY"), +]) +def test_github_rejects_unbounded_or_ambiguous_responses(monkeypatch, raw, headers, code): + mock_http(monkeypatch, raw=raw, headers=headers) + with pytest.raises(publisher.PublicationError, match=code): + publisher.GitHub(TOKEN).request("GET", publisher.endpoint("pulls"), single_page=True) + + +def test_validate_cli_uses_readonly_token_and_only_reads_main(package, monkeypatch, capsys): + requests = mock_http(monkeypatch, raw=publisher.json_bytes({"object": {"sha": BASE}})) + for key, value in {**trusted_env(), "GITHUB_TOKEN": TOKEN}.items(): + monkeypatch.setenv(key, value) + monkeypatch.setattr(publisher, "verify_checkout", lambda _: None) + original_validate = publisher.validate_package + monkeypatch.setattr(publisher, "validate_package", lambda *args: original_validate(*args, root=package.root, now=NOW)) + original_api = publisher.GitHub + clients = [] + + def client(token, *, read_only): + assert token == TOKEN and read_only is True + clients.append(original_api(token, read_only=read_only)) + return clients[-1] + + monkeypatch.setattr(publisher, "GitHub", client) + arguments = ["validate", "--package-dir", str(package.directory), "--artifact-id", ARTIFACT_ID, "--artifact-digest", ARTIFACT_DIGEST] + for key, value in package.hashes.items(): + arguments.extend(["--" + key.replace("_", "-"), value]) + assert publisher.main(arguments) == 0 + assert len(clients) == len(requests) == 1 + assert requests[0].get_method() == "GET" + assert requests[0].full_url.endswith("/git/ref/heads/main") + result = json.loads(capsys.readouterr().out) + assert result["status"] == "package_validated" and result["published"] is False + + +@pytest.mark.parametrize("refuse", [False, True]) +def test_publish_cli_records_verified_draft_or_failure_without_secrets(package, monkeypatch, capsys, tmp_path, refuse): + api = FakeGitHub(package.raw) + if refuse: + api.fail_write = publisher.endpoint("pulls") + for key, value in { + **trusted_env(), "RESEARCH_PUBLISHER_TOKEN": TOKEN, + "RESEARCH_PUBLISHER_APP_SLUG": APP_SLUG, "RESEARCH_PUBLISHER_INSTALLATION_ID": INSTALLATION_ID, + }.items(): + monkeypatch.setenv(key, value) + checkouts = [] + monkeypatch.setattr(publisher, "verify_checkout", lambda context: checkouts.append(context)) + original_validate = publisher.validate_package + monkeypatch.setattr(publisher, "validate_package", lambda *args: original_validate(*args, root=package.root, now=NOW)) + original_publish = publisher.publish_candidate + monkeypatch.setattr(publisher, "publish_candidate", lambda *args, **kwargs: original_publish(*args, root=package.root, now=NOW, **kwargs)) + monkeypatch.setattr(publisher, "GitHub", lambda token: api if token == TOKEN else pytest.fail("wrong token")) + result_path = tmp_path / "result.json" + arguments = ["publish", "--package-dir", str(package.directory), "--artifact-id", ARTIFACT_ID, + "--artifact-digest", ARTIFACT_DIGEST, "--result", str(result_path)] + for key, value in package.hashes.items(): + arguments.extend(["--" + key.replace("_", "-"), value]) + assert publisher.main(arguments) == int(refuse) + assert checkouts == [package.context] + result = json.loads(result_path.read_bytes()) + assert result["published"] is False + assert result["status"] == ("failed" if refuse else "draft_pr_created") + assert result["reason_code"] == ("API_HTTP_403" if refuse else None) + assert TOKEN not in result_path.read_text() + capsys.readouterr().out diff --git a/tests/test_theme_momentum_snapshot_workflow.py b/tests/test_theme_momentum_snapshot_workflow.py index 220f732..5a94bed 100644 --- a/tests/test_theme_momentum_snapshot_workflow.py +++ b/tests/test_theme_momentum_snapshot_workflow.py @@ -4,6 +4,10 @@ import copy import io import re +import os +import subprocess +import sys +from datetime import datetime import textwrap import urllib.request @@ -11,16 +15,6 @@ import pytest -def test_theme_momentum_snapshot_workflow_tolerates_repo_rule_push_rejection() -> None: - workflow = Path(".github/workflows/theme_momentum_snapshot.yml").read_text(encoding="utf-8") - - assert "Push blocked by repository rules; keeping artifact-only output." in workflow - assert "GH013: Repository rule violations found for refs/heads/main" in workflow - assert "grep -Eqi 'required status check' \"${push_log}\"" in workflow - assert "GH013|repository rule violations|required status check" not in workflow - assert "git push 2>\"${push_log}\"" in workflow - - IDENTITY_REPOSITORY = "QuantStrategyLab/ResearchSignalContextPipelines" IDENTITY_WORKFLOW = "theme_momentum_snapshot.yml" @@ -36,8 +30,20 @@ def identity_workflow(): return Path(__file__).parents[1] / ".github" / "workflows" / IDENTITY_WORKFLOW +def workflow_job(job_id): + jobs = identity_workflow().read_text().split("\njobs:\n", 1)[1] + return next( + " " + block for block in re.split(r"(?m)^ (?=[a-z][a-z-]+:\n)", jobs) + if block.startswith(job_id + ":\n") + ) + + +def workflow_steps(job_id): + return [" - " + block for block in re.split(r"(?m)^ - ", workflow_job(job_id))[1:]] + + def identity_steps(): - return [" - " + block for block in re.split(r"(?m)^ - ", identity_workflow().read_text())[1:]] + return workflow_steps("build-theme-momentum") def identity_step(step_id): @@ -59,6 +65,7 @@ def identity_env(monkeypatch, tmp_path): "GITHUB_REPOSITORY_ID": "123456", "GITHUB_WORKFLOW_REF": f"{IDENTITY_REPOSITORY}/.github/workflows/{IDENTITY_WORKFLOW}@refs/heads/main", "RESEARCH_PUBLISHER_APP_ID": "5214420", + "PUBLISH_DRAFT_PR": "false", "RESEARCH_PUBLISHER_INSTALLATION_ID": "168603414", "RESEARCH_PUBLISHER_TOKEN": "synthetic-test-token-never-real", "GITHUB_OUTPUT": str(tmp_path / "outputs"), @@ -117,7 +124,10 @@ def test_identity_only_is_opt_in_and_isolates_all_normal_steps(): assert len(identity) == 4 for step in steps: if step not in identity: - assert "if: ${{ !inputs.identity_only }}" in step or "if: ${{ always() && !inputs.identity_only }}" in step + guard = next(line for line in step.splitlines() if line.strip().startswith("if:")) + assert "!inputs.identity_only" in guard + if "id: snapshot_upload" not in step: + assert "||" not in guard for step in identity: assert "github.event_name == 'workflow_dispatch' && inputs.identity_only" in step assert "always()" in identity_step("publisher_identity_summary") @@ -148,6 +158,7 @@ def test_identity_preflight_succeeds_without_network(monkeypatch, tmp_path): ("GITHUB_REPOSITORY_ID", "not-an-id"), ("GITHUB_WORKFLOW_REF", "other/other/.github/workflows/test.yml@refs/heads/main"), ("RESEARCH_PUBLISHER_APP_ID", "123"), + ("PUBLISH_DRAFT_PR", "true"), ]) def test_identity_preflight_fails_closed(monkeypatch, tmp_path, capsys, key, value): identity_env(monkeypatch, tmp_path) @@ -230,3 +241,433 @@ def test_identity_summary_defaults_to_unpublished_and_unverified(monkeypatch, tm summary = json.loads(capsys.readouterr().out) assert summary == {"app_id_matches": False, "installation_matches": False, "single_repo_scope_matches": False, "identity_verified": False, "published": False} assert "synthetic-test-token" not in Path(values["GITHUB_STEP_SUMMARY"]).read_text() + + +def publication_step(step_id): + return next( + block for block in workflow_steps("publish-theme-draft-pr") + if f" id: {step_id}\n" in block + ) + + +def run_inline_python(block): + script = textwrap.dedent(block.split(" run: |\n", 1)[1]) + assert script.startswith("python - <<'PY'\n") + source = script.removeprefix("python - <<'PY'\n").rsplit("\nPY", 1)[0] + exec(compile(source, "", "exec"), {"__name__": "__main__"}) + + +def test_workflow_never_commits_outputs_or_grants_default_write_permissions(): + text = identity_workflow().read_text() + assert "permissions:\n contents: read\n" in text + assert not re.search(r"(?m)^\s+contents: write$", text) + for removed in ("git push", "git commit", "git add", "GH013", "continue-on-error", "[skip ci]"): + assert removed not in text + assert 'description: "Deprecated, ignored:' in text + assert " commit_outputs:\n" in text + assert " publish_draft_pr:\n" in text + publication_input = text.split(" publish_draft_pr:\n", 1)[1].split(" as_of:\n", 1)[0] + assert " default: false\n type: boolean\n" in publication_input + schedule = text.split(" schedule:\n", 1)[1].split("\npermissions:\n", 1)[0] + assert "publish_draft_pr" not in schedule + assert "artifact-only" in schedule + for step in identity_steps(): + if "commit_outputs" in step: + assert "id: generation_summary" in step + + +def test_all_workflow_action_dependencies_are_immutable(): + actions = re.findall(r"(?m)^\s+uses: (\S+)", identity_workflow().read_text()) + assert actions + for action in actions: + assert re.fullmatch(r"actions/[a-z-]+@[a-f0-9]{40}", action) + assert "skip-token-revoke" not in identity_workflow().read_text() + + +def test_content_admission_is_immediately_after_checkout_and_before_collection(): + steps = identity_steps() + checkout = steps.index(identity_step("generation_checkout")) + admission = steps.index(identity_step("publisher_admission")) + assert admission == checkout + 1 + assert "python scripts/publish_theme_snapshot_pr.py admission" in steps[admission] + assert "!inputs.identity_only && inputs.publish_draft_pr" in steps[admission] + assert "github.sha" in steps[checkout] + assert "persist-credentials: false" in steps[checkout] + for step in steps[admission + 1:]: + if any(label in step for label in ("setup-python", "name: Install package", "id: generation\n", "id: publisher_package\n")): + assert "always()" not in step + assert "continue-on-error" not in step + assert steps.index(identity_step("generation")) > admission + preflight = identity_step("publisher_identity_preflight") + assert "PUBLISH_DRAFT_PR: ${{ inputs.publish_draft_pr }}" in preflight + assert "success()" in identity_step("publisher_identity_token") + assert "success()" in identity_step("publisher_identity_verify") + + +def test_same_run_package_has_only_candidate_and_receipt_and_bound_outputs(): + build = workflow_job("build-theme-momentum") + package = identity_step("publisher_package") + upload = identity_step("publisher_package_upload") + assert '--candidate data/output/theme_momentum_snapshot.json' in package + assert 'GENERATION_STARTED_AT: ${{ steps.generation.outputs.started_at }}' in package + assert 'GENERATION_COMPLETED_AT: ${{ steps.generation.outputs.completed_at }}' in package + assert 'PACKAGE_DIR: ${{ runner.temp }}/theme-publication-package' in package + assert 'if: ${{ !inputs.identity_only && inputs.publish_draft_pr }}' in package + assert 'if: ${{ !inputs.identity_only && inputs.publish_draft_pr }}' in upload + paths = upload.split(' path: |\n', 1)[1].split(' if-no-files-found:', 1)[0] + assert paths.splitlines() == [ + ' ${{ runner.temp }}/theme-publication-package/theme_momentum_snapshot.json', + ' ${{ runner.temp }}/theme-publication-package/publication-package.json', + ] + assert '${{ github.run_id }}-${{ github.run_attempt }}' in upload + for output, source in ( + ('artifact_id', 'steps.publisher_package_upload.outputs.artifact-id'), + ('artifact_digest', 'steps.publisher_package_upload.outputs.artifact-digest'), + ('candidate_sha256', 'steps.publisher_package.outputs.candidate_sha256'), + ('receipt_sha256', 'steps.publisher_package.outputs.receipt_sha256'), + ): + assert f' {output}: ${{{{ {source} }}}}' in build + assert '!inputs.publish_draft_pr' in identity_step('snapshot_upload') + + +def test_isolated_publication_requires_trusted_manual_main_request(): + job = workflow_job("publish-theme-draft-pr") + guard = job.split(" if: >-\n", 1)[1].split(" runs-on:", 1)[0] + for required in ( + "github.event_name == 'workflow_dispatch'", + f"github.repository == '{IDENTITY_REPOSITORY}'", + "github.ref == 'refs/heads/main'", + f"github.workflow_ref == '{IDENTITY_REPOSITORY}/.github/workflows/{IDENTITY_WORKFLOW}@refs/heads/main'", + "inputs.publish_draft_pr && !inputs.identity_only", + ): + assert required in guard + assert "||" not in guard + assert "needs: build-theme-momentum" in job + assert "permissions:\n contents: read" in job + assert "actions: read" not in job + checkout = publication_step("publisher_checkout") + assert "ref: ${{ github.sha }}" in checkout + assert "persist-credentials: false" in checkout + download = publication_step("publisher_download") + assert "artifact-ids: ${{ needs.build-theme-momentum.outputs.artifact_id }}" in download + assert "path: ${{ runner.temp }}/theme-publication-package" in download + assert "digest-mismatch: error" in download + for disallowed in ("github-token:", "repository:", "run-id:", "name:"): + assert disallowed not in download.split(" with:\n", 1)[1] + + +def test_package_and_main_validation_precedes_isolated_single_repo_mint(): + steps = workflow_steps("publish-theme-draft-pr") + validate = publication_step("publisher_validate") + mint = publication_step("draft_publisher_token") + publish = publication_step("draft_publish") + assert steps.index(validate) < steps.index(mint) < steps.index(publish) + assert "GITHUB_TOKEN: ${{ github.token }}" in validate + assert "scripts/publish_theme_snapshot_pr.py validate" in validate + assert IDENTITY_ACTION in mint + assert 'app-id: "5214420"' in mint + assert " repositories: ResearchSignalContextPipelines\n" in mint + assert " owner: QuantStrategyLab\n" in mint + assert " permission-contents: write\n" in mint + assert " permission-pull-requests: write\n" in mint + assert mint.count(" permission-") == 2 + assert "scripts/publish_theme_snapshot_pr.py publish" in publish + assert "GITHUB_TOKEN:" not in publish + assert "RESEARCH_PUBLISHER_TOKEN: ${{ steps.draft_publisher_token.outputs.token }}" in publish + assert "RESEARCH_PUBLISHER_INSTALLATION_ID: ${{ steps.draft_publisher_token.outputs.installation-id }}" in publish + assert "RESEARCH_PUBLISHER_APP_SLUG: ${{ steps.draft_publisher_token.outputs.app-slug }}" in publish + for step in steps: + assert "pip install" not in step + assert "setup-python" not in step + if step != publish: + assert "steps.draft_publisher_token.outputs.token" not in step + if step != mint: + assert "secrets.RESEARCH_PUBLISHER_APP_PRIVATE_KEY" not in step + for step in (validate, mint, publish): + assert "always()" not in step + assert "continue-on-error" not in step + assert "always()" in publication_step("draft_publisher_summary") + assert "if: ${{ always() }}" in steps[-1] + assert "publication-result.json" in steps[-1] + + +@pytest.mark.parametrize(("requested", "generation", "admission", "package", "upload", "snapshot", "expected"), [ + ("", "success", "skipped", "skipped", "skipped", "success", "artifact_only"), + ("false", "success", "skipped", "skipped", "skipped", "success", "artifact_only"), + ("true", "skipped", "failure", "skipped", "skipped", "skipped", "admission_rejected"), + ("false", "failure", "skipped", "skipped", "skipped", "skipped", "generation_failed"), + ("true", "success", "success", "failure", "skipped", "skipped", "package_rejected"), + ("true", "success", "success", "success", "failure", "skipped", "artifact_upload_failed"), + ("false", "success", "skipped", "skipped", "skipped", "failure", "artifact_upload_failed"), + ("true", "success", "success", "success", "success", "skipped", "package_ready"), +]) +def test_generation_summary_distinguishes_artifact_mode_and_failures( + monkeypatch, tmp_path, capsys, requested, generation, admission, package, upload, snapshot, expected, +): + for name, value in { + "PUBLICATION_REQUESTED": requested, + "LEGACY_COMMIT_OUTPUTS": "true", + "GENERATION_OUTCOME": generation, + "ADMISSION_OUTCOME": admission, + "PACKAGE_OUTCOME": package, + "PACKAGE_UPLOAD_OUTCOME": upload, + "SNAPSHOT_UPLOAD_OUTCOME": snapshot, + "GITHUB_STEP_SUMMARY": str(tmp_path / "summary"), + }.items(): + monkeypatch.setenv(name, value) + run_inline_python(identity_step("generation_summary")) + result = json.loads(capsys.readouterr().out) + assert result == { + "generated": generation == "success", + "publication_requested": requested == "true", + "legacy_commit_outputs_ignored": True, + "status": expected, + "published": False, + } + + +@pytest.mark.parametrize(("outcome", "expected"), [ + ("CHECKOUT_OUTCOME", "checkout_failed"), + ("DOWNLOAD_OUTCOME", "download_failed"), + ("VALIDATE_OUTCOME", "validation_rejected"), + ("MINT_OUTCOME", "mint_failed"), +]) +def test_publisher_summary_keeps_failure_receipt_without_a_token(monkeypatch, tmp_path, capsys, outcome, expected): + for key in ("CHECKOUT_OUTCOME", "DOWNLOAD_OUTCOME", "VALIDATE_OUTCOME", "MINT_OUTCOME"): + monkeypatch.setenv(key, "failure" if key == outcome else "success") + result_path = tmp_path / "result" / "publication-result.json" + monkeypatch.setenv("RESULT_PATH", str(result_path)) + monkeypatch.setenv("GITHUB_STEP_SUMMARY", str(tmp_path / "summary")) + monkeypatch.delenv("RESEARCH_PUBLISHER_TOKEN", raising=False) + run_inline_python(publication_step("draft_publisher_summary")) + result = json.loads(capsys.readouterr().out) + assert result == {"published": False, "review_required": True, "status": expected} + assert json.loads(result_path.read_text()) == result + + +def test_publisher_summary_never_logs_untrusted_extra_fields(monkeypatch, tmp_path, capsys): + result_path = tmp_path / "publication-result.json" + result_path.write_text(json.dumps({ + "status": "draft_pr_created", + "published": True, + "review_required": False, + "branch_created": True, + "pr_created": True, + "pr_url": f"https://github.com/{IDENTITY_REPOSITORY}/pull/12", + "head_sha": "a" * 40, + "candidate_sha256": "b" * 64, + "reason_code": "secret-token-value", + "raw_response": "secret-token-value", + })) + for key in ("CHECKOUT_OUTCOME", "DOWNLOAD_OUTCOME", "VALIDATE_OUTCOME", "MINT_OUTCOME", "PUBLISH_OUTCOME"): + monkeypatch.setenv(key, "success") + monkeypatch.setenv("RESULT_PATH", str(result_path)) + monkeypatch.setenv("GITHUB_STEP_SUMMARY", str(tmp_path / "summary")) + run_inline_python(publication_step("draft_publisher_summary")) + logged = capsys.readouterr().out + assert "secret-token-value" not in logged + result_path.read_text() + (tmp_path / "summary").read_text() + result = json.loads(logged) + assert result["status"] == "draft_pr_created" + assert result["published"] is False + assert result["review_required"] is True + assert result["pr_created"] is True + assert result["pr_url"] == f"https://github.com/{IDENTITY_REPOSITORY}/pull/12" + assert result["head_sha"] == "a" * 40 + + +@pytest.mark.parametrize(("as_of", "prices", "strict", "extra"), [ + ("", "", "false", []), + ("2026-09-30", "prices path.csv", "true", ["--as-of", "2026-09-30", "--prices", "prices path.csv", "--strict-downloads"]), +]) +def test_generation_preserves_arguments_and_records_real_interval(tmp_path, as_of, prices, strict, extra): + # Replace only the Python executable with a local stand-in; never collect prices. + binary_dir = tmp_path / "bin" + binary_dir.mkdir() + python = binary_dir / "python" + python.write_text( + f"#!{sys.executable}\n" + "import json, os, sys\n" + "from pathlib import Path\n" + "Path(os.environ['CAPTURE']).write_text(json.dumps(sys.argv[1:]))\n" + "print('{}')\n" + ) + python.chmod(0o700) + block = identity_step("generation") + shell = textwrap.dedent(block.split(" run: |\n", 1)[1]) + # Keep all test output in the temporary directory, including the tee file. + shell = shell.replace("/tmp/theme_momentum_summary.json", str(tmp_path / "generated-summary.json")) + env = dict(os.environ) + env.update({ + "PATH": f"{binary_dir}:{env['PATH']}", + "CAPTURE": str(tmp_path / "arguments"), + "GITHUB_OUTPUT": str(tmp_path / "outputs"), + "GITHUB_STEP_SUMMARY": str(tmp_path / "summary"), + "INPUT_AS_OF": as_of, + "PRICES_PATH": prices, + "STRICT_DOWNLOADS": strict, + }) + before = datetime.now().astimezone() + run = subprocess.run(["bash", "-c", shell], cwd=tmp_path, env=env, capture_output=True, text=True) + after = datetime.now().astimezone() + assert run.returncode == 0, run.stderr + assert json.loads((tmp_path / "arguments").read_text()) == [ + "scripts/build_theme_momentum_snapshot.py", "--output", "data/output/theme_momentum_snapshot.json", *extra, + ] + outputs = dict(line.split("=", 1) for line in (tmp_path / "outputs").read_text().splitlines()) + start = datetime.fromisoformat(outputs["started_at"]) + finish = datetime.fromisoformat(outputs["completed_at"]) + assert before <= start <= finish <= after + assert start.tzinfo is not None and finish.tzinfo is not None + + +def test_publisher_summary_preserves_unknown_write_state_and_transport_binding(monkeypatch, tmp_path, capsys): + result_path = tmp_path / "publication-result.json" + result_path.write_text(json.dumps({ + "status": "failed", "reason_code": "API_HTTP_503", + "published": False, "review_required": True, + "branch_created": True, "pr_created": None, + "branch": "automation/theme-12-1-" + "f" * 16, + "package_key": "f" * 64, + })) + for key in ("CHECKOUT_OUTCOME", "DOWNLOAD_OUTCOME", "VALIDATE_OUTCOME", "MINT_OUTCOME"): + monkeypatch.setenv(key, "success") + bindings = {"artifact_id": "1234", "artifact_digest": "b" * 64, "candidate_sha256": "c" * 64, "receipt_sha256": "d" * 64} + for key, value in bindings.items(): + monkeypatch.setenv(key.upper(), value) + monkeypatch.setenv("PUBLISH_OUTCOME", "failure") + monkeypatch.setenv("RESULT_PATH", str(result_path)) + monkeypatch.setenv("GITHUB_STEP_SUMMARY", str(tmp_path / "summary")) + run_inline_python(publication_step("draft_publisher_summary")) + result = json.loads(capsys.readouterr().out) + assert result["reason_code"] == "API_HTTP_503" + assert result["branch_created"] is True + assert result["pr_created"] is None + assert result["published"] is False + assert result["package_key"] == "f" * 64 + for key, value in bindings.items(): + assert result[key] == value + assert json.loads(result_path.read_text()) == result + + +@pytest.mark.parametrize("identity_only", [False, True]) +@pytest.mark.parametrize("publish_draft_pr", [False, True]) +@pytest.mark.parametrize("generation_outcome", ["success", "failure", "skipped"]) +@pytest.mark.parametrize("package_upload_outcome", ["success", "failure", "skipped"]) +def test_snapshot_fallback_retains_only_a_successfully_generated_candidate( + identity_only, publish_draft_pr, generation_outcome, package_upload_outcome, +): + step = identity_step("snapshot_upload") + guard = next(line.strip() for line in step.splitlines() if line.strip().startswith("if:")) + expression = guard.removeprefix("if: ${{ ").removesuffix(" }}") + assert "always()" in expression + assert "continue-on-error" not in step + # Evaluate the checked-in guard with only synthetic outcome/input values. + expression = expression.replace("always()", "True") + expression = expression.replace("inputs.identity_only", "identity_only") + expression = expression.replace("inputs.publish_draft_pr", "publish_draft_pr") + expression = expression.replace("steps.generation.outcome", "generation_outcome") + expression = expression.replace("steps.publisher_package_upload.outcome", "package_upload_outcome") + expression = expression.replace("!identity_only", "not identity_only") + expression = expression.replace("!publish_draft_pr", "not publish_draft_pr") + expression = expression.replace("&&", "and").replace("||", "or") + actual = eval(expression, {"__builtins__": {}}, { + "identity_only": identity_only, + "publish_draft_pr": publish_draft_pr, + "generation_outcome": generation_outcome, + "package_upload_outcome": package_upload_outcome, + }) + expected = ( + not identity_only and generation_outcome == "success" + and (not publish_draft_pr or package_upload_outcome != "success") + ) + assert actual is expected + + +@pytest.mark.parametrize(("package_outcome", "expected"), [ + ("failure", "package_rejected"), + ("success", "artifact_upload_failed"), +]) +def test_fallback_artifact_does_not_hide_publication_package_failure(monkeypatch, tmp_path, capsys, package_outcome, expected): + for key, value in { + "PUBLICATION_REQUESTED": "true", + "GENERATION_OUTCOME": "success", + "ADMISSION_OUTCOME": "success", + "PACKAGE_OUTCOME": package_outcome, + "PACKAGE_UPLOAD_OUTCOME": "failure", + "SNAPSHOT_UPLOAD_OUTCOME": "success", + "GITHUB_STEP_SUMMARY": str(tmp_path / "summary"), + }.items(): + monkeypatch.setenv(key, value) + run_inline_python(identity_step("generation_summary")) + result = json.loads(capsys.readouterr().out) + assert result["generated"] is True + assert result["publication_requested"] is True + assert result["published"] is False + assert result["status"] == expected + + +@pytest.mark.parametrize("reason_code", [ + "API_WRITE_FORBIDDEN", "REF_WRITE_FORBIDDEN", "TREE_WRITE_FORBIDDEN", + "PR_WRITE_FORBIDDEN", "RECEIPT_NONCANONICAL", "PUBLICATION_FAILED", +]) +def test_summary_preserves_fixed_write_guard_reason_codes(monkeypatch, tmp_path, capsys, reason_code): + result_path = tmp_path / "publication-result.json" + result_path.write_text(json.dumps({"status": "failed", "reason_code": reason_code})) + for key in ("CHECKOUT_OUTCOME", "DOWNLOAD_OUTCOME", "VALIDATE_OUTCOME", "MINT_OUTCOME"): + monkeypatch.setenv(key, "success") + monkeypatch.setenv("RESULT_PATH", str(result_path)) + monkeypatch.setenv("GITHUB_STEP_SUMMARY", str(tmp_path / "summary")) + run_inline_python(publication_step("draft_publisher_summary")) + result = json.loads(capsys.readouterr().out) + assert result["reason_code"] == reason_code + assert result["published"] is False + + +@pytest.mark.parametrize("mutation", [None, "repository", "repository_id", "workflow_ref", "source_sha", "run_id", "run_attempt"]) +def test_summary_producer_provenance_comes_only_from_fixed_runtime(monkeypatch, tmp_path, capsys, mutation): + values = { + "GITHUB_EVENT_NAME": "workflow_dispatch", + "GITHUB_REF": "refs/heads/main", + "GITHUB_REPOSITORY": IDENTITY_REPOSITORY, + "GITHUB_REPOSITORY_ID": "1252248184", + "GITHUB_WORKFLOW_REF": f"{IDENTITY_REPOSITORY}/.github/workflows/{IDENTITY_WORKFLOW}@refs/heads/main", + "GITHUB_SHA": "a" * 40, + "GITHUB_RUN_ID": "1234", + "GITHUB_RUN_ATTEMPT": "2", + } + expected = { + "repository": values["GITHUB_REPOSITORY"], + "repository_id": values["GITHUB_REPOSITORY_ID"], + "workflow_ref": values["GITHUB_WORKFLOW_REF"], + "source_sha": values["GITHUB_SHA"], + "expected_base_sha": values["GITHUB_SHA"], + "run_id": values["GITHUB_RUN_ID"], + "run_attempt": values["GITHUB_RUN_ATTEMPT"], + } + if mutation is not None: + env_name = "GITHUB_SHA" if mutation == "source_sha" else "GITHUB_" + mutation.upper() + values[env_name] = "untrusted" + for key, value in values.items(): + monkeypatch.setenv(key, value) + result_path = tmp_path / "publication-result.json" + result_path.write_text(json.dumps({ + "status": "failed", "reason_code": "BASE_MOVED", + **{key: "arbitrary-receipt-string" for key in expected}, + })) + for key in ("CHECKOUT_OUTCOME", "DOWNLOAD_OUTCOME", "VALIDATE_OUTCOME", "MINT_OUTCOME"): + monkeypatch.setenv(key, "success") + monkeypatch.setenv("RESULT_PATH", str(result_path)) + monkeypatch.setenv("GITHUB_STEP_SUMMARY", str(tmp_path / "summary")) + run_inline_python(publication_step("draft_publisher_summary")) + logged = capsys.readouterr().out + assert "arbitrary-receipt-string" not in logged + result_path.read_text() + result = json.loads(logged) + if mutation in ("repository", "repository_id", "workflow_ref"): + assert not set(expected) & set(result) + else: + for key, value in expected.items(): + if key == mutation or (mutation == "source_sha" and key == "expected_base_sha"): + assert key not in result + else: + assert result[key] == value + assert result["published"] is False