From 777f9f60ecda970b240279c6eb40611eddbfe9ee Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Wed, 9 Sep 2026 23:52:25 +0800 Subject: [PATCH] Add bounded manual SOXL learning workflow Co-Authored-By: Codex --- .github/workflows/research_input_readback.yml | 99 ++++- .github/workflows/vps_codex_service_ops.yml | 2 +- scripts/deploy_codex_audit_service.sh | 2 +- scripts/run_soxl_manual_learning.py | 420 ++++++++++++++++++ .../test_research_input_readback_workflow.py | 14 +- tests/test_run_soxl_manual_learning.py | 388 ++++++++++++++++ 6 files changed, 918 insertions(+), 7 deletions(-) create mode 100644 scripts/run_soxl_manual_learning.py create mode 100644 tests/test_run_soxl_manual_learning.py diff --git a/.github/workflows/research_input_readback.yml b/.github/workflows/research_input_readback.yml index 7063911e..323570ae 100644 --- a/.github/workflows/research_input_readback.yml +++ b/.github/workflows/research_input_readback.yml @@ -2,6 +2,20 @@ name: VPS Research Input Readback on: workflow_dispatch: + inputs: + operation: + description: Bounded operation to perform + required: true + default: readback + type: choice + options: + - readback + - soxl_learning + blend_gate_mid_soxl_weights: + description: Reviewed baseline and up to two bounded learning values + required: true + default: '0.65,0.60,0.55' + type: string permissions: contents: read @@ -12,9 +26,9 @@ concurrency: jobs: readback: - if: github.repository == 'QuantStrategyLab/AIAuditBridge' && github.ref == 'refs/heads/main' + if: github.repository == 'QuantStrategyLab/AIAuditBridge' && github.ref == 'refs/heads/main' && (inputs.operation != 'soxl_learning' || github.run_attempt == 1) runs-on: [self-hosted, codex-vps] - timeout-minutes: 10 + timeout-minutes: ${{ inputs.operation == 'soxl_learning' && 35 || 10 }} permissions: contents: read id-token: write @@ -24,6 +38,16 @@ jobs: with: persist-credentials: false + - name: Initialize the sanitized learning record + if: inputs.operation == 'soxl_learning' + env: + LEARNING_OUTPUT: ${{ runner.temp }}/aab-soxl-learning-${{ github.run_id }}-${{ github.run_attempt }}/summary.json + run: | + set -euo pipefail + umask 077 + mkdir -m 700 "$(dirname "$LEARNING_OUTPUT")" + python3 -c 'import os; from pathlib import Path; from scripts.run_soxl_manual_learning import initialize_record; initialize_record(Path(os.environ["LEARNING_OUTPUT"]), {"repository": os.environ["GITHUB_REPOSITORY"], "ref": os.environ["GITHUB_REF"], "event_name": os.environ["GITHUB_EVENT_NAME"], "actor": os.environ["GITHUB_ACTOR"], "run_id": os.environ["GITHUB_RUN_ID"], "run_attempt": os.environ["GITHUB_RUN_ATTEMPT"]})' + - name: Checkout the frozen original P1 validator uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3 with: @@ -32,6 +56,24 @@ jobs: path: validator-source persist-credentials: false + - name: Checkout the bounded learning consumer + if: inputs.operation == 'soxl_learning' + uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3 + with: + repository: QuantStrategyLab/UsEquitySnapshotPipelines + ref: b03ecbe4e0a7a0de22f298499f867a7039e4b60a + path: consumer-source + persist-credentials: false + + - name: Checkout the frozen strategy runtime + if: inputs.operation == 'soxl_learning' + uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3 + with: + repository: QuantStrategyLab/UsEquityStrategies + ref: 7756fe32585e85cf1d09a163203a02e3eee39fe1 + path: ues-source + persist-credentials: false + - uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v6.2.0 with: python-version: '3.11' @@ -44,6 +86,21 @@ jobs: python -m pip install --quiet 'uv==0.11.19' env -u UV_PYTHON -u VIRTUAL_ENV uv sync --locked --no-dev --no-editable --python 3.11 + - name: Install the bounded learning consumer runtime + if: inputs.operation == 'soxl_learning' + working-directory: consumer-source + run: env -u UV_PYTHON -u VIRTUAL_ENV uv sync --locked --no-dev --no-editable --python 3.11 + + - name: Install the frozen strategy runtime + if: inputs.operation == 'soxl_learning' + working-directory: ues-source + env: + UV_PROJECT_ENVIRONMENT: ${{ runner.temp }}/aab-soxl-ues-env-${{ github.run_id }}-${{ github.run_attempt }} + run: | + env -u UV_PYTHON -u VIRTUAL_ENV uv sync --locked --no-dev --no-editable --python 3.11 + rm -rf -- build + test -z "$(git status --porcelain --untracked-files=all)" + # Direct federation grants only object reads on the fixed P1 root below. # The provider additionally binds numeric repo/owner IDs, main and this workflow. # No service-account key, user ADC, daemon credential or AI invocation is used. @@ -64,6 +121,14 @@ jobs: EXPECTED_MANIFEST_SHA256: 06ce97ad581fdc465896ef324ce16ac65f2695af95ec18ce27238bdd71032f74 READBACK_ROOT: ${{ runner.temp }}/aab-research-input-${{ github.run_id }}-${{ github.run_attempt }} VALIDATOR_SOURCE: ${{ github.workspace }}/validator-source + OPERATION: ${{ inputs.operation }} + PARAMETER_GRID: ${{ inputs.blend_gate_mid_soxl_weights }} + CONSUMER_SOURCE: ${{ github.workspace }}/consumer-source + UES_SOURCE: ${{ github.workspace }}/ues-source + LEARNING_OUTPUT: ${{ runner.temp }}/aab-soxl-learning-${{ github.run_id }}-${{ github.run_attempt }}/summary.json + CODEX_AUDIT_SERVICE_URL: ${{ secrets.CODEX_AUDIT_SERVICE_URL }} + CODEX_AUDIT_SERVICE_AUDIENCE: ${{ vars.CODEX_AUDIT_SERVICE_AUDIENCE || 'quant-codex-audit' }} + UES_ENV_ROOT: ${{ runner.temp }}/aab-soxl-ues-env-${{ github.run_id }}-${{ github.run_attempt }} run: | set -euo pipefail cleanup() { @@ -182,8 +247,36 @@ jobs: })) PY + if [ "$OPERATION" = "soxl_learning" ]; then + cd "$GITHUB_WORKSPACE" + UV_PROJECT_ENVIRONMENT="$UES_ENV_ROOT" python3 -m scripts.run_soxl_manual_learning \ + --parameter-grid "$PARAMETER_GRID" \ + --manifest-sha256 "$EXPECTED_MANIFEST_SHA256" \ + --root "$READBACK_ROOT/root" \ + --consumer-source "$CONSUMER_SOURCE" \ + --ues-source "$UES_SOURCE" \ + --output "$LEARNING_OUTPUT" \ + --repository "$GITHUB_REPOSITORY" \ + --ref "$GITHUB_REF" \ + --event-name "$GITHUB_EVENT_NAME" \ + --actor "$GITHUB_ACTOR" \ + --run-id "$GITHUB_RUN_ID" \ + --run-attempt "$GITHUB_RUN_ATTEMPT" + fi + + - name: Upload the sanitized learning record + if: always() && inputs.operation == 'soxl_learning' + uses: actions/upload-artifact@v7 + with: + name: soxl-manual-learning-${{ github.run_id }}-${{ github.run_attempt }} + path: ${{ runner.temp }}/aab-soxl-learning-${{ github.run_id }}-${{ github.run_attempt }}/summary.json + if-no-files-found: error + retention-days: 35 + - name: Remove the bounded readback workspace if: always() env: READBACK_ROOT: ${{ runner.temp }}/aab-research-input-${{ github.run_id }}-${{ github.run_attempt }} - run: rm -rf -- "$READBACK_ROOT" + LEARNING_OUTPUT_ROOT: ${{ runner.temp }}/aab-soxl-learning-${{ github.run_id }}-${{ github.run_attempt }} + UES_ENV_ROOT: ${{ runner.temp }}/aab-soxl-ues-env-${{ github.run_id }}-${{ github.run_attempt }} + run: rm -rf -- "$READBACK_ROOT" "$LEARNING_OUTPUT_ROOT" "$UES_ENV_ROOT" diff --git a/.github/workflows/vps_codex_service_ops.yml b/.github/workflows/vps_codex_service_ops.yml index 922673fa..1cab198f 100644 --- a/.github/workflows/vps_codex_service_ops.yml +++ b/.github/workflows/vps_codex_service_ops.yml @@ -61,7 +61,7 @@ jobs: CODEX_AUDIT_SSH_UNBAN_IP: ${{ inputs.ssh_unban_ip }} CODEX_AUDIT_SERVICE_ALLOWED_REPOSITORIES: QuantStrategyLab/AIAuditBridge,QuantStrategyLab/BinancePlatform,QuantStrategyLab/CharlesSchwabPlatform,QuantStrategyLab/CnEquitySnapshotPipelines,QuantStrategyLab/CnEquityStrategies,QuantStrategyLab/CryptoLivePoolPipelines,QuantStrategyLab/CryptoStrategies,QuantStrategyLab/FirstradePlatform,QuantStrategyLab/HkEquitySnapshotPipelines,QuantStrategyLab/HkEquityStrategies,QuantStrategyLab/IBKRGatewayManager,QuantStrategyLab/InteractiveBrokersPlatform,QuantStrategyLab/LongBridgePlatform,QuantStrategyLab/MarketSignalSources,QuantStrategyLab/PoliticalEventTrackingResearch,QuantStrategyLab/QmtPlatform,QuantStrategyLab/QuantAdvisorResearch,QuantStrategyLab/QuantPlatformKit,QuantStrategyLab/QuantRuntimeSettings,QuantStrategyLab/QuantStrategyPlugins,QuantStrategyLab/ResearchSignalContextPipelines,QuantStrategyLab/SchwabTokenAutoRefresher,QuantStrategyLab/UsEquitySnapshotPipelines,QuantStrategyLab/UsEquityStrategies # workflow_dispatch emits protected-main workflow_ref claims; the deploy script pins delegated QPK code by exact job_workflow_ref SHA. - CODEX_AUDIT_SERVICE_ALLOWED_WORKFLOW_REFS: QuantStrategyLab/AIAuditBridge/.github/workflows/codex_audit.yml@refs/heads/main,QuantStrategyLab/AIAuditBridge/.github/workflows/strategy_optimization_watcher.yml@refs/heads/main,QuantStrategyLab/AIAuditBridge/.github/workflows/portfolio_research_proposal_diagnosis.yml@refs/heads/main,QuantStrategyLab/CnEquityStrategies/.github/workflows/drift-check.yml@refs/heads/main,QuantStrategyLab/HkEquityStrategies/.github/workflows/drift-check.yml@refs/heads/main,QuantStrategyLab/UsEquityStrategies/.github/workflows/drift-check.yml@refs/heads/main,QuantStrategyLab/CryptoStrategies/.github/workflows/drift-check.yml@refs/heads/main + CODEX_AUDIT_SERVICE_ALLOWED_WORKFLOW_REFS: QuantStrategyLab/AIAuditBridge/.github/workflows/codex_audit.yml@refs/heads/main,QuantStrategyLab/AIAuditBridge/.github/workflows/research_input_readback.yml@refs/heads/main,QuantStrategyLab/AIAuditBridge/.github/workflows/strategy_optimization_watcher.yml@refs/heads/main,QuantStrategyLab/AIAuditBridge/.github/workflows/portfolio_research_proposal_diagnosis.yml@refs/heads/main,QuantStrategyLab/CnEquityStrategies/.github/workflows/drift-check.yml@refs/heads/main,QuantStrategyLab/HkEquityStrategies/.github/workflows/drift-check.yml@refs/heads/main,QuantStrategyLab/UsEquityStrategies/.github/workflows/drift-check.yml@refs/heads/main,QuantStrategyLab/CryptoStrategies/.github/workflows/drift-check.yml@refs/heads/main CODEX_AUDIT_SERVICE_ALLOWED_REFS: refs/heads/main # Exact canonical audit job plus immutable QPK `uses:` refs pinned by strategy drift callers. # Rotation tracked in #64; remove the old QPK SHA after final strategy-run verification. diff --git a/scripts/deploy_codex_audit_service.sh b/scripts/deploy_codex_audit_service.sh index cac42b6a..f0a231f3 100644 --- a/scripts/deploy_codex_audit_service.sh +++ b/scripts/deploy_codex_audit_service.sh @@ -11,7 +11,7 @@ ALLOWED_REPOSITORIES="${CODEX_AUDIT_SERVICE_ALLOWED_REPOSITORIES:-QuantStrategyL # Direct review and strategy workflow identities are pinned to protected main because GitHub emits workflow_ref with the dispatch branch. # Delegated reusable code is constrained separately by the exact job_workflow_ref SHA below. # The ref allowlist retains PR merge refs because GitHub can preserve the incoming PR ref for reusable calls; _verify_github_oidc requires both allowlists. -ALLOWED_WORKFLOW_REFS="${CODEX_AUDIT_SERVICE_ALLOWED_WORKFLOW_REFS:-QuantStrategyLab/AIAuditBridge/.github/workflows/codex_audit.yml@refs/heads/main,QuantStrategyLab/AIAuditBridge/.github/workflows/strategy_optimization_watcher.yml@refs/heads/main,QuantStrategyLab/AIAuditBridge/.github/workflows/portfolio_research_proposal_diagnosis.yml@refs/heads/main,QuantStrategyLab/CnEquityStrategies/.github/workflows/drift-check.yml@refs/heads/main,QuantStrategyLab/HkEquityStrategies/.github/workflows/drift-check.yml@refs/heads/main,QuantStrategyLab/UsEquityStrategies/.github/workflows/drift-check.yml@refs/heads/main,QuantStrategyLab/CryptoStrategies/.github/workflows/drift-check.yml@refs/heads/main}" +ALLOWED_WORKFLOW_REFS="${CODEX_AUDIT_SERVICE_ALLOWED_WORKFLOW_REFS:-QuantStrategyLab/AIAuditBridge/.github/workflows/codex_audit.yml@refs/heads/main,QuantStrategyLab/AIAuditBridge/.github/workflows/research_input_readback.yml@refs/heads/main,QuantStrategyLab/AIAuditBridge/.github/workflows/strategy_optimization_watcher.yml@refs/heads/main,QuantStrategyLab/AIAuditBridge/.github/workflows/portfolio_research_proposal_diagnosis.yml@refs/heads/main,QuantStrategyLab/CnEquityStrategies/.github/workflows/drift-check.yml@refs/heads/main,QuantStrategyLab/HkEquityStrategies/.github/workflows/drift-check.yml@refs/heads/main,QuantStrategyLab/UsEquityStrategies/.github/workflows/drift-check.yml@refs/heads/main,QuantStrategyLab/CryptoStrategies/.github/workflows/drift-check.yml@refs/heads/main}" ALLOWED_REFS="${CODEX_AUDIT_SERVICE_ALLOWED_REFS:-refs/heads/main}" ALLOWED_REPOSITORY_VISIBILITIES="${CODEX_AUDIT_SERVICE_ALLOWED_REPOSITORY_VISIBILITIES:-public}" # Exact canonical audit job identity. Single source of truth for delegated drift code follows. diff --git a/scripts/run_soxl_manual_learning.py b/scripts/run_soxl_manual_learning.py new file mode 100644 index 00000000..a63f8729 --- /dev/null +++ b/scripts/run_soxl_manual_learning.py @@ -0,0 +1,420 @@ +#!/usr/bin/env python3 +"""Gate one bounded, human-dispatched SOXL three-asset learning run through Codex.""" + +from __future__ import annotations + +import argparse +import json +import math +import re +import subprocess +from collections.abc import Callable, Mapping, Sequence +from pathlib import Path +from typing import Any + +from client.config import GatewayConfig +from client.gateway_client import AiGatewayClient + +ADVICE_SCHEMA = "qsl.soxl-manual-learning-advice.v1" +ARTIFACT_SCHEMA = "qsl.soxl-manual-learning-run.v1" +NUMERIC_SCHEMA = "qsl.soxl-soxx-three-asset-learning.v1" +REPLAY_SCHEMA = "qsl.soxl-soxx-three-asset-learning-replay-result.v1" +BASELINE = 0.65 +COST_BPS = (5.0, 10.0, 15.0) +DEVELOPMENT_CUTOFF = "2025-07-31" +UESP_REVISION = "b03ecbe4e0a7a0de22f298499f867a7039e4b60a" +UES_REVISION = "7756fe32585e85cf1d09a163203a02e3eee39fe1" +EXPECTED_REPOSITORY = "QuantStrategyLab/AIAuditBridge" +EXPECTED_REF = "refs/heads/main" +EXPECTED_EVENT = "workflow_dispatch" +SAFE_REASON = re.compile(r"[a-z0-9_]{1,64}\Z") + + +class ManualLearningError(ValueError): + """A fixed, safe failure at the manual-learning boundary.""" + + +def parse_parameter_grid(value: str) -> tuple[float, ...]: + try: + parts = value.split(",") + values = tuple(float(item.strip()) for item in parts) + except (AttributeError, ValueError) as exc: + raise ManualLearningError("learning_parameters_invalid") from exc + if ( + not 1 <= len(values) <= 3 + or any(not math.isfinite(item) or item < 0 or item > BASELINE for item in values) + or len(set(values)) != len(values) + or BASELINE not in values + ): + raise ManualLearningError("learning_parameters_invalid") + return values + + +def _validate_authority(context: Mapping[str, str]) -> None: + if ( + context.get("repository") != EXPECTED_REPOSITORY + or context.get("ref") != EXPECTED_REF + or context.get("event_name") != EXPECTED_EVENT + or not context.get("actor", "").strip() + or not context.get("run_id", "").isdigit() + or context.get("run_attempt") != "1" + ): + raise ManualLearningError("manual_authority_invalid") + + +def _safe_base(context: Mapping[str, str], values: Sequence[float], manifest_sha256: str) -> dict[str, Any]: + return { + "schema_version": ARTIFACT_SCHEMA, + "operation": "soxl_learning", + "status": "unavailable", + "authority": { + "event_name": context["event_name"], + "actor": context["actor"], + "repository": context["repository"], + "ref": context["ref"], + "run_id": context["run_id"], + "run_attempt": 1, + }, + "parameter_key": "blend_gate_mid_soxl_weight", + "parameter_values": list(values), + "cost_bps": list(COST_BPS), + "development_cutoff": DEVELOPMENT_CUTOFF, + "input_identity": {"manifest_sha256": manifest_sha256, "member_count": 4}, + "consumer_source": { + "repository": "QuantStrategyLab/UsEquitySnapshotPipelines", + "revision": UESP_REVISION, + }, + "learning_only": True, + "no_order": True, + "size_zero_required": True, + "promotion_eligible": False, + "research_executed": False, + } + + +def _prompt(context: Mapping[str, str], values: Sequence[float], manifest_sha256: str) -> str: + request = { + "task": "human_dispatched_soxl_three_asset_learning_advice", + "authority": { + "event_name": context["event_name"], "actor": context["actor"], + "run_id": context["run_id"], "run_attempt": 1, + }, + "facts": { + "manual_research": True, "drift_detected": False, "fault_detected": False, + "symbols": ["SOXL", "SOXX", "BOXX"], + "parameter_key": "blend_gate_mid_soxl_weight", + "parameter_values": list(values), "baseline": BASELINE, + "cost_bps": list(COST_BPS), "development_cutoff": DEVELOPMENT_CUTOFF, + "input_manifest_sha256": manifest_sha256, + "learning_only": True, "no_order": True, "promotion_eligible": False, + }, + "allowed_decision": "Recommend execute or reject for this exact fixed grid only.", + "forbidden": [ + "change code, parameters, commands, data permissions, symbols, or source revisions", + "claim drift, fault, promotion, WFA, OOS, shadow, or trading authority", + ], + "required_output": { + "schema_version": ADVICE_SCHEMA, + "recommendation": "execute|reject", + "reason_code": "short_fixed_identifier", + "parameter_values": list(values), + "learning_only": True, "no_order": True, "promotion_eligible": False, + }, + } + return json.dumps(request, sort_keys=True, separators=(",", ":")) + + +def _advice(result: object, values: Sequence[float]) -> tuple[dict[str, Any] | None, str]: + raw = getattr(result, "raw", None) + if isinstance(raw, Mapping) and raw.get("status") == "deferred": + return None, "deferred" + output = getattr(result, "output", None) + if not ( + getattr(result, "success", False) is True + and getattr(result, "provider", None) == "codex" + and isinstance(getattr(result, "model", None), str) + and bool(result.model.strip()) + and isinstance(output, str) + and output.strip() + and not getattr(result, "error", "") + and not getattr(result, "note", "") + and isinstance(raw, Mapping) + and raw.get("status") == "succeeded" + and raw.get("provider") == "codex" + and raw.get("research_stage") == "optimization" + and raw.get("model") == result.model + and raw.get("reasoning_effort") in {"low", "medium", "high", "xhigh"} + and isinstance(raw.get("job_id"), str) + and bool(raw["job_id"].strip()) + and raw.get("output") == output + and raw.get("policy_verdict", "advisory") in {"ok", "eligible", "advisory"} + ): + return None, "unavailable" + try: + value = json.loads(output) + except (TypeError, json.JSONDecodeError): + return None, "unavailable" + if not isinstance(value, dict) or set(value) != { + "schema_version", "recommendation", "reason_code", "parameter_values", + "learning_only", "no_order", "promotion_eligible", + }: + return None, "unavailable" + if ( + value["schema_version"] != ADVICE_SCHEMA + or value["recommendation"] not in {"execute", "reject"} + or not isinstance(value["reason_code"], str) + or SAFE_REASON.fullmatch(value["reason_code"]) is None + or value["parameter_values"] != list(values) + or value["learning_only"] is not True + or value["no_order"] is not True + or value["promotion_eligible"] is not False + ): + return None, "unavailable" + return { + "status": "succeeded", "job_id": raw["job_id"], "provider": "codex", + "model": result.model, "reasoning_effort": raw["reasoning_effort"], + "research_stage": "optimization", "recommendation": value["recommendation"], + "reason_code": value["reason_code"], + }, "succeeded" + + +def _numeric_command(root: Path, consumer: Path, ues: Path, values: Sequence[float]) -> list[str]: + command = [ + str(consumer / ".venv/bin/python"), + str(consumer / "scripts/run_soxl_three_asset_learning.py"), + "--p1-binding", str(root / "binding.json"), + "--input-manifest", str(root / "manifest.json"), + "--bars-member", str(root / "bars.json"), + "--ues-project", str(ues), + "--p2-candidate", str(consumer / "config/soxl_soxx_core_only_p2_v3.json"), + ] + for value in values: + command.extend(("--blend-gate-mid-soxl-weight", f"{value:g}")) + return command + + +def _sanitize_numeric(value: object, values: Sequence[float], manifest_sha256: str) -> tuple[list[dict[str, Any]], dict[str, Any], str]: + if not isinstance(value, Mapping) or value.get("schema_version") != NUMERIC_SCHEMA or value.get("status") != "SUCCESS": + raise ManualLearningError("numeric_result_invalid") + if ( + value.get("learning_only") is not True or value.get("no_order") is not True + or value.get("size_zero_required") is not True or value.get("promotion_eligible") is not False + or value.get("research_executed") is not True or value.get("development_cutoff") != DEVELOPMENT_CUTOFF + or value.get("parameter_key") != "blend_gate_mid_soxl_weight" + or value.get("trial_count") != len(values) or value.get("cost_bps") != list(COST_BPS) + ): + raise ManualLearningError("numeric_result_invalid") + p1 = value.get("p1_identity") + source = value.get("source_identity") + if not isinstance(p1, Mapping) or p1.get("input_manifest_sha256") != manifest_sha256: + raise ManualLearningError("numeric_result_invalid") + if ( + not isinstance(source, Mapping) + or source.get("repository") != "QuantStrategyLab/UsEquityStrategies" + or source.get("revision") != UES_REVISION + or not isinstance(source.get("quant_platform_kit_revision"), str) + or re.fullmatch(r"[0-9a-f]{40}", source["quant_platform_kit_revision"]) is None + or not isinstance(source.get("uv_lock_sha256"), str) + or re.fullmatch(r"[0-9a-f]{64}", source["uv_lock_sha256"]) is None + ): + raise ManualLearningError("numeric_result_invalid") + results = value.get("results") + expected = [(parameter, cost) for parameter in values for cost in COST_BPS] + if not isinstance(results, list) or len(results) != len(expected): + raise ManualLearningError("numeric_result_invalid") + safe: list[dict[str, Any]] = [] + metrics = ("strategy_profile", "sharpe_ratio", "max_drawdown", "cagr", "volatility", "total_return", "start_date", "end_date", "observation_count") + for item, (parameter, cost) in zip(results, expected, strict=True): + if not isinstance(item, Mapping) or item.get("schema_version") != REPLAY_SCHEMA or item.get("status") != "SUCCESS" or item.get("parameter_override") != {"blend_gate_mid_soxl_weight": parameter} or item.get("cost_bps") != cost: + raise ManualLearningError("numeric_result_invalid") + backtest = item.get("backtest_result") + if not isinstance(backtest, Mapping) or any(key not in backtest for key in metrics): + raise ManualLearningError("numeric_result_invalid") + numeric_metrics = ("sharpe_ratio", "max_drawdown", "cagr", "volatility", "total_return") + if any( + isinstance(backtest[key], bool) + or not isinstance(backtest[key], (int, float)) + or not math.isfinite(float(backtest[key])) + for key in numeric_metrics + ): + raise ManualLearningError("numeric_result_invalid") + if ( + not isinstance(backtest["observation_count"], int) + or isinstance(backtest["observation_count"], bool) + or backtest["observation_count"] < 2 + or not isinstance(backtest["strategy_profile"], str) + or backtest["strategy_profile"] != "soxl_soxx_three_asset_mid_weight_learning_v1" + or not isinstance(backtest["start_date"], str) + or re.fullmatch(r"\d{4}-\d{2}-\d{2}", backtest["start_date"]) is None + or not isinstance(backtest["end_date"], str) + or re.fullmatch(r"\d{4}-\d{2}-\d{2}", backtest["end_date"]) is None + or backtest["start_date"] > backtest["end_date"] + or backtest["end_date"] > DEVELOPMENT_CUTOFF + or not isinstance(item.get("output_sha256"), str) + or re.fullmatch(r"[0-9a-f]{64}", item["output_sha256"]) is None + ): + raise ManualLearningError("numeric_result_invalid") + safe.append({ + "parameter_override": {"blend_gate_mid_soxl_weight": parameter}, + "cost_bps": cost, + "backtest_result": {key: backtest[key] for key in metrics}, + "output_sha256": item.get("output_sha256"), + }) + digest = value.get("result_sha256") + if not isinstance(digest, str) or re.fullmatch(r"[0-9a-f]{64}", digest) is None: + raise ManualLearningError("numeric_result_invalid") + safe_source = { + key: source[key] + for key in ("repository", "revision", "quant_platform_kit_revision", "uv_lock_sha256") + } + return safe, safe_source, digest + + +def run_manual_learning( + *, parameter_grid: str, manifest_sha256: str, root: Path, consumer_source: Path, + ues_source: Path, context: Mapping[str, str], + client_factory: Callable[[GatewayConfig], Any] = AiGatewayClient, + gateway_config: GatewayConfig | None = None, + command_runner: Callable[[list[str]], Any] | None = None, + progress_writer: Callable[[Mapping[str, Any]], None] | None = None, +) -> dict[str, Any]: + values = parse_parameter_grid(parameter_grid) + _validate_authority(context) + if len(manifest_sha256) != 64 or any(char not in "0123456789abcdef" for char in manifest_sha256): + raise ManualLearningError("input_identity_invalid") + required = ( + root / "binding.json", root / "manifest.json", root / "bars.json", + consumer_source / "scripts/run_soxl_three_asset_learning.py", + consumer_source / "config/soxl_soxx_core_only_p2_v3.json", + ) + interpreter = consumer_source / ".venv/bin/python" + if ( + any(path.is_symlink() or not path.is_file() for path in required) + or not interpreter.is_file() + or ues_source.is_symlink() + or not ues_source.is_dir() + ): + raise ManualLearningError("source_or_input_unavailable") + artifact = _safe_base(context, values, manifest_sha256) + try: + config = gateway_config or GatewayConfig.from_env() + result = client_factory(config).execute( + _prompt(context, values, manifest_sha256), mode="review_only", + research_stage="optimization", allowed_providers=["codex"], + source_repository=EXPECTED_REPOSITORY, source_ref="main", timeout=600, + ) + except Exception: # noqa: BLE001 - provider detail must not cross this boundary + artifact["failure_stage"] = "codex_unavailable" + return artifact + advice, state = _advice(result, values) + if advice is None: + artifact["status"] = state + artifact["failure_stage"] = "codex_admission_or_result" + return artifact + artifact["ai_execution"] = advice + if advice["recommendation"] != "execute": + artifact["status"] = "rejected" + return artifact + runner = command_runner or (lambda argv: subprocess.run(argv, capture_output=True, text=True, timeout=1200, check=False)) + artifact["numeric_execution"] = {"status": "started"} + artifact["research_executed"] = None + if progress_writer is not None: + progress_writer(artifact) + try: + completed = runner(_numeric_command(root, consumer_source, ues_source, values)) + except (OSError, subprocess.SubprocessError): + artifact["status"] = "parked" + artifact["failure_stage"] = "numeric_outcome_unknown" + artifact["numeric_execution"] = {"status": "outcome_unknown"} + return artifact + if getattr(completed, "returncode", None) != 0: + artifact["status"] = "parked" + artifact["failure_stage"] = "numeric_execution_failed" + artifact["numeric_execution"] = {"status": "failed"} + return artifact + try: + numeric = json.loads(completed.stdout) + safe, source, digest = _sanitize_numeric(numeric, values, manifest_sha256) + except (AttributeError, TypeError, json.JSONDecodeError, ManualLearningError): + artifact["status"] = "parked" + artifact["failure_stage"] = "numeric_result_invalid" + artifact["numeric_execution"] = {"status": "outcome_unknown"} + return artifact + artifact.update( + status="accepted", research_executed=True, numeric_summary=safe, + numeric_source_identity=source, numeric_result_sha256=digest, + numeric_execution={"status": "succeeded"}, + ) + return artifact + + +def _write(path: Path, value: Mapping[str, Any]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(value, sort_keys=True, separators=(",", ":")) + "\n") + + +def initialize_record(path: Path, context: Mapping[str, str]) -> None: + """Write the bound terminal placeholder before setup or remote reads begin.""" + _validate_authority(context) + _write( + path, + { + "schema_version": ARTIFACT_SCHEMA, + "operation": "soxl_learning", + "status": "parked", + "failure_stage": "setup_incomplete", + "authority": { + "event_name": context["event_name"], + "actor": context["actor"], + "repository": context["repository"], + "ref": context["ref"], + "run_id": context["run_id"], + "run_attempt": 1, + }, + "research_executed": False, + "learning_only": True, + "no_order": True, + "size_zero_required": True, + "promotion_eligible": False, + }, + ) + + +def main(argv: Sequence[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--parameter-grid", required=True) + parser.add_argument("--manifest-sha256", required=True) + parser.add_argument("--root", required=True, type=Path) + parser.add_argument("--consumer-source", required=True, type=Path) + parser.add_argument("--ues-source", required=True, type=Path) + parser.add_argument("--output", required=True, type=Path) + parser.add_argument("--repository", required=True) + parser.add_argument("--ref", required=True) + parser.add_argument("--event-name", required=True) + parser.add_argument("--actor", required=True) + parser.add_argument("--run-id", required=True) + parser.add_argument("--run-attempt", required=True) + args = parser.parse_args(argv) + context = {key: getattr(args, key) for key in ("repository", "ref", "event_name", "actor", "run_id", "run_attempt")} + try: + result = run_manual_learning( + parameter_grid=args.parameter_grid, manifest_sha256=args.manifest_sha256, + root=args.root, consumer_source=args.consumer_source, ues_source=args.ues_source, + context=context, progress_writer=lambda value: _write(args.output, value), + ) + exit_code = 0 if result["status"] == "accepted" else 2 + except ManualLearningError as exc: + result = { + "schema_version": ARTIFACT_SCHEMA, "operation": "soxl_learning", + "status": "parked", "failure_stage": str(exc), "research_executed": False, + "learning_only": True, "no_order": True, "size_zero_required": True, + "promotion_eligible": False, + } + exit_code = 2 + _write(args.output, result) + print(json.dumps({"status": result["status"], "operation": "soxl_learning"}, sort_keys=True)) + return exit_code + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_research_input_readback_workflow.py b/tests/test_research_input_readback_workflow.py index 5c770d66..b9c49907 100644 --- a/tests/test_research_input_readback_workflow.py +++ b/tests/test_research_input_readback_workflow.py @@ -198,17 +198,27 @@ def test_validator_rejection_outputs_only_the_fixed_safe_category( assert "accepted" not in output -def test_workflow_keeps_frozen_source_main_identity_and_no_research_execution() -> None: +def test_workflow_keeps_readback_default_and_bounds_manual_learning() -> None: text = workflow_text() assert "github.ref == 'refs/heads/main'" in text + assert "inputs.operation != 'soxl_learning' || github.run_attempt == 1" in text + assert "default: readback" in text + assert "- soxl_learning" in text assert "repository: QuantStrategyLab/UsEquitySnapshotPipelines" in text assert "ref: ca61b82c2a508a1cc81fb5831294ba9835ac41c2" in text + assert "ref: b03ecbe4e0a7a0de22f298499f867a7039e4b60a" in text + assert "ref: 7756fe32585e85cf1d09a163203a02e3eee39fe1" in text assert "uv sync --locked --no-dev --no-editable --python 3.11" in text assert text.index("name: Install the frozen validator runtime") < text.index( "name: Authenticate for this research input only" ) assert "trap cleanup EXIT" in text assert "if: always()" in text - assert "upload-artifact" not in text + assert "if: always() && inputs.operation == 'soxl_learning'" in text + assert "python3 -m scripts.run_soxl_manual_learning" in text + assert 'UV_PROJECT_ENVIRONMENT="$UES_ENV_ROOT" python3 -m scripts.run_soxl_manual_learning' in text + assert text.index("uv run --no-sync python") < text.index( + 'UV_PROJECT_ENVIRONMENT="$UES_ENV_ROOT" python3 -m scripts.run_soxl_manual_learning' + ) assert "run_soxl_core_only_p3_evidence" not in text diff --git a/tests/test_run_soxl_manual_learning.py b/tests/test_run_soxl_manual_learning.py new file mode 100644 index 00000000..ece08753 --- /dev/null +++ b/tests/test_run_soxl_manual_learning.py @@ -0,0 +1,388 @@ +from __future__ import annotations + +import json +import os +import subprocess +import sys +import time +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import patch + +import pytest + +from scripts.run_soxl_manual_learning import ( + ManualLearningError, + initialize_record, + parse_parameter_grid, + run_manual_learning, +) + + +UES_REVISION = "7756fe32585e85cf1d09a163203a02e3eee39fe1" + + +class FakeClient: + def __init__(self, result: object) -> None: + self.result = result + self.calls: list[tuple[str, dict[str, object]]] = [] + + def execute(self, prompt: str, **kwargs: object) -> object: + self.calls.append((prompt, kwargs)) + return self.result + + +def ai_result(*, recommendation: str = "execute", status: str = "succeeded") -> SimpleNamespace: + output = json.dumps( + { + "schema_version": "qsl.soxl-manual-learning-advice.v1", + "recommendation": recommendation, + "reason_code": "bounded_hypothesis_worth_testing", + "parameter_values": [0.65, 0.6, 0.55], + "learning_only": True, + "no_order": True, + "promotion_eligible": False, + } + ) + return SimpleNamespace( + provider="codex", + model="gpt-5.6-sol", + success=status == "succeeded", + output=output, + error="" if status == "succeeded" else "private provider detail", + note="deferred" if status == "deferred" else "", + raw={ + "status": status, + "job_id": "job-123", + "provider": "codex", + "research_stage": "optimization", + "model": "gpt-5.6-sol", + "reasoning_effort": "medium", + "output": output, + "policy_verdict": "advisory", + }, + ) + + +@pytest.fixture +def paths(tmp_path: Path) -> dict[str, Path]: + root = tmp_path / "root" + root.mkdir() + for name in ("binding.json", "manifest.json", "bars.json"): + (root / name).write_text("{}") + consumer = tmp_path / "consumer" + (consumer / "scripts").mkdir(parents=True) + (consumer / "config").mkdir() + (consumer / ".venv/bin").mkdir(parents=True) + (consumer / ".venv/bin/python").symlink_to(sys.executable) + (consumer / "scripts/run_soxl_three_asset_learning.py").write_text("# synthetic") + (consumer / "config/soxl_soxx_core_only_p2_v3.json").write_text("{}") + ues = tmp_path / "ues" + ues.mkdir() + return {"root": root, "consumer": consumer, "ues": ues} + + +def context() -> dict[str, str]: + return { + "repository": "QuantStrategyLab/AIAuditBridge", + "ref": "refs/heads/main", + "event_name": "workflow_dispatch", + "actor": "human-reviewer", + "run_id": "12345", + "run_attempt": "1", + } + + +def numeric_result() -> dict[str, object]: + rows = [] + for parameter in (0.65, 0.6, 0.55): + for cost in (5.0, 10.0, 15.0): + rows.append( + { + "schema_version": "qsl.soxl-soxx-three-asset-learning-replay-result.v1", + "status": "SUCCESS", + "parameter_override": {"blend_gate_mid_soxl_weight": parameter}, + "cost_bps": cost, + "backtest_result": { + "strategy_profile": "soxl_soxx_three_asset_mid_weight_learning_v1", + "sharpe_ratio": 0.25, + "max_drawdown": 0.1, + "cagr": 0.05, + "volatility": 0.2, + "total_return": 0.03, + "start_date": "2025-01-02", + "end_date": "2025-07-31", + "observation_count": 80, + "target_values": {"SOXL": 100.0}, + }, + "output_sha256": "a" * 64, + } + ) + return { + "schema_version": "qsl.soxl-soxx-three-asset-learning.v1", + "status": "SUCCESS", + "learning_profile": "soxl_soxx_three_asset_mid_weight_learning_v1", + "learning_only": True, + "no_order": True, + "size_zero_required": True, + "promotion_eligible": False, + "research_executed": True, + "development_cutoff": "2025-07-31", + "p1_identity": {"input_manifest_sha256": "0" * 64}, + "source_identity": { + "repository": "QuantStrategyLab/UsEquityStrategies", + "revision": UES_REVISION, + "quant_platform_kit_revision": "3acab1923a97b805b077c85c6c19657be0143bac", + "uv_lock_sha256": "6c12df9b3412681829295f15de7e2ce7fc5b708d1de815f72d654fc16b7848e6", + "private_path": "/private/source", + }, + "parameter_key": "blend_gate_mid_soxl_weight", + "trial_count": 3, + "cost_bps": [5.0, 10.0, 15.0], + "results": rows, + "result_sha256": "b" * 64, + } + + +def test_parameter_grid_requires_unique_bounded_baseline() -> None: + assert parse_parameter_grid("0.65,0.60,0.55") == (0.65, 0.6, 0.55) + for invalid in ("0.60", "0.65,0.65", "0.65,0.60,0.55,0.50", "0.65,nan", "0.70,0.65"): + with pytest.raises(ManualLearningError): + parse_parameter_grid(invalid) + + +def test_initialized_terminal_record_binds_the_manual_run(tmp_path: Path) -> None: + output = tmp_path / "summary.json" + initialize_record(output, context()) + value = json.loads(output.read_text()) + assert value["status"] == "parked" + assert value["failure_stage"] == "setup_incomplete" + assert value["authority"]["run_id"] == "12345" + assert value["research_executed"] is False + + +def test_invalid_manual_context_calls_neither_ai_nor_numeric(paths: dict[str, Path]) -> None: + client = FakeClient(ai_result()) + commands: list[list[str]] = [] + bad = context() | {"run_attempt": "2"} + with pytest.raises(ManualLearningError, match="manual_authority_invalid"): + run_manual_learning( + parameter_grid="0.65,0.60,0.55", manifest_sha256="0" * 64, + root=paths["root"], consumer_source=paths["consumer"], ues_source=paths["ues"], + context=bad, client_factory=lambda _config: client, + gateway_config=SimpleNamespace(), + command_runner=lambda argv: commands.append(argv), + ) + assert not client.calls + assert not commands + + +@pytest.mark.parametrize("kind", ["deferred", "unknown", "reject"]) +def test_non_executable_codex_outcome_never_calls_numeric( + kind: str, paths: dict[str, Path] +) -> None: + result = ai_result(recommendation="reject" if kind == "reject" else "execute", status="deferred" if kind == "deferred" else "succeeded") + if kind == "unknown": + result.raw["job_id"] = "" + client = FakeClient(result) + commands: list[list[str]] = [] + artifact = run_manual_learning( + parameter_grid="0.65,0.60,0.55", manifest_sha256="0" * 64, + root=paths["root"], consumer_source=paths["consumer"], ues_source=paths["ues"], + context=context(), client_factory=lambda _config: client, + gateway_config=SimpleNamespace(), + command_runner=lambda argv: commands.append(argv), + ) + assert artifact["status"] in {"deferred", "unavailable", "rejected"} + assert artifact["research_executed"] is False + assert not commands + + +def test_free_text_reason_code_is_rejected_before_numeric(paths: dict[str, Path]) -> None: + result = ai_result() + payload = json.loads(result.output) + payload["reason_code"] = "private model prose with spaces" + result.output = json.dumps(payload) + result.raw["output"] = result.output + client = FakeClient(result) + commands: list[list[str]] = [] + artifact = run_manual_learning( + parameter_grid="0.65,0.60,0.55", manifest_sha256="0" * 64, + root=paths["root"], consumer_source=paths["consumer"], ues_source=paths["ues"], + context=context(), client_factory=lambda _config: client, + command_runner=lambda argv: commands.append(argv), gateway_config=SimpleNamespace(), + ) + assert artifact["status"] == "unavailable" + assert not commands + + +def test_success_calls_one_codex_job_then_exact_fixed_numeric_cli( + paths: dict[str, Path] +) -> None: + client = FakeClient(ai_result()) + commands: list[list[str]] = [] + + def run_command(argv: list[str]) -> SimpleNamespace: + commands.append(argv) + return SimpleNamespace(returncode=0, stdout=json.dumps(numeric_result()), stderr="") + + artifact = run_manual_learning( + parameter_grid="0.65,0.60,0.55", manifest_sha256="0" * 64, + root=paths["root"], consumer_source=paths["consumer"], ues_source=paths["ues"], + context=context(), client_factory=lambda _config: client, command_runner=run_command, + gateway_config=SimpleNamespace(), + ) + + assert len(client.calls) == 1 + assert client.calls[0][1] == { + "mode": "review_only", "research_stage": "optimization", + "allowed_providers": ["codex"], "source_repository": "QuantStrategyLab/AIAuditBridge", + "source_ref": "main", "timeout": 600, + } + assert len(commands) == 1 + argv = commands[0] + assert argv[:2] == [ + str(paths["consumer"] / ".venv/bin/python"), + str(paths["consumer"] / "scripts/run_soxl_three_asset_learning.py"), + ] + assert argv.count("--blend-gate-mid-soxl-weight") == 3 + assert "0.65" in argv and "0.6" in argv and "0.55" in argv + assert artifact["status"] == "accepted" + assert artifact["research_executed"] is True + assert artifact["learning_only"] is True + assert artifact["no_order"] is True + assert artifact["size_zero_required"] is True + assert artifact["promotion_eligible"] is False + assert artifact["authority"]["actor"] == "human-reviewer" + assert artifact["ai_execution"]["job_id"] == "job-123" + assert len(artifact["numeric_summary"]) == 9 + serialized = json.dumps(artifact) + assert "decisions" not in serialized + assert "target_values" not in serialized + assert "private_path" not in serialized + assert "private provider detail" not in serialized + + +def test_numeric_failure_is_sanitized_and_nonaccepted(paths: dict[str, Path]) -> None: + client = FakeClient(ai_result()) + + def fail(_argv: list[str]) -> SimpleNamespace: + return SimpleNamespace(returncode=2, stdout="private bars", stderr="private trace") + + artifact = run_manual_learning( + parameter_grid="0.65,0.60,0.55", manifest_sha256="0" * 64, + root=paths["root"], consumer_source=paths["consumer"], ues_source=paths["ues"], + context=context(), client_factory=lambda _config: client, command_runner=fail, + gateway_config=SimpleNamespace(), + ) + assert artifact["status"] == "parked" + assert artifact["research_executed"] is None + assert artifact["failure_stage"] == "numeric_execution_failed" + assert artifact["numeric_execution"] == {"status": "failed"} + assert artifact["ai_execution"]["job_id"] == "job-123" + assert "private" not in json.dumps(artifact) + + +def test_numeric_unknown_keeps_manual_and_ai_binding(paths: dict[str, Path]) -> None: + client = FakeClient(ai_result()) + progress: list[dict[str, object]] = [] + + def unknown(_argv: list[str]) -> object: + raise subprocess.TimeoutExpired("private command", 1) + + artifact = run_manual_learning( + parameter_grid="0.65,0.60,0.55", manifest_sha256="0" * 64, + root=paths["root"], consumer_source=paths["consumer"], ues_source=paths["ues"], + context=context(), client_factory=lambda _config: client, command_runner=unknown, + gateway_config=SimpleNamespace(), + progress_writer=lambda value: progress.append(dict(value)), + ) + assert artifact["status"] == "parked" + assert artifact["research_executed"] is None + assert artifact["numeric_execution"] == {"status": "outcome_unknown"} + assert artifact["authority"]["run_id"] == "12345" + assert artifact["ai_execution"]["job_id"] == "job-123" + assert progress[0]["research_executed"] is None + assert progress[0]["numeric_execution"] == {"status": "started"} + + +def test_cli_failure_writes_only_safe_terminal_and_exits_nonzero( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + from scripts import run_soxl_manual_learning as module + + output = tmp_path / "summary.json" + monkeypatch.setattr( + module, + "run_manual_learning", + lambda **_kwargs: (_ for _ in ()).throw(ManualLearningError("numeric_execution_unavailable")), + ) + code = module.main( + [ + "--parameter-grid", "0.65,0.60", "--manifest-sha256", "0" * 64, + "--root", str(tmp_path), "--consumer-source", str(tmp_path), + "--ues-source", str(tmp_path), "--output", str(output), + "--repository", "QuantStrategyLab/AIAuditBridge", "--ref", "refs/heads/main", + "--event-name", "workflow_dispatch", "--actor", "reviewer", + "--run-id", "123", "--run-attempt", "1", + ] + ) + assert code != 0 + assert json.loads(output.read_text()) == { + "schema_version": "qsl.soxl-manual-learning-run.v1", + "operation": "soxl_learning", + "status": "parked", + "failure_stage": "numeric_execution_unavailable", + "research_executed": False, + "learning_only": True, + "no_order": True, + "size_zero_required": True, + "promotion_eligible": False, + } + assert "private" not in capsys.readouterr().out + + +def test_gateway_auth_accepts_only_the_exact_manual_workflow_on_main() -> None: + from scripts import codex_audit_service + + workflow_ref = ( + "QuantStrategyLab/AIAuditBridge/.github/workflows/" + "research_input_readback.yml@refs/heads/main" + ) + payload = { + "aud": "quant-codex-audit", + "iss": codex_audit_service.GITHUB_OIDC_ISSUER, + "exp": int(time.time()) + 300, + "repository": "QuantStrategyLab/AIAuditBridge", + "workflow_ref": workflow_ref, + "ref": "refs/heads/main", + "repository_visibility": "public", + } + env = { + "CODEX_AUDIT_SERVICE_ALLOWED_REPOSITORIES": "QuantStrategyLab/AIAuditBridge", + "CODEX_AUDIT_SERVICE_ALLOWED_WORKFLOW_REFS": workflow_ref, + "CODEX_AUDIT_SERVICE_ALLOWED_REFS": "refs/heads/main", + "CODEX_AUDIT_SERVICE_ALLOWED_DIRECT_REPOSITORIES": "QuantStrategyLab/AIAuditBridge", + "CODEX_AUDIT_SERVICE_ALLOWED_REPOSITORY_VISIBILITIES": "public", + } + + def verify(active: dict[str, object]) -> dict[str, object]: + with ( + patch.dict(os.environ, env, clear=True), + patch.object( + codex_audit_service, + "_jwt_parts", + return_value=({"alg": "RS256", "kid": "1"}, active, b"x", b"y"), + ), + patch.object( + codex_audit_service, + "_load_jwks", + return_value={"keys": [{"kid": "1", "kty": "RSA"}]}, + ), + patch.object(codex_audit_service, "_verify_rs256", return_value=None), + ): + return codex_audit_service._verify_github_oidc("header.payload.signature") + + assert verify(payload)["workflow_ref"] == workflow_ref + with pytest.raises(PermissionError, match="workflow_ref .* not allowed"): + verify(payload | {"workflow_ref": workflow_ref.replace("refs/heads/main", "refs/heads/other")})