diff --git a/.github/actionlint.yaml b/.github/actionlint.yaml new file mode 100644 index 0000000..5ebf472 --- /dev/null +++ b/.github/actionlint.yaml @@ -0,0 +1,3 @@ +self-hosted-runner: + labels: + - research-runner diff --git a/.github/workflows/cn-index-etf-research.yml b/.github/workflows/cn-index-etf-research.yml new file mode 100644 index 0000000..0067134 --- /dev/null +++ b/.github/workflows/cn-index-etf-research.yml @@ -0,0 +1,38 @@ +name: CN Index ETF Research V2 +on: + workflow_dispatch: + inputs: + execute: + description: Run the root-policy-bound cloud experiment + type: boolean + default: false +permissions: + contents: read + id-token: write +concurrency: + group: cn-index-etf-research-v2 + cancel-in-progress: false +jobs: + research: + if: github.repository == 'QuantStrategyLab/CnEquityStrategies' && github.ref == 'refs/heads/main' && github.run_attempt == 1 && vars.AI_SERVICE_RELEASE_READY == 'true' + runs-on: [self-hosted, research-runner] + timeout-minutes: 60 + env: + AI_SERVICE_URL: ${{ vars.AI_SERVICE_URL }} + AI_SERVICE_AUDIENCE: ${{ vars.AI_SERVICE_AUDIENCE }} + AI_SERVICE_MODE: agent + AI_SERVICE_MODEL: ${{ vars.AI_SERVICE_RESEARCH_MODEL }} + AI_SERVICE_PROFILE: ${{ vars.AI_SERVICE_RESEARCH_PROFILE || 'default' }} + AI_SERVICE_SUMMARY_MODEL: ${{ vars.AI_SERVICE_SUMMARY_MODEL }} + STRATEGY_WATCH_SOURCE_REPO: QuantStrategyLab/CnEquitySnapshotPipelines + GH_TOKEN: ${{ github.token }} + WATCHER_RESULT: ${{ vars.CN_RESEARCH_WATCHER_RESULT_PATH }} + EXECUTE: ${{ inputs.execute }} + steps: + - name: Run the preinstalled approved source revision + run: | + set -euo pipefail + test -n "$WATCHER_RESULT" + args=(--dry-run) + if [[ "$EXECUTE" == "true" ]]; then args=(); fi + python -m cn_equity_strategies.backtest.index_etf_research_task --watcher-result "$WATCHER_RESULT" "${args[@]}" diff --git a/.github/workflows/drift-check.yml b/.github/workflows/drift-check.yml index 0e6a3c9..86bf988 100644 --- a/.github/workflows/drift-check.yml +++ b/.github/workflows/drift-check.yml @@ -194,7 +194,7 @@ jobs: contents: read issues: write id-token: write - uses: QuantStrategyLab/QuantPlatformKit/.github/workflows/reusable-drift-check.yml@c7646a7168b3dafa763ef7751a182d23e8de7790 + uses: ./.github/workflows/reusable-drift-check.yml with: strategy_domain: cn_equity caller_event_name: ${{ github.event_name }} @@ -202,9 +202,10 @@ jobs: snapshot_repository: QuantStrategyLab/CnEquitySnapshotPipelines snapshot_checkout_path: external/CnEquitySnapshotPipelines snapshot_repository_ref: ${{ needs.preflight_backtests.outputs.snapshot_repository_ref }} - ai_gateway_service_url: ${{ vars.AI_GATEWAY_SERVICE_URL }} + ai_service_url: ${{ vars.AI_SERVICE_URL }} + ai_service_audience: ${{ vars.AI_SERVICE_AUDIENCE }} + ai_service_reviewers_json: ${{ vars.AI_SERVICE_REVIEWERS_JSON }} quant_platform_kit_ref: c7646a7168b3dafa763ef7751a182d23e8de7790 lifecycle_preflight_artifact: lifecycle-preflight-${{ github.run_id }}-${{ github.run_attempt }} secrets: - codex_audit_service_url: ${{ secrets.CODEX_AUDIT_SERVICE_URL }} snapshot_repository_token: ${{ secrets.SNAPSHOT_REPOSITORY_TOKEN || secrets.QSL_REPO_SYNC_TOKEN || github.token }} diff --git a/.github/workflows/evidence-gate.yml b/.github/workflows/evidence-gate.yml index 166e81e..0386636 100644 --- a/.github/workflows/evidence-gate.yml +++ b/.github/workflows/evidence-gate.yml @@ -70,17 +70,10 @@ jobs: python -m pip install -e ".[research]" pandas python -m pip check - - name: Checkout AIAuditBridge - uses: actions/checkout@v6 - with: - repository: QuantStrategyLab/AIAuditBridge - ref: 60bd64a2ae059a082614181eeb845b46df395523 - path: external/AIAuditBridge - - name: Evaluate Evidence Gate env: GITHUB_BASE_REF: ${{ github.base_ref }} - AIAUDIT_BRIDGE_ROOT: external/AIAuditBridge - CODEX_AUDIT_SERVICE_URL: ${{ secrets.CODEX_AUDIT_SERVICE_URL }} - AI_GATEWAY_SERVICE_URL: ${{ vars.AI_GATEWAY_SERVICE_URL }} + AI_SERVICE_URL: ${{ vars.AI_SERVICE_URL }} + AI_SERVICE_AUDIENCE: ${{ vars.AI_SERVICE_AUDIENCE }} + AI_SERVICE_REVIEWERS_JSON: ${{ vars.AI_SERVICE_REVIEWERS_JSON }} run: python scripts/gate_evidence_package.py diff --git a/.github/workflows/reusable-drift-check.yml b/.github/workflows/reusable-drift-check.yml new file mode 100644 index 0000000..716a3ae --- /dev/null +++ b/.github/workflows/reusable-drift-check.yml @@ -0,0 +1,352 @@ +name: Reusable Drift Check + +on: + workflow_call: + inputs: + strategy_domain: + required: true + type: string + snapshot_repository: + required: true + type: string + snapshot_checkout_path: + required: true + type: string + snapshot_repository_ref: + required: false + type: string + default: "main" + python_version: + required: false + type: string + default: "3.11" + quant_platform_kit_ref: + required: false + type: string + default: "main" + ai_service_url: + required: false + type: string + default: "" + ai_service_audience: + required: false + type: string + default: "" + ai_service_reviewers_json: + required: false + type: string + default: "[]" + lifecycle_performance_bucket: + required: false + type: string + default: "" + caller_event_name: + required: true + type: string + caller_pr_head_repository: + required: false + type: string + default: "" + lifecycle_preflight_artifact: + required: false + type: string + default: "" + strategy_profile: + required: false + type: string + default: "" + gcp_workload_identity_provider: + description: "Optional WIF provider for writing LIFECYCLE_PERFORMANCE_BUCKET." + required: false + type: string + default: "" + gcp_service_account: + description: "Optional GCP service account email for lifecycle bucket writes." + required: false + type: string + default: "" + gcp_project_id: + description: "Optional GCP project id for gcloud setup." + required: false + type: string + default: "" + live_stream_id: + description: >- + Optional PerformanceStore live_runs stream id for monitor. When empty, + monitor uses preflight/research return CSVs only (not paper/live account + PnL). Set only for a verified account-bound stream. + required: false + type: string + default: "" + secrets: + snapshot_repository_token: + required: false + +permissions: + contents: read + issues: write + id-token: write + +jobs: + drift: + if: vars.AI_SERVICE_RELEASE_READY == 'true' + runs-on: ubuntu-latest + # A critical drift can require up to three sequential primary reviews. Each + # service review has a 15-minute budget plus its polling grace period. + timeout-minutes: 60 + env: + FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true" + STRATEGY_DOMAIN: ${{ inputs.strategy_domain }} + QUANT_PROJECTS_ROOT: ${{ github.workspace }}/external + LIFECYCLE_PERFORMANCE_BUCKET: ${{ inputs.lifecycle_performance_bucket || vars.LIFECYCLE_PERFORMANCE_BUCKET || '' }} + LIFECYCLE_LOCAL_ROOT: ${{ github.workspace }}/data/lifecycle_store + LIFECYCLE_STRATEGY_PROFILE: ${{ inputs.strategy_profile }} + LIFECYCLE_LIVE_STREAM_ID: ${{ inputs.live_stream_id }} + + steps: + - name: Validate trusted caller + env: + CALLER_REPOSITORY: ${{ github.repository }} + CALLER_EVENT_NAME: ${{ inputs.caller_event_name }} + CALLER_PR_HEAD_REPOSITORY: ${{ inputs.caller_pr_head_repository }} + SNAPSHOT_REPOSITORY: ${{ inputs.snapshot_repository }} + shell: bash + run: | + set -euo pipefail + caller_repo="$CALLER_REPOSITORY" + caller_event="$CALLER_EVENT_NAME" + caller_pr_head_repo="$CALLER_PR_HEAD_REPOSITORY" + snapshot_repo="$SNAPSHOT_REPOSITORY" + expected_snapshot_repo="" + case "$caller_repo" in + QuantStrategyLab/CnEquityStrategies) + expected_snapshot_repo="QuantStrategyLab/CnEquitySnapshotPipelines" + ;; + QuantStrategyLab/HkEquityStrategies) + expected_snapshot_repo="QuantStrategyLab/HkEquitySnapshotPipelines" + ;; + QuantStrategyLab/UsEquityStrategies) + expected_snapshot_repo="QuantStrategyLab/UsEquitySnapshotPipelines" + ;; + QuantStrategyLab/CryptoStrategies) + expected_snapshot_repo="QuantStrategyLab/CryptoLivePoolPipelines" + ;; + *) + echo "::error::Untrusted reusable workflow caller: $caller_repo" + exit 1 + ;; + esac + case "$caller_event" in + schedule|workflow_dispatch) + ;; + pull_request|pull_request_target) + if [ -n "$caller_pr_head_repo" ] && [ "$caller_pr_head_repo" != "$caller_repo" ]; then + echo "::error::Fork pull_request callers are not trusted: $caller_pr_head_repo" + else + echo "::error::pull_request callers are not trusted: $caller_repo" + fi + exit 1 + ;; + "") + echo "::error::caller_event_name must be provided by the caller workflow" + exit 1 + ;; + *) + echo "::error::Unsupported caller_event_name: $caller_event" + exit 1 + ;; + esac + if [ "$snapshot_repo" != "$expected_snapshot_repo" ]; then + echo "::error::snapshot_repository mismatch for caller $caller_repo: $snapshot_repo" + exit 1 + fi + + - name: Checkout + uses: actions/checkout@v6 + + - name: Checkout snapshot pipeline repo + uses: actions/checkout@v6 + with: + repository: ${{ inputs.snapshot_repository }} + ref: ${{ inputs.snapshot_repository_ref }} + token: ${{ secrets.snapshot_repository_token || github.token }} + path: ${{ inputs.snapshot_checkout_path }} + + - name: Checkout QuantPlatformKit + uses: actions/checkout@v6 + with: + repository: QuantStrategyLab/QuantPlatformKit + ref: ${{ inputs.quant_platform_kit_ref }} + path: external/QuantPlatformKit + + - name: Set up Python + uses: actions/setup-python@v6 + with: + python-version: ${{ inputs.python_version }} + + - name: Install dependencies + run: | + set -euo pipefail + python -m pip install --upgrade pip + python -m pip install -e . pandas + python -m pip install --no-deps -e external/QuantPlatformKit + # Required when LIFECYCLE_PERFORMANCE_BUCKET is set (PerformanceStore GCS client). + python -m pip install "google-cloud-storage>=2.14" + + - name: Gate cloud lifecycle writes without GCP auth + env: + WIF_PROVIDER: ${{ inputs.gcp_workload_identity_provider }} + WIF_SA: ${{ inputs.gcp_service_account }} + shell: bash + run: | + set -euo pipefail + if [ -n "${LIFECYCLE_PERFORMANCE_BUCKET:-}" ] && { [ -z "${WIF_PROVIDER}" ] || [ -z "${WIF_SA}" ]; }; then + echo "::warning::LIFECYCLE_PERFORMANCE_BUCKET is set without GCP WIF inputs; keeping local-only lifecycle store for this run" + echo "LIFECYCLE_PERFORMANCE_BUCKET=" >> "$GITHUB_ENV" + fi + + - name: Authenticate to Google Cloud + if: ${{ inputs.gcp_workload_identity_provider != '' && inputs.gcp_service_account != '' }} + uses: google-github-actions/auth@v3 + with: + workload_identity_provider: ${{ inputs.gcp_workload_identity_provider }} + service_account: ${{ inputs.gcp_service_account }} + + - name: Set up gcloud + if: ${{ inputs.gcp_workload_identity_provider != '' && inputs.gcp_service_account != '' && inputs.gcp_project_id != '' }} + uses: google-github-actions/setup-gcloud@v3 + with: + project_id: ${{ inputs.gcp_project_id }} + + - name: Download lifecycle preflight artifact + if: inputs.lifecycle_preflight_artifact != '' + uses: actions/download-artifact@v7 + with: + name: ${{ inputs.lifecycle_preflight_artifact }} + path: ${{ runner.temp }}/lifecycle-preflight + + - name: Restore lifecycle preflight inputs + if: inputs.lifecycle_preflight_artifact != '' + env: + PREFLIGHT_BUNDLE_ROOT: ${{ runner.temp }}/lifecycle-preflight + SNAPSHOT_CHECKOUT_PATH: ${{ inputs.snapshot_checkout_path }} + shell: bash + run: | + set -euo pipefail + case "$SNAPSHOT_CHECKOUT_PATH" in + external/*) ;; + *) echo "::error::snapshot_checkout_path must remain under external/"; exit 1 ;; + esac + if [[ "$SNAPSHOT_CHECKOUT_PATH" == *".."* ]]; then + echo "::error::snapshot_checkout_path must not contain traversal segments" + exit 1 + fi + if find "$PREFLIGHT_BUNDLE_ROOT" -type l -print -quit | grep -q .; then + echo "::error::lifecycle preflight artifact must not contain symlinks" + exit 1 + fi + lifecycle_source="$PREFLIGHT_BUNDLE_ROOT/data/lifecycle_store" + snapshot_source="$PREFLIGHT_BUNDLE_ROOT/$SNAPSHOT_CHECKOUT_PATH/data/output" + if [ ! -d "$lifecycle_source" ] || [ ! -d "$snapshot_source" ]; then + echo "::error::lifecycle preflight artifact is missing required allowlisted directories" + exit 1 + fi + snapshot_root="$GITHUB_WORKSPACE/$SNAPSHOT_CHECKOUT_PATH" + resolved_snapshot_root="$(realpath "$snapshot_root")" + case "$resolved_snapshot_root" in + "$GITHUB_WORKSPACE"/external/*) ;; + *) echo "::error::snapshot checkout resolved outside the external workspace"; exit 1 ;; + esac + if find "$snapshot_root" -type l -print -quit | grep -q .; then + echo "::error::snapshot checkout must not contain symlinks" + exit 1 + fi + snapshot_target="$snapshot_root/data/output" + rm -rf -- "$LIFECYCLE_LOCAL_ROOT" "$snapshot_target" + mkdir -p "$LIFECYCLE_LOCAL_ROOT" "$GITHUB_WORKSPACE/$SNAPSHOT_CHECKOUT_PATH/data/output" + cp -a "$lifecycle_source/." "$LIFECYCLE_LOCAL_ROOT/" + cp -a "$snapshot_source/." "$snapshot_target/" + if [ "$STRATEGY_DOMAIN" = "crypto" ]; then + echo "CRYPTO_LIFECYCLE_PREFLIGHT_ROOT=$PREFLIGHT_BUNDLE_ROOT" >> "$GITHUB_ENV" + fi + + - name: Build lifecycle performance snapshots + shell: bash + run: | + set -euo pipefail + # Without LIFECYCLE_LIVE_STREAM_ID, monitor reads preflight/research + # portfolio_and_tracker_returns.csv proxies — not paper/live account PnL. + lifecycle_args=() + if [ -n "${LIFECYCLE_STRATEGY_PROFILE:-}" ]; then + lifecycle_args+=(--strategy "${LIFECYCLE_STRATEGY_PROFILE}") + fi + if [ -n "${LIFECYCLE_LIVE_STREAM_ID:-}" ]; then + lifecycle_args+=(--live-stream-id "${LIFECYCLE_LIVE_STREAM_ID}") + fi + quant-lifecycle monitor --domain ${{ inputs.strategy_domain }} --source-revision "${GITHUB_SHA}" "${lifecycle_args[@]}" + + - name: Validate lifecycle prerequisites + shell: bash + run: | + set -euo pipefail + lifecycle_args=() + if [ -n "${LIFECYCLE_STRATEGY_PROFILE:-}" ]; then + lifecycle_args+=(--strategy "${LIFECYCLE_STRATEGY_PROFILE}") + fi + quant-lifecycle doctor --domain ${{ inputs.strategy_domain }} --require-snapshot --require-backtest --max-freshness-days 7 "${lifecycle_args[@]}" + + - name: Install approved task-service artifacts + env: + AI_SERVICE_CLIENT_WHEEL: ${{ vars.AI_SERVICE_CLIENT_WHEEL }} + AI_SERVICE_CLIENT_SHA256: ${{ vars.AI_SERVICE_CLIENT_SHA256 }} + AI_SERVICE_QPK_WHEEL: ${{ vars.AI_SERVICE_QPK_WHEEL }} + AI_SERVICE_QPK_SHA256: ${{ vars.AI_SERVICE_QPK_SHA256 }} + run: | + python - <<'PYCODE' + import hashlib, os, pathlib, re, subprocess, sys + for prefix, name in (("CLIENT", "personal_ai_service-2.0.0-"), ("QPK", "quant_platform_kit-1.0.0-")): + path = pathlib.Path(os.environ.get("AI_SERVICE_" + prefix + "_WHEEL", "")) + digest = os.environ.get("AI_SERVICE_" + prefix + "_SHA256", "") + if not re.fullmatch(r"[0-9a-f]{64}", digest) or not path.is_file() or path.is_symlink() or not path.name.startswith(name) or path.suffix != ".whl": + raise SystemExit("approved artifact required") + if hashlib.sha256(path.read_bytes()).hexdigest() != digest: + raise SystemExit("artifact digest mismatch") + subprocess.run([sys.executable, "-m", "pip", "install", "--no-deps", "--no-index", str(path)], check=True) + PYCODE + + - name: Run drift detection + shell: bash + run: | + set -euo pipefail + lifecycle_args=() + if [ -n "${LIFECYCLE_STRATEGY_PROFILE:-}" ]; then + lifecycle_args+=(--strategy "${LIFECYCLE_STRATEGY_PROFILE}") + fi + quant-lifecycle drift --domain ${{ inputs.strategy_domain }} --no-alerts "${lifecycle_args[@]}" + + - name: Sync drift alerts to GitHub Issues + env: + GH_TOKEN: ${{ github.token }} + GITHUB_REPOSITORY: ${{ github.repository }} + run: | + python - <<'PY' + import os + + from quant_platform_kit.strategy_lifecycle.codex_integration import create_issues_for_domain + + owner, _, repository = os.environ["GITHUB_REPOSITORY"].partition("/") + if not owner or not repository: + raise RuntimeError("GITHUB_REPOSITORY must identify the reusable workflow caller") + os.environ["CODEX_AUDIT_ORG"] = owner + os.environ["CODEX_AUDIT_ORCHESTRATOR_REPO"] = repository + results = create_issues_for_domain("${{ inputs.strategy_domain }}") + created = [item for item in results if item.get("issue_url")] + print({"created_issues": len(created), "results": len(results)}) + PY + + - name: Review critical drift through task service + env: + AI_SERVICE_URL: ${{ inputs.ai_service_url }} + AI_SERVICE_AUDIENCE: ${{ inputs.ai_service_audience }} + AI_SERVICE_REVIEWERS_JSON: ${{ inputs.ai_service_reviewers_json }} + run: python -m quant_platform_kit.strategy_lifecycle.drift_review --domain "$STRATEGY_DOMAIN" diff --git a/docs/ai-service/TASK-SERVICE-V2.zh-CN.md b/docs/ai-service/TASK-SERVICE-V2.zh-CN.md new file mode 100644 index 0000000..46ac7d1 --- /dev/null +++ b/docs/ai-service/TASK-SERVICE-V2.zh-CN.md @@ -0,0 +1,19 @@ +# CN ETF 通用任务调用 + +本地迁移基线：CnEquityStrategies `8f2c7296ddad5245203f445667f18fa41dc13d91`。没有修改主工作区、提交、发布、部署或调用真实模型。 + +`backtest/index_etf_research_task.py` 接管原 AIAuditBridge 的 CN ETF 研究调用、根用户保护的本地政策、来源与问题绑定、每日一次新研究准入、真实前向观察及终态归档校验。策略边界由本仓库提供，AI 服务只接收通用任务。 + +新入口身份为 `QuantStrategyLab/CnEquityStrategies/.github/workflows/cn-index-etf-research.yml@refs/heads/main`；新政策与持久目录分别为 `/etc/cn-equity-strategies-policy/` 和 `/var/lib/cn-equity-strategies/`。这些只是本地代码约定，尚未创建或更改云端配置。旧研究记录不能直接搬到新目录并视为已授权；需按历史来源和原任务身份审核。 + +任务结果完成并绑定原请求后才能解释研究建议。等待或未知结果保存 task ID；恢复只查询原任务。原有 QPK 持久流程新增 `read_pending_diagnosis` 传递，冻结数值输入、参数空间、手续费、回测和人工审批不由助手选择。已过前向窗口不能触发新研究；读取既有任务不会制造新的观察证据。 + +离线验证覆盖根用户政策、Actions/OIDC 身份、任务来源、每日准入、日历与 shadow 证据、归档幂等、等待恢复及既有数值任务测试。测试输入均为 synthetic。 + +发行尚未完成：新的入口 workflow、已核验 wheel 安装与依赖版本更新待落实。`research` extra 和锁文件已移除旧 AIAuditBridge SDK，证据审查 workflow 不再 checkout AIAuditBridge。QPK 的现有固定版本还没有新接口，必须在同批正式发行后更新；现在不能直接切换入口。CN 保护政策仍要求 QPK 从非 editable 的固定 Git 提交安装并核对 direct_url；本地 wheel 只用于打包验收，不替代该生产来源准入。PersonalAIService 使用经过核验的构建材料。没有伪造新的远端提交号或已发布包。 + +证据审查由策略调用方构造任务，保持 primary、secondary、independent 三名角色。未完成、缺少角色、非法格式或冲突不会通过；服务配置缺失返回阻断，不能因旧脚本缺失而静默成功。保留原有显式 `DUAL_REVIEW_GATE_SKIP` 操作入口；没有新增自动跳过。漂移 workflow 的旧复用入口和 CN 新研究 workflow 仍待迁移。 + +## 所有者流程收尾 + +漂移流程使用本仓库的新 reusable workflow，模型审查改为可配置三角色；来源、快照和金融准入保持原有边界。策略研究的新手动 workflow 默认关闭，通过 AI_SERVICE_RELEASE_READY 与经过批准的云端环境启用，未增加交易或自动采用权限。共享依赖必须使用准确批准的源码/产物；当前未发布，不能改成一个不存在的远端 SHA。 diff --git a/pyproject.toml b/pyproject.toml index 4025b68..95e7cda 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -10,12 +10,12 @@ readme = "README.md" requires-python = ">=3.11" dependencies = [ "pandas>=2.0", - "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@c7646a7168b3dafa763ef7751a182d23e8de7790", + "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@f6f2079f6b53d3cbb3b72bdcc5c2476ed8f170f1", ] [project.optional-dependencies] test = ["pytest>=8"] -research = ["ai-gateway-client @ git+https://github.com/QuantStrategyLab/AIAuditBridge.git@60bd64a2ae059a082614181eeb845b46df395523"] +research = [] [tool.setuptools] package-dir = {"" = "src"} diff --git a/qsl.toml b/qsl.toml index 9ef6833..8f64167 100644 --- a/qsl.toml +++ b/qsl.toml @@ -5,6 +5,5 @@ upgrade_ring = "ring_b" bundle = "2026.09.1" requires = [ "pandas>=2.0", - "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@c7646a7168b3dafa763ef7751a182d23e8de7790", - "ai-gateway-client @ git+https://github.com/QuantStrategyLab/AIAuditBridge.git@60bd64a2ae059a082614181eeb845b46df395523", + "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@f6f2079f6b53d3cbb3b72bdcc5c2476ed8f170f1", ] diff --git a/scripts/gate_evidence_package.py b/scripts/gate_evidence_package.py index 89caf51..ca873d6 100755 --- a/scripts/gate_evidence_package.py +++ b/scripts/gate_evidence_package.py @@ -4,6 +4,8 @@ from __future__ import annotations import os +import json +import tomllib import re import subprocess import sys @@ -94,38 +96,38 @@ def _validate_with_promotion_standard(path: Path) -> tuple[bool, list[str]]: def _run_promotion_dual_review(evidence_files: list[Path]) -> int: - root = Path(os.environ.get("AIAUDIT_BRIDGE_ROOT", "external/AIAuditBridge")) - script = root / "scripts" / "run_dual_review_pipeline.py" - if not script.is_file(): - print("[evidence-gate] dual-review skipped: AIAuditBridge not found") - return 0 if str(os.environ.get("DUAL_REVIEW_GATE_SKIP", "")).strip().lower() in {"1", "true", "yes"}: print("[evidence-gate] dual-review skipped by DUAL_REVIEW_GATE_SKIP") return 0 - - failed = False + try: + from quant_platform_kit.strategy_lifecycle.task_review import review_material + except ImportError: + print("[evidence-gate] review task runtime unavailable", file=sys.stderr) + return 3 for path in evidence_files: - proc = subprocess.run( - [ - sys.executable, - str(script), - "--from-evidence", - str(path), - "--dispatch", - ], - cwd=str(root), - env={**os.environ, "PYTHONPATH": str(root)}, - check=False, - ) - print(f"[evidence-gate] dual-review {path.name} exit={proc.returncode}") - if proc.stdout: - print(proc.stdout.strip()) - if proc.stderr: - print(proc.stderr.strip(), file=sys.stderr) - failed = failed or proc.returncode != 0 - if failed: - print("::error::Dual-review blocked promotion (review did not succeed)", file=sys.stderr) - return 1 + try: + if path.is_symlink() or not path.is_file() or path.stat().st_size > 200000: + raise ValueError("invalid evidence file") + text = path.read_text(encoding="utf-8") + material = tomllib.loads(text) if path.suffix == ".toml" else json.loads(text) + if not isinstance(material, dict): + raise ValueError("invalid evidence object") + caller_ref = os.environ.get("GITHUB_SHA", "") + if re.fullmatch(r"[0-9a-f]{40}", caller_ref) is None: + raise ValueError("caller revision required") + result = review_material( + material, operation_id="cn-evidence:" + caller_ref + ":" + path.as_posix(), + required_roles=("reviewer-primary", "reviewer-secondary", "reviewer-independent"), + ) + except Exception: + print("[evidence-gate] review task unavailable", file=sys.stderr) + return 3 + if result.get("status") != "completed": + print("[evidence-gate] review quorum pending or unavailable", file=sys.stderr) + return 3 + if result.get("outcome") != "agree_approve" or result.get("advisory_only") is not True: + print("::error::Evidence review needs human resolution", file=sys.stderr) + return 1 return 0 diff --git a/src/cn_equity_strategies/backtest/index_etf_research_job.py b/src/cn_equity_strategies/backtest/index_etf_research_job.py index 1cba9d1..6d44ae2 100644 --- a/src/cn_equity_strategies/backtest/index_etf_research_job.py +++ b/src/cn_equity_strategies/backtest/index_etf_research_job.py @@ -387,7 +387,7 @@ def run_index_etf_research_job( locked_oos_start: date, locked_oos_end: date, purge_days: int, embargo_days: int, code_revision: str, ticket_dir, store_root, as_of, drift_score, source_revision, record_shadow, sync_console, diagnose=None, summarize=None, pull_console=None, admit_new_research=None, - read_pending_shadow=None, + read_pending_shadow=None, read_pending_diagnosis=None, research_owner=None, config=None, cost_model=None, evaluation_date=None, ) -> dict[str, Any]: @@ -395,7 +395,7 @@ def run_index_etf_research_job( No command text/model result can select this configuration. The deployment binds historical roots only after approving provider/license/range evidence. - AAB keeps its job/lease; QPK owns stage persistence. This is not a scheduler. + The owning caller admits work; QPK owns stage persistence. This is not a scheduler. """ from pathlib import Path from collections.abc import Mapping @@ -439,6 +439,8 @@ def run_index_etf_research_job( # A dependency without the persisted interface fails before any remote call. import inspect required_qpk_parameters = {"research_identity", "admit_new_research", "read_pending_shadow"} + if read_pending_diagnosis is not None: + required_qpk_parameters.add("read_pending_diagnosis") if summarize is not None: required_qpk_parameters.add("summarize") if research_owner is not None: @@ -453,6 +455,8 @@ def run_index_etf_research_job( admit_new_research=admit_new_research, read_pending_shadow=read_pending_shadow, research_owner=research_owner, ) + if read_pending_diagnosis is not None: + promotion_kwargs["read_pending_diagnosis"] = read_pending_diagnosis if summarize is not None: promotion_kwargs["summarize"] = summarize result = dict(run_actionable_research_promotion(**promotion_kwargs)) diff --git a/src/cn_equity_strategies/backtest/index_etf_research_task.py b/src/cn_equity_strategies/backtest/index_etf_research_task.py new file mode 100644 index 0000000..05aed5e --- /dev/null +++ b/src/cn_equity_strategies/backtest/index_etf_research_task.py @@ -0,0 +1,766 @@ +#!/usr/bin/env python3 +"""One policy-bound CN experiment, owned by the existing watcher Actions job. + +The watcher task is a trigger, not optimization or trading authority. Only the +root-owned local policy selects inputs, installed code and the bounded runner. +QPK's existing ticket is the sole progress/admission record. +""" +from __future__ import annotations + +import argparse +from datetime import date, datetime, timezone +import hashlib +import importlib.metadata +import json +import math +import os +from pathlib import Path +import re +import stat +import subprocess +from collections.abc import Mapping +from types import SimpleNamespace +from typing import Any +from zoneinfo import ZoneInfo + +from quant_platform_kit.strategy_lifecycle.research_task import validate_strategy_diagnosis_task + +PROFILE = "cn_index_etf_tactical_rotation" +STRATEGY_REPOSITORY = "QuantStrategyLab/CnEquityStrategies" +CALLER_REPOSITORY = "QuantStrategyLab/CnEquityStrategies" +ISSUE_REPOSITORY = "QuantStrategyLab/CnEquitySnapshotPipelines" +WORKFLOW_REF = f"{CALLER_REPOSITORY}/.github/workflows/cn-index-etf-research.yml@refs/heads/main" +POLICY_PATH = Path("/etc/cn-equity-strategies-policy/cn-index-etf-research.json") +STATE_ROOT = Path("/var/lib/cn-equity-strategies/cn-index-etf-research") +_REVISION = re.compile(r"[0-9a-f]{40}") +_ISSUE_URL = re.compile(r"^https://github\.com/([A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+)/issues/([1-9][0-9]*)$") +_WATCHER_KEY = re.compile(r"[A-Za-z0-9_-]{8,64}") +_IDENTITY_FIELDS = {"code_revision", "input_revision", "param_space_revision", "cost_model_revision", "validator_revision"} + + +def _now() -> datetime: + return datetime.now(timezone.utc) + + +def _json(raw: str | bytes) -> Any: + def pairs(items): + value = {} + for key, item in items: + if key in value: + raise ValueError("duplicate_json_key") + value[key] = item + return value + return json.loads(raw, object_pairs_hook=pairs, + parse_constant=lambda _: (_ for _ in ()).throw(ValueError("nonfinite_json"))) + + +def _read_bytes(path: Path, *, limit: int = 2_000_000) -> bytes: + if path.is_symlink() or not path.is_file() or path.stat().st_size > limit: + raise ValueError("input_file_unavailable") + raw = path.read_bytes() + if len(raw) > limit: + raise ValueError("input_file_unavailable") + return raw + + +def _read_json(path: Path, *, limit: int = 2_000_000) -> Any: + return _json(_read_bytes(path, limit=limit)) + + +def _protected_file(path: Path, *, secret: bool = False) -> None: + if not path.is_absolute() or path.is_symlink() or not path.is_file(): + raise ValueError("protected_file_required") + for item in (path, *path.parents): + info = item.lstat() + if stat.S_ISLNK(info.st_mode) or info.st_uid != 0 or info.st_mode & 0o022: + raise ValueError("protected_file_required") + if secret and path.stat().st_mode & 0o007: + raise ValueError("protected_secret_required") + + +def _read_policy(path: Path) -> dict[str, Any]: + if not path.exists(): + raise FileNotFoundError + _protected_file(path) + value = _read_json(path, limit=128_000) + if not isinstance(value, dict) or type(value.get("enabled")) is not bool: + raise ValueError("policy_invalid") + return value + + +def _workflow_authenticated() -> bool: + return ( + os.environ.get("GITHUB_REPOSITORY") == CALLER_REPOSITORY + and os.environ.get("GITHUB_REF") == "refs/heads/main" + and os.environ.get("GITHUB_WORKFLOW_REF") == WORKFLOW_REF + and os.environ.get("GITHUB_EVENT_NAME") in {"schedule", "workflow_dispatch"} + and bool(re.fullmatch(r"[1-9][0-9]*", os.environ.get("GITHUB_RUN_ID", ""))) + and bool(_REVISION.fullmatch(os.environ.get("GITHUB_SHA", ""))) + and bool(os.environ.get("ACTIONS_ID_TOKEN_REQUEST_URL")) + and bool(os.environ.get("ACTIONS_ID_TOKEN_REQUEST_TOKEN")) + and not os.environ.get("AI_SERVICE_TOKEN") + ) + + +def _timestamp(value: Any) -> datetime: + if not isinstance(value, str): + raise ValueError("timestamp_required") + timestamp = datetime.fromisoformat(value.replace("Z", "+00:00")) + if timestamp.tzinfo is None: + raise ValueError("timestamp_timezone_required") + return timestamp.astimezone(timezone.utc) + + +def _select_task(watcher: dict, revision: str, now: datetime, *, allow_saved_shadow: bool = False) -> dict: + source = watcher["research_task_source_snapshot"] + if source.get("schema_version") != "qsl_research_task_source_snapshot.v1" or source.get("data_status") != "ready": + raise ValueError("watcher_task_unavailable") + matches = [] + for raw in source["tasks"]: + task = validate_strategy_diagnosis_task(raw) + target = task["target"] + if target["candidate_id"] != PROFILE: + continue + if (target["repository"] != STRATEGY_REPOSITORY or target["domain"] != "cn_equity" + or target["candidate_kind"] != "individual" or target["strategy_revision"] != revision): + raise ValueError("watcher_task_identity_mismatch") + age = (now - _timestamp(task["created_at"])).total_seconds() + if age < 0 or (not allow_saved_shadow and age > 7 * 86400): + raise ValueError("watcher_task_stale") + matches.append(task) + if len(matches) != 1: + raise ValueError("single_verified_task_required") + return matches[0] + + +def _github_issue_get(repository: str, issue_number: int) -> dict[str, Any]: + if not os.environ.get("GH_TOKEN"): + raise ValueError("github_issue_token_unavailable") + completed = subprocess.run( + ["gh", "api", "--method", "GET", f"/repos/{repository}/issues/{issue_number}"], + check=True, capture_output=True, text=True, timeout=30, + ) + payload = _json(completed.stdout) + if not isinstance(payload, dict): + raise ValueError("github_issue_unavailable") + return payload + + +def _github_issue_close(repository: str, issue_number: int, body: str) -> dict[str, Any]: + if not os.environ.get("GH_TOKEN"): + raise ValueError("github_issue_token_unavailable") + completed = subprocess.run( + ["gh", "api", "--method", "PATCH", f"/repos/{repository}/issues/{issue_number}", + "-f", f"body={body}", "-f", "state=closed"], + check=True, capture_output=True, text=True, timeout=30, + ) + payload = _json(completed.stdout) + if not isinstance(payload, dict): + raise ValueError("github_issue_archive_unconfirmed") + return payload + + +def _issue_candidate(watcher: Mapping[str, Any], task: Mapping[str, Any], repository: str) -> dict[str, Any] | None: + """Join one current task to its watcher Issue without trusting task_id as identity.""" + issues = watcher.get("issues") + if not isinstance(issues, list): + return None + task_id = str(task.get("task_id") or "") + event_key = task_id.removeprefix("watcher-") if task_id.startswith("watcher-") else "" + candidates: list[dict[str, Any]] = [] + for raw in issues: + if not isinstance(raw, Mapping) or str(raw.get("repo") or "") != repository: + continue + summary = raw.get("task") + if not isinstance(summary, Mapping) or str(summary.get("event_key") or "") != event_key: + continue + key = str(raw.get("watcher_issue_key") or "") + url = str(raw.get("url") or raw.get("existing_url") or "") + match = _ISSUE_URL.fullmatch(url) + if key and _WATCHER_KEY.fullmatch(key) and match and match.group(1) == repository: + candidates.append({"key": key, "url": url, "issue_number": int(match.group(2))}) + return candidates[0] if len(candidates) == 1 else None + + +def _resolve_research_owner(watcher: Mapping[str, Any], task: Mapping[str, Any]) -> dict[str, Any] | None: + """Return a real source Issue owner only after a strict GitHub readback.""" + repository = os.environ.get("STRATEGY_WATCH_SOURCE_REPO", "").strip() + if repository != ISSUE_REPOSITORY: + return None + candidate = _issue_candidate(watcher, task, repository) + if candidate is None: + return None + try: + issue = _github_issue_get(repository, candidate["issue_number"]) + if issue.get("pull_request") is not None or str(issue.get("state") or "").upper() != "OPEN": + return None + issue_url = issue.get("html_url") + url_match = _ISSUE_URL.fullmatch(issue_url) + if (not url_match or url_match.group(1) != repository + or int(url_match.group(2)) != candidate["issue_number"] + or type(issue.get("number")) is not int + or issue.get("number") != candidate["issue_number"]): + return None + body = issue.get("body") + marker = f"" + if not isinstance(body, str) or marker not in body: + return None + except (OSError, ValueError, TypeError, KeyError, subprocess.CalledProcessError, subprocess.TimeoutExpired): + return None + return { + "repository": repository, + "issue_number": candidate["issue_number"], + "watcher_issue_key": candidate["key"], + } + + +_ARCHIVE_REASONS = frozenset({"no_improvement_limit", "idle_timeout"}) +_ARCHIVE_STATUSES = frozenset({"running", "unknown", "pending"}) +_ARCHIVE_PROTECTED_STATES = frozenset({ + "shadow", "shadow_recorded", "awaiting_human", "human_accepted", "human_rejected", + "pending", "running", "unknown", +}) + + +def _archive_marker(owner: Mapping[str, Any], scope_key: str, ticket_id: str, reason: str) -> str: + payload = { + "automation": "strategy_optimization_watcher", + "issue_number": owner["issue_number"], + "repository": owner["repository"], + "reason": reason, + "scope_key": scope_key, + "ticket_id": ticket_id, + "watcher_issue_key": owner["watcher_issue_key"], + } + return "" + + +def _archive_research_issue(runtime: Any, result: Mapping[str, Any], owner: Mapping[str, Any] | None) -> dict[str, Any]: + """Close exactly one verified watcher Issue for one locally archived scope.""" + base = {"status": "unavailable", "reason": "research_owner_unavailable"} + if owner is None: + return base + path_value = result.get("ticket_path") + cycle = getattr(runtime, "cycle", None) + if not isinstance(path_value, str) or cycle is None: + return {**base, "reason": "research_ticket_unavailable"} + try: + path = Path(path_value).resolve() + ticket_root = (STATE_ROOT / "research_promotion_tickets").resolve() + path.relative_to(ticket_root) + loader = cycle.load_research_promotion_ticket + saver = cycle.save_research_promotion_ticket + locker = cycle._research_directory_lock + except (AttributeError, OSError, ValueError, TypeError): + return {**base, "reason": "research_ticket_unavailable"} + try: + with locker(ticket_root) as acquired: + if not acquired: + return {**base, "status": "deferred", "reason": "research_in_progress"} + ticket = loader(path) + progress = dict(ticket.research_progress) + identity = progress.get("identity") + saved_owner = identity.get("owner") if isinstance(identity, Mapping) else None + scope_key = str(progress.get("scope_key") or "") + ticket_id = str(ticket.ticket_id or "") + lifecycle = progress.get("lifecycle") + archive_reason = str(lifecycle.get("archive_reason") or "") if isinstance(lifecycle, Mapping) else "" + stages = progress.get("stages") + protected = ( + not isinstance(identity, Mapping) or dict(saved_owner or {}) != dict(owner) + or not re.fullmatch(r"[0-9a-f]{64}", scope_key) + or not ticket_id or ticket.live_authority_granted is True + or not isinstance(lifecycle, Mapping) or lifecycle.get("archived") is not True + or lifecycle.get("paused") is True or archive_reason not in _ARCHIVE_REASONS + or str(ticket.state.value if hasattr(ticket.state, "value") else ticket.state) in + _ARCHIVE_PROTECTED_STATES + or not isinstance(stages, Mapping) + or any(isinstance(stage, Mapping) and stage.get("status") in _ARCHIVE_STATUSES + for stage in stages.values()) + ) + if protected: + return {**base, "reason": "research_ticket_not_archiveable"} + try: + scope_records = cycle._saved_scope_records(ticket_root, scope_key) + except (OSError, ValueError, TypeError, KeyError, AttributeError): + return {**base, "reason": "research_ticket_not_archiveable"} + if not isinstance(scope_records, list) or not any( + isinstance(record, tuple) and len(record) == 2 + and isinstance(record[1], type(ticket)) + and str(record[1].ticket_id or "") == ticket_id + for record in scope_records): + return {**base, "reason": "research_ticket_not_archiveable"} + for _, scope_ticket in scope_records: + scope_progress = scope_ticket.research_progress + scope_lifecycle = scope_progress.get("lifecycle") + scope_stages = scope_progress.get("stages") + scope_state = str( + scope_ticket.state.value if hasattr(scope_ticket.state, "value") else scope_ticket.state + ) + if ( + scope_ticket.live_authority_granted is True + or not isinstance(scope_lifecycle, Mapping) + or scope_lifecycle.get("archived") is not True + or scope_lifecycle.get("paused") is True + or str(scope_lifecycle.get("archive_reason") or "") not in _ARCHIVE_REASONS + or scope_state in _ARCHIVE_PROTECTED_STATES + or not isinstance(scope_stages, Mapping) + or any( + isinstance(stage, Mapping) and stage.get("status") in _ARCHIVE_STATUSES + for stage in scope_stages.values() + ) + ): + return {**base, "reason": "research_ticket_not_archiveable"} + marker = _archive_marker(owner, scope_key, ticket_id, archive_reason) + delivery = progress.get("issue_archive") + if isinstance(delivery, Mapping) and delivery.get("status") == "confirmed": + return {"status": "confirmed", "reason": "issue_already_archived"} + + issue = _github_issue_get(owner["repository"], owner["issue_number"]) + body = issue.get("body") + issue_url = issue.get("html_url") + original_marker = f"" + if (issue.get("pull_request") is not None or not isinstance(body, str) + or type(issue.get("number")) is not int or issue.get("number") != owner["issue_number"] + or not isinstance(issue_url, str) + or not (_ISSUE_URL.fullmatch(issue_url) + and _ISSUE_URL.fullmatch(issue_url).group(1) == owner["repository"] + and int(_ISSUE_URL.fullmatch(issue_url).group(2)) == owner["issue_number"]) + or original_marker not in body + or str(issue.get("state") or "").upper() not in {"OPEN", "CLOSED"}): + raise ValueError("github_issue_archive_unavailable") + if marker in body and str(issue.get("state") or "").upper() == "CLOSED": + progress["issue_archive"] = {"status": "confirmed", "marker": marker} + ticket.research_progress = progress + saver(ticket, path) + return {"status": "confirmed", "reason": "issue_already_archived"} + if isinstance(delivery, Mapping) and delivery.get("status") in {"unknown", "running"}: + progress["issue_archive"] = {"status": "unknown", "marker": marker} + ticket.research_progress = progress + saver(ticket, path) + return {**base, "reason": "issue_archive_unknown"} + + progress["issue_archive"] = {"status": "running", "marker": marker} + ticket.research_progress = progress + saver(ticket, path) + updated_body = body.rstrip() + "\n\n" + marker + response = _github_issue_close(owner["repository"], owner["issue_number"], updated_body) + if str(response.get("state") or "").upper() != "CLOSED" or marker not in str(response.get("body") or ""): + raise ValueError("github_issue_archive_unconfirmed") + progress["issue_archive"] = {"status": "confirmed", "marker": marker} + ticket.research_progress = progress + saver(ticket, path) + return {"status": "confirmed", "reason": "issue_archived"} + except (OSError, ValueError, TypeError, KeyError, subprocess.CalledProcessError, subprocess.TimeoutExpired): + try: + with locker(ticket_root) as acquired: + if acquired: + ticket = loader(path) + progress = dict(ticket.research_progress) + progress["issue_archive"] = {"status": "unknown"} + ticket.research_progress = progress + saver(ticket, path) + except Exception: + pass + return {**base, "reason": "issue_archive_unknown"} + + +def _read_drift(binding: dict, now: datetime, *, allow_saved_shadow: bool = False) -> dict: + path = Path(binding["path"]) + if not path.is_absolute(): + raise ValueError("observation_source_required") + raw = _read_json(path) + score = raw.get("drift_score") + if not isinstance(raw.get("as_of"), str): + raise ValueError("observation_date_required") + as_of = date.fromisoformat(raw["as_of"]) + if (raw.get("strategy_profile") != PROFILE or raw.get("domain") != "cn_equity" + or raw.get("source_revision") != binding["source_revision"] + or not isinstance(raw.get("source_revision"), str) or not raw["source_revision"].strip() + or raw.get("status") not in {"review", "critical"} + or type(score) not in (float, int) or not math.isfinite(score) or not 0 <= score <= 1 + or raw.get("alert_suppressed") is True or raw.get("baseline_available") is False + or as_of.isoformat() != raw["as_of"] or (now.date() - as_of).days < 0 + or (not allow_saved_shadow and (now.date() - as_of).days > 7)): + raise ValueError("observation_unavailable") + return {"strategy_profile": PROFILE, "domain": "cn_equity", "as_of": raw["as_of"], + "drift_score": score, "status": raw["status"], "source_revision": raw["source_revision"]} + + +def admit_one_new_experiment(ticket_dir: Path, created_at: str) -> bool: + """Called inside QPK's directory lock; no second lock, ledger or retry.""" + try: + current = _timestamp(created_at) + count = 0 + for path in ticket_dir.glob("*.json"): + ticket = _read_json(path) + if (not re.fullmatch(r"rpt_[0-9a-f]{64}", path.stem) or ticket["ticket_id"] != path.stem + or ticket["strategy_profile"] != PROFILE or ticket["domain"] != "cn_equity" + or ticket["live_authority_granted"] is not False): + return False + timestamp = _timestamp(ticket["created_at"]) + if timestamp > current: + return False + count += timestamp.date() == current.date() + return count == 0 + except (OSError, ValueError, TypeError, KeyError): + return False + + +def _installed_revision(distribution: str, expected: str) -> None: + if not isinstance(expected, str) or not _REVISION.fullmatch(expected): + raise ValueError("installed_revision_unavailable") + raw = importlib.metadata.distribution(distribution).read_text("direct_url.json") + source = _json(raw or "{}") + if (not isinstance(source, dict) or source.get("dir_info") or source.get("archive_info") + or source.get("vcs_info", {}).get("vcs") != "git" + or source.get("vcs_info", {}).get("commit_id") != expected): + raise ValueError("installed_revision_mismatch") + + +def _load_runtime(policy: dict) -> SimpleNamespace: + for package, revision in (("cn-equity-strategies", "code_revision"), + ("quant-platform-kit", "qpk_revision")): + _installed_revision(package, policy[revision]) + if importlib.metadata.version("personal-ai-service") != "2.0.0": + raise ValueError("task_client_version_mismatch") + from quant_platform_kit.strategy_lifecycle.ai_provider import AiServiceClient, AiServiceConfig, AiProviderConfig + from cn_equity_strategies.backtest import index_etf_research_job as cn + from cn_equity_strategies.backtest.index_etf_strict_runner import IndexEtfExecutionConfig, read_index_etf_input + from quant_platform_kit.strategy_lifecycle.codex_integration import AiOptimizationContext, build_optimization_prompt + from quant_platform_kit.strategy_lifecycle.contracts import DriftResult, DriftStatus, PromotionCostModel, PurgedWalkForwardFold + from quant_platform_kit.strategy_lifecycle.production_drift_health_probe import probe_production_drift_health + import quant_platform_kit.strategy_lifecycle.research_promotion_cycle as research_cycle + return SimpleNamespace(cn=cn, client=AiServiceClient, config=AiServiceConfig, route=AiProviderConfig, + read_input=read_index_etf_input, execution_config=IndexEtfExecutionConfig, cost_model=PromotionCostModel, + fold=PurgedWalkForwardFold, drift=DriftResult, status=DriftStatus, context=AiOptimizationContext, + prompt=build_optimization_prompt, probe=probe_production_drift_health, cycle=research_cycle) + + +def _diagnosis(runtime, drift, revision): + client = runtime.client(runtime.config.reliability(primary=runtime.route.from_env())) + context = runtime.context(strategy_profile=PROFILE, domain="cn_equity", drift=drift, + current_params=runtime.cn.BASELINE_PARAMS) + prompt = runtime.prompt(context) + key = "cn-research:" + hashlib.sha256((revision + "\n" + prompt).encode()).hexdigest() + + def diagnose(*_, resume_task_id=None): + options = {"timeout": 600, "idempotency_key": key} + if resume_task_id is not None: + options["resume_task_id"] = resume_task_id + result = client.execute(prompt, **options) + raw = result.raw + if not result.success and isinstance(raw, dict) and raw.get("status") in { + "queued", "submitting", "running", "cancel_requested", "outcome_unknown", + }: + if not isinstance(raw.get("id"), str) or not raw["id"]: + raise ValueError("research_task_identity_missing") + return {"optimization_needed": False, "reason": "ai_task_pending", "task_id": raw["id"]} + if (result.success is not True or not isinstance(raw, dict) + or raw.get("status") != "completed" or raw.get("result_kind") != "advisory" + or not isinstance(raw.get("id"), str) or not raw["id"] + or raw.get("model_verification") not in {"unavailable", "provider_reported"}): + raise ValueError("research_model_outcome_unavailable") + decision = _json(result.output) + if (not isinstance(decision, dict) or type(decision.get("optimization_needed")) is not bool + or (decision["optimization_needed"] and decision.get("recommended_method") != "grid_search")): + raise ValueError("research_model_decision_invalid") + return {"optimization_needed": decision["optimization_needed"], "recommended_method": "grid_search", + "reason": "ai_research_decision", "provider": result.provider, + "model_requested": raw.get("model_requested"), + "model_verification": raw["model_verification"], "task_id": raw["id"]} + + def read_pending(pending): + task_id = pending.get("task_id") if isinstance(pending, Mapping) else None + if not isinstance(task_id, str) or not task_id: + raise ValueError("research_task_identity_missing") + return diagnose(resume_task_id=task_id) + + diagnose.read_pending = read_pending + return diagnose + + +_SUMMARY_FIELDS = { + "identity", "strategy_description", "plugins", "comparison", "limitations", +} +_SUMMARY_IDENTITY_FIELDS = {"strategy_profile", "domain", "proposed_params"} +_SUMMARY_COMPARISON_FIELDS = {"status", "baseline", "candidate", "start_date", "end_date", "cost_model"} + + +def _json_data(value: Any) -> bool: + try: + json.dumps(value, ensure_ascii=False, allow_nan=False) + except (TypeError, ValueError, OverflowError): + return False + return True + + +def _validate_summary_context(value: Any) -> dict[str, Any]: + if not isinstance(value, dict) or set(value) != _SUMMARY_FIELDS: + raise ValueError("summary_context_fields_invalid") + identity = value["identity"] + if (not isinstance(identity, dict) or set(identity) != _SUMMARY_IDENTITY_FIELDS + or not isinstance(identity["strategy_profile"], str) + or not isinstance(identity["domain"], str) + or not isinstance(identity["proposed_params"], dict)): + raise ValueError("summary_context_identity_invalid") + comparison = value["comparison"] + if not isinstance(comparison, dict) or set(comparison) != _SUMMARY_COMPARISON_FIELDS: + raise ValueError("summary_context_comparison_invalid") + if value["plugins"] is not None and not isinstance(value["plugins"], list): + raise ValueError("summary_context_plugins_invalid") + if (value["limitations"] is not None + and (not isinstance(value["limitations"], list) + or any(not isinstance(item, str) for item in value["limitations"]))): + raise ValueError("summary_context_limitations_invalid") + if not isinstance(value["strategy_description"], str): + raise ValueError("summary_context_text_invalid") + if not _json_data(value): + raise ValueError("summary_context_data_invalid") + return value + + +def _summary_callback(runtime, revision): + from quant_platform_kit.strategy_lifecycle.research_summary import make_summary_callback + return make_summary_callback( + revision=revision, repository=STRATEGY_REPOSITORY, + local_facts={ + "strategy_description": "CN momentum and trend selection with risk reduction and volatility weighting.", + "plugins": ["CN index ETF tactical rotation"], + "limitations": ["research_only", "no order authority"], + }, validate_context=_validate_summary_context, + ) + + +def _summary(status: str, reason: str, **extra) -> dict: + return {"status": status, "reason": reason, "no_order": True, "size_zero_required": True, + "live_authority_granted": False, **extra} + + +def _make_shadow_reader(binding: dict, identity: dict, revision: str): + """Read producer evidence only; never create receipts or start observation.""" + from quant_platform_kit.strategy_lifecycle.forward_observation import ForwardObservationPolicy + from quant_platform_kit.strategy_lifecycle.paired_shadow_adapter import collect_paired_shadow_for_promotion + + if not isinstance(binding, dict) or not binding.get("observation_path") or not binding.get("forward_policy"): + raise ValueError("shadow_provider_not_configured") + interval = binding["retry_after_seconds"] + if type(interval) is not int or not 60 <= interval <= 86400: + raise ValueError("shadow_read_interval_invalid") + + policy = ForwardObservationPolicy(**binding["forward_policy"]) + if (policy.strategy_profile != PROFILE or policy.domain != "cn_equity" + or policy.observation_calendar != "XSHG" or policy.observation_window_type != "fixed" + or tuple(policy.automatic_non_live_modes) != ("shadow",) + or tuple(policy.non_live_evidence_modes) != ("shadow_decision",)): + raise ValueError("shadow_policy_invalid") + calendar_path = Path(binding["calendar_path"]) + calendar_bytes = _read_bytes(calendar_path) + calendar = _json(calendar_bytes) + if hashlib.sha256(calendar_bytes).hexdigest() != binding["calendar_sha256"]: + raise ValueError("shadow_calendar_mismatch") + if not isinstance(calendar, list) or calendar != sorted(set(calendar)): + raise ValueError("shadow_calendar_invalid") + sessions = [date.fromisoformat(day).isoformat() for day in calendar] + start = sessions.index(policy.observation_start_session) + expected = sessions[start:start + policy.required_trading_sessions] + if len(expected) != policy.required_trading_sessions: + raise ValueError("shadow_calendar_incomplete") + path = Path(binding["observation_path"]) + if not path.is_absolute() or path.is_symlink(): + raise ValueError("shadow_source_invalid") + frozen = binding["frozen_dependency_digests"] + if (not isinstance(frozen, dict) + or set(frozen) != {"p2_config", "p3_evidence", "risk_policy", "strategy_release", "plugin_bundle"} + or any(not isinstance(value, str) or not re.fullmatch(r"[0-9a-f]{64}", value) for value in frozen.values()) + or not isinstance(binding["baseline_id"], str) or not binding["baseline_id"].strip() + or binding["baseline_id"] == policy.candidate_id): + raise ValueError("shadow_source_dependencies_missing") + + def read(proposal): + guard = {"passed": False, "no_order": True, "live_authority_granted": False} + pending = {**guard, "status": "pending", "retry_at": _now().timestamp() + interval} + try: + # The existing CN next-open contract freezes decisions before 09:25. + # A candidate created later cannot count that session as forward. + if _timestamp(proposal.computed_at) >= _timestamp(expected[0] + "T09:25:00+08:00"): + raise ValueError("shadow_candidate_created_after_window_start") + if not path.exists(): + return pending + payload = _read_json(path) + if (payload["strategy_profile"] != PROFILE or payload["domain"] != "cn_equity" + or proposal.strategy_profile != PROFILE or proposal.domain != "cn_equity" + or payload["source_revision"] != revision or payload["research_identity"] != identity + or payload["current_params"] != dict(proposal.current_params) + or payload["proposed_params"] != dict(proposal.proposed_params)): + raise ValueError("shadow_candidate_mismatch") + observations = payload["observations"] + if not isinstance(observations, list) or len(observations) > len(expected): + raise ValueError("shadow_observation_count_invalid") + previous_evidence = previous_receipt = None + for index, raw in enumerate(observations, 1): + receipt = raw["forward_observation_receipt"] + if (receipt["observation_index"] != index or receipt["observation_session"] != expected[index - 1] + or raw["baseline_id"] != binding["baseline_id"] + or raw["input_snapshot_sha256"] != receipt["dependency_digests"]["p1_manifest"] + or any(receipt["dependency_digests"].get(key) != value for key, value in frozen.items()) + or not _timestamp(proposal.computed_at) <= _timestamp(raw["observed_at"]) <= _now() + or _timestamp(raw["observed_at"]).astimezone(ZoneInfo("Asia/Shanghai")).date().isoformat() != expected[index - 1] + or _timestamp(raw["observed_at"]) < _timestamp(expected[index - 1] + "T15:00:00+08:00")): + raise ValueError("shadow_observation_binding_invalid") + observation = {"policy": policy, **{key: raw[key] for key in ( + "forward_observation_receipt", "baseline_id", "observed_at", "input_snapshot_sha256", "candidate", "baseline")}, + "previous_evidence": previous_evidence, "previous_forward_observation_receipt": previous_receipt} + record = collect_paired_shadow_for_promotion(observation) + previous_evidence, previous_receipt = record["evidence"], receipt + if len(observations) != policy.required_trading_sessions: + return pending + # QPK repeats the canonical paired validation before saving complete. + return {"status": "complete", "observation": observation} + except (ValueError, TypeError, KeyError, IndexError): + return {**guard, "status": "failed", "reason": "shadow_observation_invalid"} + except OSError: + # An unreadable existing source is unknown, never a fresh attempt. + raise ValueError("shadow_observation_unavailable") from None + return read + + +def _console_bindings(binding: dict | None): + from quant_platform_kit.strategy_lifecycle.research_promotion_cycle import ( + make_console_research_promotion_pull, make_console_research_promotion_sync, + ) + if not binding: + raise ValueError("console_not_configured") + from urllib.parse import urlsplit + endpoints = [urlsplit(binding[key]) for key in ("sync_url", "pull_url")] + if (any(url.scheme != "https" or not url.hostname or url.username or url.password or url.query or url.fragment + for url in endpoints) or endpoints[0].netloc != endpoints[1].netloc): + raise ValueError("console_url_invalid") + token_path = Path(binding["token_path"]) + _protected_file(token_path, secret=True) + if token_path.stat().st_size > 4096: + raise ValueError("console_token_unavailable") + token = token_path.read_text().strip() + if not token or "\n" in token or "\r" in token: + raise ValueError("console_token_unavailable") + pull = make_console_research_promotion_pull(endpoint_url=binding["pull_url"], sync_token=token, + printer=lambda *_, **__: None, raise_on_unavailable=True) + sync = make_console_research_promotion_sync(endpoint_url=binding["sync_url"], sync_token=token, + pull_console=pull, printer=lambda *_, **__: None) + return sync, pull + + +def run_from_watcher(watcher: dict, *, policy_path: Path = POLICY_PATH, dry_run: bool = False) -> dict: + try: + policy = _read_policy(policy_path) + except FileNotFoundError: + return _summary("parked", "cn_research_not_configured") + except (ValueError, OSError): + return _summary("parked", "cn_research_policy_invalid") + if policy["enabled"] is not True: + return _summary("parked", "cn_research_disabled") + if not _workflow_authenticated(): + return _summary("parked", "cn_research_workflow_auth_required") + try: + # Old observations are passed unchanged only so QPK can locate a saved + # shadow checkpoint. QPK still forbids creating/restarting stale research. + task = _select_task(watcher, policy["code_revision"], _now(), allow_saved_shadow=True) + research_owner = _resolve_research_owner(watcher, task) + source_repository = os.environ.get("STRATEGY_WATCH_SOURCE_REPO", "").strip() + task_event_key = str(task.get("task_id") or "").removeprefix("watcher-") + declared_owner_binding = bool(source_repository) or ( + isinstance(watcher.get("issues"), list) + and any( + isinstance(item, Mapping) + and isinstance(item.get("task"), Mapping) + and str(item["task"].get("event_key") or "") == task_event_key + for item in watcher["issues"] + ) + ) + if declared_owner_binding and research_owner is None: + return _summary("parked", "cn_research_issue_owner_unavailable", task_id=task["task_id"]) + drift = _read_drift(policy["drift"], _now(), allow_saved_shadow=True) + runtime = _load_runtime(policy) + if policy["candidate_id"] != PROFILE or policy["domain"] != "cn_equity": + raise ValueError("policy_candidate_invalid") + # The shared evaluator owns thresholds. Never derive score/date from a task. + health = runtime.probe(strategy_profile=PROFILE, domain="cn_equity", as_of=drift["as_of"], + drift_score=drift["drift_score"]) + saved_shadow_only = health.get("reason") == "observation_stale" + if ((not saved_shadow_only and health["actionable"] is not True) + or (health.get("risk_status") if saved_shadow_only else health["status"]) != drift["status"]): + raise ValueError("observation_not_actionable") + inputs = {name: runtime.read_input(Path(policy["inputs"][name]["path"]), + expected_manifest_sha256=policy["inputs"][name]["manifest_sha256"]) + for name in ("development", "validation")} + plan = policy["plan"] + arguments = dict(development_input=inputs["development"], validation_input=inputs["validation"], + trusted_input_roots={name: policy["inputs"][name]["manifest_sha256"] for name in inputs}, + development_start=date.fromisoformat(plan["development_start"]), + development_end=date.fromisoformat(plan["development_end"]), + folds=tuple(runtime.fold(**{key: date.fromisoformat(value) for key, value in fold.items()}) for fold in plan["folds"]), + locked_oos_start=date.fromisoformat(plan["locked_oos_start"]), locked_oos_end=date.fromisoformat(plan["locked_oos_end"]), + purge_days=plan["purge_days"], embargo_days=plan["embargo_days"], code_revision=policy["code_revision"], + config=runtime.execution_config(**policy["execution_config"]), cost_model=runtime.cost_model(**policy["cost_model"])) + identity = runtime.cn.preflight_index_etf_research_job(**arguments) + if set(identity) != _IDENTITY_FIELDS or identity != policy["research_identity"]: + raise ValueError("frozen_research_identity_mismatch") + if dry_run: + return _summary("dry_run", "validated_without_execution", task_id=task["task_id"]) + shadow = _make_shadow_reader(policy["shadow"], identity, policy["code_revision"]) + sync, pull = _console_bindings(policy["console"]) + active_drift = runtime.drift(strategy_profile=PROFILE, domain="cn_equity", + as_of=date.fromisoformat(drift["as_of"]), status=runtime.status(drift["status"]), + drift_score=drift["drift_score"], source_revision=drift["source_revision"]) + model_diagnose = _diagnosis(runtime, active_drift, policy["code_revision"]) + summarize = _summary_callback(runtime, policy["code_revision"]) + window_start = _timestamp(policy["shadow"]["forward_policy"]["observation_start_session"] + "T09:25:00+08:00") + + def diagnose(*args): + # A quota-deferred ticket may resume after the observation window + # became impossible. Completed stages never call this again. + if _now() >= window_start: + return {"optimization_needed": False, "reason": "forward_window_start_elapsed"} + return model_diagnose(*args) + + def admit_new(ticket_dir: Path, created_at: str) -> bool: + # QPK calls this only for a new ticket, under its existing lock. + # An existing pending/awaiting ticket keeps its original identity. + return _timestamp(created_at) < window_start and admit_one_new_experiment(ticket_dir, created_at) + + result = runtime.cn.run_index_etf_research_job(**arguments, + ticket_dir=STATE_ROOT / "research_promotion_tickets", store_root=STATE_ROOT, + as_of=drift["as_of"], drift_score=drift["drift_score"], source_revision=drift["source_revision"], + record_shadow=shadow, read_pending_shadow=shadow, sync_console=sync, pull_console=pull, + diagnose=diagnose, read_pending_diagnosis=model_diagnose.read_pending, + summarize=summarize, admit_new_research=admit_new, + research_owner=research_owner) + issue_archive = ( + _archive_research_issue(runtime, result, research_owner) + if result.get("reason") == "research_scope_archived" + else {"status": "unavailable", "reason": "research_not_archived"} + ) + return _summary(result["status"], result["reason"], task_id=task["task_id"], + observation_as_of=drift["as_of"], observation_source_revision=drift["source_revision"], + **({"research_owner": research_owner} if research_owner is not None else {}), + issue_archive=issue_archive, + **{key: result[key] for key in ("research_key", "resumed", "console_synced", "retry_at") if key in result}) + except Exception: + return _summary("parked", "cn_research_preflight_unavailable") + + +def main(argv=None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--watcher-result", required=True) + parser.add_argument("--dry-run", action="store_true") + args = parser.parse_args(argv) + try: + result = run_from_watcher(_read_json(Path(args.watcher_result)), dry_run=args.dry_run) + except Exception: + result = _summary("parked", "cn_research_watcher_unavailable") + print(json.dumps(result, sort_keys=True, allow_nan=False)) + return 0 if result["status"] == "dry_run" or "research_key" in result else 3 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_drift_workflow_config.py b/tests/test_drift_workflow_config.py index 747f9a1..e7b3f4a 100644 --- a/tests/test_drift_workflow_config.py +++ b/tests/test_drift_workflow_config.py @@ -30,13 +30,13 @@ def test_drift_workflow_wires_real_snapshot_history_and_preflight_bundle() -> No assert "Upload lifecycle preflight artifact" in workflow assert "actions/upload-artifact@v7" in workflow assert workflow.count("github.ref == format('refs/heads/{0}', github.event.repository.default_branch)") == 2 - assert "uses: QuantStrategyLab/QuantPlatformKit/.github/workflows/reusable-drift-check.yml@c7646a7168b3dafa763ef7751a182d23e8de7790" in workflow + assert "uses: ./.github/workflows/reusable-drift-check.yml" in workflow assert "strategy_domain: cn_equity" in workflow assert "caller_event_name: ${{ github.event_name }}" in workflow assert "caller_pr_head_repository: ${{ github.event.pull_request.head.repo.full_name || '' }}" in workflow assert "snapshot_repository: QuantStrategyLab/CnEquitySnapshotPipelines" in workflow assert "snapshot_checkout_path: external/CnEquitySnapshotPipelines" in workflow - assert "ai_gateway_service_url: ${{ vars.AI_GATEWAY_SERVICE_URL }}" in workflow + assert "ai_service_url: ${{ vars.AI_SERVICE_URL }}" in workflow assert "lifecycle_preflight_artifact: lifecycle-preflight-${{ github.run_id }}-${{ github.run_attempt }}" in workflow - assert "codex_audit_service_url: ${{ secrets.CODEX_AUDIT_SERVICE_URL }}" in workflow + assert "codex_audit_service_url:" not in workflow assert "secrets.SNAPSHOT_REPOSITORY_TOKEN || secrets.QSL_REPO_SYNC_TOKEN || github.token" in workflow diff --git a/tests/test_gate_evidence_package.py b/tests/test_gate_evidence_package.py index 38a2ea1..751b52c 100644 --- a/tests/test_gate_evidence_package.py +++ b/tests/test_gate_evidence_package.py @@ -1,66 +1,72 @@ -from __future__ import annotations - -import contextlib -import io -import os -import tempfile -import unittest -from pathlib import Path +"""Promotion keeps a three-role advisory quorum and blocks missing results.""" +import json from unittest.mock import patch - +import pytest from scripts.gate_evidence_package import _run_promotion_dual_review -class PromotionReviewExitTests(unittest.TestCase): - def setUp(self) -> None: - self.temp = tempfile.TemporaryDirectory() - self.addCleanup(self.temp.cleanup) - self.root = Path(self.temp.name) - scripts = self.root / "scripts" - scripts.mkdir() - self.script = scripts / "run_dual_review_pipeline.py" - self.script.write_text( - "import os, signal, sys\n" - "from pathlib import Path\n" - "code = Path(sys.argv[sys.argv.index('--from-evidence') + 1]).stem\n" - "with Path('calls').open('a') as log: log.write(code + '\\n')\n" - "if code == 'signal': os.kill(os.getpid(), signal.SIGTERM)\n" - "raise SystemExit(int(code))\n", - encoding="utf-8", - ) - - def review(self, codes: tuple[str, ...], *, skip: str = "") -> int: - # This is a real local Python child process, with synthetic exit behavior - # only. It never imports AIAuditBridge, reads evidence or calls a model. - with patch.dict(os.environ, {"AIAUDIT_BRIDGE_ROOT": str(self.root), "DUAL_REVIEW_GATE_SKIP": skip}): - with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()): - return _run_promotion_dual_review([self.root / f"{code}.json" for code in codes]) - - def test_every_nonzero_exit_blocks_promotion(self) -> None: - for code in ("1", "2", "3", "127", "signal"): - with self.subTest(code=code): - self.assertEqual(self.review((code,)), 1) - - def test_failure_remains_blocking_after_later_success(self) -> None: - for failure in ("1", "2", "signal"): - with self.subTest(failure=failure): - (self.root / "calls").write_text("", encoding="utf-8") - self.assertEqual(self.review((failure, "0")), 1) - self.assertEqual((self.root / "calls").read_text().splitlines(), [failure, "0"]) - - def test_all_successful_reviews_pass(self) -> None: - self.assertEqual(self.review(("0", "0")), 0) - self.assertEqual((self.root / "calls").read_text().splitlines(), ["0", "0"]) - - def test_explicit_skip_does_not_run_review(self) -> None: - self.assertEqual(self.review(("1",), skip="true"), 0) - self.assertFalse((self.root / "calls").exists()) - - def test_missing_optional_script_keeps_existing_skip(self) -> None: - self.script.unlink() - self.assertEqual(self.review(("1",)), 0) - self.assertFalse((self.root / "calls").exists()) - - -if __name__ == "__main__": - unittest.main() +@pytest.fixture +def evidence(tmp_path, monkeypatch): + monkeypatch.setenv("GITHUB_SHA", "a" * 40) + monkeypatch.delenv("DUAL_REVIEW_GATE_SKIP", raising=False) + path = tmp_path / "evidence.json" + path.write_text(json.dumps({"synthetic": True})) + return path + + +@pytest.mark.parametrize("status", ["pending", "unavailable", "failed", "outcome_unknown"]) +def test_noncompleted_quorum_blocks_promotion(evidence, status): + with patch("quant_platform_kit.strategy_lifecycle.task_review.review_material", return_value={"status": status}): + assert _run_promotion_dual_review([evidence]) != 0 + + +@pytest.mark.parametrize("outcome", ["agree_reject", "requires_human", "unknown"]) +def test_rejection_or_disagreement_blocks(evidence, outcome): + with patch("quant_platform_kit.strategy_lifecycle.task_review.review_material", return_value={ + "status": "completed", "outcome": outcome, "advisory_only": True, + }): + assert _run_promotion_dual_review([evidence]) != 0 + + +def test_three_roles_and_caller_bound_material(evidence): + with patch("quant_platform_kit.strategy_lifecycle.task_review.review_material", return_value={ + "status": "completed", "outcome": "agree_approve", "advisory_only": True, + }) as review: + assert _run_promotion_dual_review([evidence]) == 0 + assert review.call_args.args[0] == {"synthetic": True} + assert len(review.call_args.kwargs["required_roles"]) == 3 + assert "a" * 40 in review.call_args.kwargs["operation_id"] + + +def test_failed_file_cannot_be_overwritten_by_later_success(evidence): + with patch("quant_platform_kit.strategy_lifecycle.task_review.review_material", side_effect=[ + {"status": "completed", "outcome": "agree_reject", "advisory_only": True}, + {"status": "completed", "outcome": "agree_approve", "advisory_only": True}, + ]) as review: + assert _run_promotion_dual_review([evidence, evidence]) != 0 + assert review.call_count == 1 + + +def test_explicit_existing_skip_makes_no_task(evidence, monkeypatch): + monkeypatch.setenv("DUAL_REVIEW_GATE_SKIP", "true") + with patch("quant_platform_kit.strategy_lifecycle.task_review.review_material") as review: + assert _run_promotion_dual_review([evidence]) == 0 + review.assert_not_called() + + +def test_missing_evidence_and_private_error_fail_without_leaking(evidence, capsys): + with patch("quant_platform_kit.strategy_lifecycle.task_review.review_material", side_effect=RuntimeError("private credential")): + assert _run_promotion_dual_review([evidence]) != 0 + evidence.unlink() + assert _run_promotion_dual_review([evidence]) != 0 + assert "private credential" not in capsys.readouterr().err + + +def test_toml_evidence_is_supported(evidence): + path = evidence.with_suffix(".toml") + path.write_text("synthetic = true\n") + with patch("quant_platform_kit.strategy_lifecycle.task_review.review_material", return_value={ + "status": "completed", "outcome": "agree_approve", "advisory_only": True, + }) as review: + assert _run_promotion_dual_review([path]) == 0 + assert review.call_args.args[0] == {"synthetic": True} diff --git a/tests/test_index_etf_research_task_v2.py b/tests/test_index_etf_research_task_v2.py new file mode 100644 index 0000000..a27c7e6 --- /dev/null +++ b/tests/test_index_etf_research_task_v2.py @@ -0,0 +1,554 @@ +"""Offline job admission; synthetic fixtures are not market evidence.""" + + +from __future__ import annotations + + +from datetime import datetime, timezone + + +import json + + +import os + + +from pathlib import Path + + +from unittest.mock import Mock, patch + + +from types import SimpleNamespace + + +import hashlib + + +import importlib + + +import io + + +import urllib.error + + +import subprocess + + +import sys + + +import pytest + + +from cn_equity_strategies.backtest import index_etf_research_task as job + + +from quant_platform_kit.strategy_lifecycle.research_task import build_strategy_diagnosis_task + + +NOW = datetime(2026, 9, 9, 8, tzinfo=timezone.utc) + + +class FixedClock(datetime): + @classmethod + def now(cls, tz=None): + return NOW.astimezone(tz) + + +REVISION = "c" * 40 + + +ENV = { + "GITHUB_REPOSITORY": "QuantStrategyLab/CnEquityStrategies", + "GITHUB_REF": "refs/heads/main", "GITHUB_EVENT_NAME": "workflow_dispatch", + "GITHUB_WORKFLOW_REF": "QuantStrategyLab/CnEquityStrategies/.github/workflows/cn-index-etf-research.yml@refs/heads/main", + "GITHUB_RUN_ID": "12345", "GITHUB_SHA": "a" * 40, + "ACTIONS_ID_TOKEN_REQUEST_URL": "https://synthetic.invalid/oidc", + "ACTIONS_ID_TOKEN_REQUEST_TOKEN": "synthetic-test-only", + "AI_SERVICE_URL": "https://synthetic.invalid", +} + + +def write(path, value): + path.write_text(json.dumps(value)) + return path + + +@pytest.fixture(autouse=True) +def isolated(): + with patch.dict(os.environ, ENV, clear=True), patch.object(job, "_now", return_value=NOW): + yield + + +def task(revision=REVISION): + return build_strategy_diagnosis_task( + event_key="a" * 12, created_at="2026-09-09T06:17:00Z", candidate_id=job.PROFILE, + candidate_kind="individual", domain="cn_equity", strategy_repository=job.STRATEGY_REPOSITORY, + evidence={"p1_input_digest": "1" * 64, "p2_config_digest": "2" * 64, + "p3_evidence_id": "3" * 64, "strategy_revision": revision, "producer_revision": "d" * 40}, + ) + + +def watcher(revision=REVISION): + return {"research_task_source_snapshot": {"schema_version": "qsl_research_task_source_snapshot.v1", + "data_status": "ready", "tasks": [task(revision)]}} + + +def test_missing_or_disabled_policy_never_imports_runner_or_calls_model(tmp_path): + with patch.object(job, "_load_runtime") as runtime: + result = job.run_from_watcher(watcher(), policy_path=tmp_path / "absent") + assert result["reason"] == "cn_research_not_configured" + with patch.object(job, "_read_policy", return_value={"enabled": False}): + result = job.run_from_watcher(watcher()) + assert result["reason"] == "cn_research_disabled" + runtime.assert_not_called() + + +@pytest.mark.parametrize("change", [{"AI_SERVICE_TOKEN": "synthetic-static"}, + {"ACTIONS_ID_TOKEN_REQUEST_URL": ""}, {"ACTIONS_ID_TOKEN_REQUEST_TOKEN": ""}, + {"GITHUB_REF": "refs/heads/feature"}, {"GITHUB_EVENT_NAME": "pull_request"}, + {"GITHUB_WORKFLOW_REF": "QuantStrategyLab/CnEquityStrategies/.github/workflows/unapproved.yml@refs/heads/main"}]) +def test_workflow_and_oidc_are_required_before_runtime(change): + with patch.object(job, "_read_policy", return_value={"enabled": True}), \ + patch.dict(os.environ, change), patch.object(job, "_load_runtime") as runtime: + assert job.run_from_watcher(watcher())["reason"] == "cn_research_workflow_auth_required" + runtime.assert_not_called() + + +def test_task_is_verified_and_never_supplies_the_observation_clock(): + verified = job._select_task(watcher(), REVISION, NOW) + assert verified["created_at"] == "2026-09-09T06:17:00Z" + invalid = watcher() + invalid["research_task_source_snapshot"]["tasks"][0]["experiment"]["max_runs"] = 2 + with pytest.raises(ValueError): + job._select_task(invalid, REVISION, NOW) + with pytest.raises(ValueError): + job._select_task(watcher(), "e" * 40, NOW) + + +@pytest.mark.parametrize("change", [{"as_of": None}, {"as_of": "2026-09-10"}, {"as_of": "2026-09-01"}, + {"drift_score": True}, {"drift_score": float("nan")}, {"status": "healthy"}, + {"source_revision": "wrong"}, {"strategy_profile": "another"}, {"domain": "us_equity"}, + {"alert_suppressed": True}, {"baseline_available": False}]) +def test_real_drift_requires_matching_fresh_actionable_source(tmp_path, change): + raw = {"strategy_profile": job.PROFILE, "domain": "cn_equity", "source_revision": REVISION, + "as_of": "2026-09-09", "drift_score": .8, "status": "critical", **change} + source = write(tmp_path / "drift.json", raw) + with pytest.raises(ValueError): + job._read_drift({"path": str(source), "source_revision": REVISION}, NOW) + + +def test_daily_admission_counts_existing_ticket_utc_created_at(tmp_path): + assert job.admit_one_new_experiment(tmp_path, NOW.isoformat()) is True + ticket = {"strategy_profile": job.PROFILE, "domain": "cn_equity", "live_authority_granted": False} + old_id, new_id = "rpt_" + "a" * 64, "rpt_" + "b" * 64 + write(tmp_path / f"{old_id}.json", {**ticket, "ticket_id": old_id, "created_at": "2026-09-08T23:59:59Z"}) + assert job.admit_one_new_experiment(tmp_path, NOW.isoformat()) is True + write(tmp_path / f"{new_id}.json", {**ticket, "ticket_id": new_id, "created_at": "2026-09-09T00:00:00Z"}) + assert job.admit_one_new_experiment(tmp_path, NOW.isoformat()) is False + + +@pytest.mark.parametrize("bad", [{}, {"created_at": "2026-09-09T00:00:00"}, + {"created_at": "2026-09-10T00:00:00Z"}, {"created_at": None}]) +def test_daily_admission_unknown_count_fails_closed(tmp_path, bad): + write(tmp_path / "rpt_bad.json", bad) + assert job.admit_one_new_experiment(tmp_path, NOW.isoformat()) is False + + +def test_daily_admission_does_not_add_a_ledger_or_follow_symlinks(tmp_path): + outside = write(tmp_path / "private", {"created_at": NOW.isoformat()}) + (tmp_path / "rpt_link.json").symlink_to(outside) + assert job.admit_one_new_experiment(tmp_path, NOW.isoformat()) is False + assert sorted(path.name for path in tmp_path.iterdir()) == ["private", "rpt_link.json"] + + +def test_policy_rejects_untrusted_owner_mode_and_symlink(tmp_path): + path = write(tmp_path / "policy.json", {"enabled": True}) + path.chmod(0o666) + with pytest.raises(ValueError): + job._read_policy(path) + link = tmp_path / "link.json" + link.symlink_to(path) + with pytest.raises(ValueError): + job._read_policy(link) + + +@pytest.mark.parametrize("source", [None, {}, {"dir_info": {"editable": True}}, + {"vcs_info": {"commit_id": "e" * 40}}, + {"vcs_info": {"vcs": "git", "commit_id": REVISION}, "dir_info": {"editable": True}}, + {"vcs_info": {"vcs": "svn", "commit_id": REVISION}}]) +def test_installed_runtime_requires_noneditable_git_commit(source): + raw = None if source is None else json.dumps(source) + with patch.object(job.importlib.metadata, "distribution") as distribution: + distribution.return_value.read_text.return_value = raw + with pytest.raises(ValueError, match="installed_revision"): + job._installed_revision("synthetic-package", REVISION) + + +def test_installed_runtime_accepts_exact_git_commit(): + with patch.object(job.importlib.metadata, "distribution") as distribution: + distribution.return_value.read_text.return_value = json.dumps({"vcs_info": {"vcs": "git", "commit_id": REVISION}}) + job._installed_revision("synthetic-package", REVISION) + + +def test_cli_failure_prints_only_sanitized_summary(tmp_path, capsys): + path = tmp_path / "sensitive-missing.json" + assert job.main(["--watcher-result", str(path)]) == 3 + output = capsys.readouterr().out + assert str(path) not in output and "sensitive" not in output + assert json.loads(output)["live_authority_granted"] is False + + +@pytest.fixture +def shadow_inputs(tmp_path): + importlib.import_module("quant_platform_kit") + from quant_platform_kit.strategy_lifecycle.forward_observation import ForwardObservationPolicy + from quant_platform_kit.strategy_lifecycle.forward_observation_receipt import build_forward_observation_receipt + policy_args = dict(candidate_id="cn-frozen-example", strategy_profile=job.PROFILE, domain="cn_equity", + benchmark_symbol="510300", required_trading_sessions=2, review_milestones=[1], + automatic_non_live_modes=["shadow"], auto_resume_clean_sessions=1, observation_calendar="XSHG", + observation_window_type="fixed", observation_start_session="2026-09-07", + window_rationale_ref="synthetic-test-policy", non_live_evidence_modes=["shadow_decision"]) + policy = ForwardObservationPolicy(**policy_args) + calendar = write(tmp_path / "sessions.json", ["2026-09-07", "2026-09-08"]) + dependencies = {key: "a" * 64 for key in ("p1_manifest", "p2_config", "p3_evidence", "risk_policy", "strategy_release", "plugin_bundle")} + observations = [] + previous = None + for index, day in enumerate(["2026-09-07", "2026-09-08"], 1): + receipt = build_forward_observation_receipt(policy=policy, observation_session=day, observation_index=index, + dependency_digests=dependencies, evidence_modes=["shadow_decision"], previous_receipt=previous) + leg = {key: {"synthetic": True, "value": 0} for key in ("signal", "hypothetical_order", "position", "cost", "return")} + observations.append(dict(forward_observation_receipt=receipt, baseline_id="cn-baseline", + observed_at=day + "T16:00:00+08:00", input_snapshot_sha256="a" * 64, candidate=leg, baseline=leg)) + previous = receipt + identity = {key: "sha256:" + "f" * 64 for key in job._IDENTITY_FIELDS} + proposal = SimpleNamespace(strategy_profile=job.PROFILE, domain="cn_equity", current_params={"a": 1}, + proposed_params={"a": 2}, computed_at="2026-09-06T08:00:00Z") + payload = dict(strategy_profile=job.PROFILE, domain="cn_equity", source_revision=REVISION, + research_identity=identity, current_params={"a": 1}, proposed_params={"a": 2}, observations=observations) + observation_path = write(tmp_path / "observations.json", payload) + binding = dict(forward_policy=policy_args, observation_path=str(observation_path), calendar_path=str(calendar), + calendar_sha256=hashlib.sha256(calendar.read_bytes()).hexdigest(), baseline_id="cn-baseline", + frozen_dependency_digests={key: value for key, value in dependencies.items() if key != "p1_manifest"}, + retry_after_seconds=3600) + return binding, identity, proposal, payload + + +def test_shadow_requires_complete_real_calendar_and_receipt_chain(shadow_inputs): + from quant_platform_kit.strategy_lifecycle.paired_shadow_adapter import collect_paired_shadow_for_promotion + binding, identity, proposal, _ = shadow_inputs + result = job._make_shadow_reader(binding, identity, REVISION)(proposal) + assert result["status"] == "complete" + record = collect_paired_shadow_for_promotion(result["observation"]) + assert record["passed"] and record["no_order"] and not record["live_authority_granted"] + + +@pytest.mark.parametrize("missing", [False, True]) +def test_incomplete_shadow_is_pending_without_fabricating_success(shadow_inputs, missing): + binding, identity, proposal, payload = shadow_inputs + path = Path(binding["observation_path"]) + if missing: + path.unlink() + else: + payload["observations"] = payload["observations"][:1] + write(path, payload) + result = job._make_shadow_reader(binding, identity, REVISION)(proposal) + assert result == {"status": "pending", "passed": False, "no_order": True, "live_authority_granted": False, + "retry_at": NOW.timestamp() + 3600} + + +@pytest.mark.parametrize("change", ["params", "source", "calendar", "chain", "backfill", "future", "missing_session"]) +def test_shadow_wrong_binding_or_backfilled_evidence_cannot_complete(shadow_inputs, change): + binding, identity, proposal, payload = shadow_inputs + if change == "params": + payload["proposed_params"] = {"a": 3} + elif change == "source": + payload["source_revision"] = "e" * 40 + elif change == "calendar": + binding["calendar_sha256"] = "b" * 64 + elif change == "chain": + payload["observations"][1]["forward_observation_receipt"]["previous_receipt_sha256"] = "b" * 64 + elif change == "backfill": + proposal.computed_at = "2026-09-09T00:00:00Z" + elif change == "future": + payload["observations"][1]["observed_at"] = "2026-09-10T00:00:00Z" + else: + payload["observations"] = payload["observations"][1:] + write(Path(binding["observation_path"]), payload) + if change == "calendar": + with pytest.raises(ValueError, match="shadow_calendar_mismatch"): + job._make_shadow_reader(binding, identity, REVISION) + return + result = job._make_shadow_reader(binding, identity, REVISION)(proposal) + assert result["status"] == "failed" and result["passed"] is False + assert "observation" not in result + + +def test_resolve_research_owner_requires_exact_source_issue_and_body_marker(monkeypatch): + source = watcher() + source["issues"] = [{ + "repo": job.ISSUE_REPOSITORY, + "watcher_issue_key": "watcher-key-1", + "url": f"https://github.com/{job.ISSUE_REPOSITORY}/issues/123", + "task": {"event_key": "aaaaaaaaaaaa"}, + }] + monkeypatch.setenv("STRATEGY_WATCH_SOURCE_REPO", job.ISSUE_REPOSITORY) + monkeypatch.setenv("GH_TOKEN", "synthetic") + with patch.object(job, "_github_issue_get", return_value={ + "state": "open", "number": 123, + "html_url": f"https://github.com/{job.ISSUE_REPOSITORY}/issues/123", + "body": "\noriginal", + }): + assert job._resolve_research_owner(source, source["research_task_source_snapshot"]["tasks"][0]) == { + "repository": job.ISSUE_REPOSITORY, "issue_number": 123, "watcher_issue_key": "watcher-key-1", + } + source["issues"][0]["url"] = "https://github.com/QuantStrategyLab/Other/issues/123" + assert job._resolve_research_owner(source, source["research_task_source_snapshot"]["tasks"][0]) is None + + +@pytest.mark.parametrize("issue", [ + {"state": "OPEN", "number": 123, "body": ""}, + {"state": "OPEN", "number": 123, "html_url": "https://github.com/QuantStrategyLab/CnEquitySnapshotPipelines/pull/123", + "body": ""}, + {"state": "OPEN", "number": 123, "html_url": "https://github.com/QuantStrategyLab/Other/issues/123", + "body": ""}, + {"state": "OPEN", "number": 123, "html_url": "https://github.com/QuantStrategyLab/CnEquitySnapshotPipelines/issues/123", + "body": "unrelated"}, +]) +def test_resolve_research_owner_rejects_unverifiable_github_detail(monkeypatch, issue): + source = watcher() + source["issues"] = [{ + "repo": job.ISSUE_REPOSITORY, + "watcher_issue_key": "watcher-key-1", + "url": f"https://github.com/{job.ISSUE_REPOSITORY}/issues/123", + "task": {"event_key": "aaaaaaaaaaaa"}, + }] + monkeypatch.setenv("STRATEGY_WATCH_SOURCE_REPO", job.ISSUE_REPOSITORY) + monkeypatch.setenv("GH_TOKEN", "synthetic") + with patch.object(job, "_github_issue_get", return_value=issue): + assert job._resolve_research_owner(source, source["research_task_source_snapshot"]["tasks"][0]) is None + + +def _archive_ticket_fixture(tmp_path, *, archived=True, paused=False, stage_status="completed"): + root = tmp_path / "state" + ticket_dir = root / "research_promotion_tickets" + ticket_dir.mkdir(parents=True) + ticket_path = ticket_dir / ("rpt_" + "a" * 64 + ".json") + owner = {"repository": job.ISSUE_REPOSITORY, "issue_number": 123, "watcher_issue_key": "watcher-key-1"} + progress = { + "identity": {"owner": owner}, "scope_key": "b" * 64, + "lifecycle": {"archived": archived, "paused": paused, "archive_reason": "idle_timeout"}, + "stages": {"optimize": {"status": stage_status}}, + } + ticket = SimpleNamespace( + ticket_id=ticket_path.stem, live_authority_granted=False, state=SimpleNamespace(value="parked"), + research_progress=progress, + ) + saved = [] + class Cycle: + @staticmethod + def load_research_promotion_ticket(path): + return ticket + @staticmethod + def save_research_promotion_ticket(value, path): + saved.append(dict(value.research_progress)) + @staticmethod + def _saved_scope_records(directory, scope_key): + return [(ticket_path, ticket)] + @staticmethod + def _research_directory_lock(directory): + from contextlib import contextmanager + @contextmanager + def lock(): + yield True + return lock() + return root, ticket_path, owner, ticket, saved, Cycle + + +def test_archived_issue_is_patched_once_and_confirmed_delivery_is_idempotent(tmp_path, monkeypatch): + root, path, owner, ticket, saved, cycle = _archive_ticket_fixture(tmp_path) + monkeypatch.setattr(job, "STATE_ROOT", root) + get_calls, patch_calls = [], [] + monkeypatch.setattr(job, "_github_issue_get", lambda repo, number: get_calls.append((repo, number)) or { + "state": "OPEN", "number": 123, + "html_url": f"https://github.com/{job.ISSUE_REPOSITORY}/issues/123", + "body": "", + }) + monkeypatch.setattr(job, "_github_issue_close", lambda repo, number, body: patch_calls.append((repo, number, body)) or { + "state": "CLOSED", "body": body, + }) + runtime = SimpleNamespace(cycle=cycle) + result = {"reason": "research_scope_archived", "ticket_path": str(path)} + assert job._archive_research_issue(runtime, result, owner)["status"] == "confirmed" + assert len(patch_calls) == 1 + assert "research-scope-archived:" in patch_calls[0][2] + assert job._archive_research_issue(runtime, result, owner)["reason"] == "issue_already_archived" + assert len(patch_calls) == 1 + assert saved[-1]["issue_archive"]["status"] == "confirmed" + + +def test_unknown_archive_only_gets_on_follow_up_and_never_repatches(tmp_path, monkeypatch): + root, path, owner, ticket, saved, cycle = _archive_ticket_fixture(tmp_path) + monkeypatch.setattr(job, "STATE_ROOT", root) + get_calls, patch_calls = [], [] + monkeypatch.setattr(job, "_github_issue_get", lambda repo, number: get_calls.append((repo, number)) or { + "state": "OPEN", "number": 123, + "html_url": f"https://github.com/{job.ISSUE_REPOSITORY}/issues/123", + "body": "", + }) + monkeypatch.setattr(job, "_github_issue_close", lambda *_: patch_calls.append(1) or (_ for _ in ()).throw(TimeoutError())) + runtime = SimpleNamespace(cycle=cycle) + result = {"reason": "research_scope_archived", "ticket_path": str(path)} + assert job._archive_research_issue(runtime, result, owner)["reason"] == "issue_archive_unknown" + assert len(patch_calls) == 1 + assert job._archive_research_issue(runtime, result, owner)["reason"] == "issue_archive_unknown" + assert len(patch_calls) == 1 + assert len(get_calls) == 2 + + +@pytest.mark.parametrize("kwargs", [ + {"paused": True}, {"stage_status": "unknown"}, {"archived": False}, +]) +def test_archive_requires_verified_terminal_ticket_without_protection(tmp_path, monkeypatch, kwargs): + root, path, owner, ticket, saved, cycle = _archive_ticket_fixture(tmp_path, **kwargs) + monkeypatch.setattr(job, "STATE_ROOT", root) + get = Mock() + close = Mock() + monkeypatch.setattr(job, "_github_issue_get", get) + monkeypatch.setattr(job, "_github_issue_close", close) + result = job._archive_research_issue(SimpleNamespace(cycle=cycle), {"ticket_path": str(path)}, owner) + assert result["reason"] == "research_ticket_not_archiveable" + get.assert_not_called() + close.assert_not_called() + + +def test_archive_is_blocked_by_active_ticket_in_same_scope(tmp_path, monkeypatch): + root, path, owner, ticket, saved, cycle = _archive_ticket_fixture(tmp_path) + active = SimpleNamespace( + ticket_id="rpt_" + "c" * 64, live_authority_granted=False, + state=SimpleNamespace(value="awaiting_human"), + research_progress={ + "identity": {"owner": owner}, "scope_key": "b" * 64, + "lifecycle": {"archived": False, "paused": False}, + "stages": {"optimize": {"status": "completed"}}, + }, + ) + cycle._saved_scope_records = staticmethod(lambda directory, scope_key: [(path, ticket), (path, active)]) + monkeypatch.setattr(job, "STATE_ROOT", root) + get = Mock() + close = Mock() + monkeypatch.setattr(job, "_github_issue_get", get) + monkeypatch.setattr(job, "_github_issue_close", close) + result = job._archive_research_issue(SimpleNamespace(cycle=cycle), {"ticket_path": str(path)}, owner) + assert result["reason"] == "research_ticket_not_archiveable" + get.assert_not_called() + close.assert_not_called() + + +@pytest.mark.parametrize("issue", [ + {"state": "OPEN", "number": 123, "html_url": "https://github.com/QuantStrategyLab/CnEquitySnapshotPipelines/issues/123", + "body": "unrelated"}, + {"state": "OPEN", "number": 123, "html_url": "https://github.com/QuantStrategyLab/CnEquitySnapshotPipelines/pull/123", + "body": ""}, + {"state": "OPEN", "number": 123, "html_url": "https://github.com/QuantStrategyLab/Other/issues/123", + "body": ""}, +]) +def test_archive_rechecks_exact_issue_before_patch(tmp_path, monkeypatch, issue): + root, path, owner, ticket, saved, cycle = _archive_ticket_fixture(tmp_path) + monkeypatch.setattr(job, "STATE_ROOT", root) + monkeypatch.setattr(job, "_github_issue_get", lambda *_: issue) + close = Mock() + monkeypatch.setattr(job, "_github_issue_close", close) + result = job._archive_research_issue(SimpleNamespace(cycle=cycle), {"ticket_path": str(path)}, owner) + assert result["reason"] == "issue_archive_unknown" + close.assert_not_called() + + +def test_shadow_cannot_label_future_sessions_as_completed(shadow_inputs): + binding, identity, proposal, _ = shadow_inputs + # The second receipt claims Sep 8, but its timestamp is still Sep 7. + payload = json.loads(Path(binding["observation_path"]).read_text()) + payload["observations"][1]["observed_at"] = "2026-09-07T17:00:00+08:00" + write(Path(binding["observation_path"]), payload) + result = job._make_shadow_reader(binding, identity, REVISION)(proposal) + assert result["status"] == "failed" and result["passed"] is False + + +@pytest.mark.parametrize("field", ["forward_policy", "calendar_sha256", "frozen_dependency_digests", "baseline_id"]) +def test_bad_shadow_configuration_is_rejected_before_constructing_a_callback(shadow_inputs, field): + binding, identity, _, _ = shadow_inputs + binding[field] = {} if field.endswith("digests") or field == "forward_policy" else "" + with pytest.raises(ValueError): + job._make_shadow_reader(binding, identity, REVISION) + + +@pytest.mark.parametrize("binding", [None, + {"sync_url": "http://synthetic.invalid/sync", "pull_url": "https://synthetic.invalid/pull"}, + {"sync_url": "https://synthetic.invalid/sync", "pull_url": "https://different.invalid/pull"}]) +def test_console_missing_or_wrong_origin_constructs_no_network_callback(binding): + importlib.import_module("quant_platform_kit") + with patch("urllib.request.urlopen", side_effect=AssertionError("unexpected request")) as http: + with pytest.raises(ValueError): + job._console_bindings(binding) + http.assert_not_called() + + +def test_shadow_cannot_relabel_old_sessions_with_postproposal_timestamps(shadow_inputs): + binding, identity, proposal, payload = shadow_inputs + proposal.computed_at = "2026-09-09T07:40:00Z" + payload["observations"][0]["observed_at"] = "2026-09-09T07:50:00Z" + payload["observations"][1]["observed_at"] = "2026-09-09T07:51:00Z" + write(Path(binding["observation_path"]), payload) + result = job._make_shadow_reader(binding, identity, REVISION)(proposal) + assert result["status"] == "failed" and "observation" not in result + + +@pytest.mark.parametrize("computed_at", ["2026-09-07T01:25:00Z", "2026-09-07T07:30:00Z"]) +def test_shadow_candidate_must_exist_before_first_session_auction(shadow_inputs, computed_at): + binding, identity, proposal, _ = shadow_inputs + proposal.computed_at = computed_at + result = job._make_shadow_reader(binding, identity, REVISION)(proposal) + assert result["status"] == "failed" and "observation" not in result + + + +def test_native_diagnosis_pending_reads_original_task_without_new_request(): + from quant_platform_kit.strategy_lifecycle.ai_provider import AiServiceConfig, AiProviderConfig + replies = [SimpleNamespace(success=False, raw={"status": "running", "id": "task-cn"}), + SimpleNamespace(success=True, provider="dot", output='{"optimization_needed":false}', + raw={"status": "completed", "id": "task-cn", "result_kind": "advisory", + "model_requested": "configured-dot", "model_verification": "unavailable"})] + client = Mock() + client.execute.side_effect = replies + runtime = SimpleNamespace(client=lambda _: client, config=AiServiceConfig, + route=SimpleNamespace(from_env=lambda: AiProviderConfig("primary", "agent", "configured-dot", "default")), + cn=SimpleNamespace(BASELINE_PARAMS={}), context=lambda **kwargs: kwargs, prompt=lambda _: "frozen synthetic prompt") + diagnosis = job._diagnosis(runtime, object(), REVISION) + pending = diagnosis() + assert pending == {"optimization_needed": False, "reason": "ai_task_pending", "task_id": "task-cn"} + decision = diagnosis.read_pending(pending) + assert decision["optimization_needed"] is False + assert decision["model_verification"] == "unavailable" + assert "model" not in decision + first, second = client.execute.call_args_list + assert first.kwargs["idempotency_key"] == second.kwargs["idempotency_key"] + assert "resume_task_id" not in first.kwargs + assert second.kwargs["resume_task_id"] == "task-cn" + + +@pytest.mark.parametrize("raw", [ + {"status": "completed", "id": "task", "result_kind": "execution"}, + {"status": "completed", "id": "", "result_kind": "advisory"}, + {"status": "succeeded", "id": "task", "result_kind": "advisory"}, +]) +def test_native_diagnosis_rejects_unbound_or_legacy_results(raw): + from quant_platform_kit.strategy_lifecycle.ai_provider import AiServiceConfig, AiProviderConfig + result = SimpleNamespace(success=True, provider="dot", raw=raw, output='{"optimization_needed":true,"recommended_method":"grid_search"}') + runtime = SimpleNamespace(client=lambda _: SimpleNamespace(execute=lambda *a, **kw: result), + config=AiServiceConfig, route=SimpleNamespace(from_env=lambda: AiProviderConfig("primary", "agent", "configured-dot", "default")), + cn=SimpleNamespace(BASELINE_PARAMS={}), context=lambda **kwargs: kwargs, prompt=lambda _: "synthetic") + with pytest.raises(ValueError, match="outcome_unavailable"): + job._diagnosis(runtime, object(), REVISION)() diff --git a/tests/test_qsl_compat_metadata.py b/tests/test_qsl_compat_metadata.py index afd068c..a4b7e7d 100644 --- a/tests/test_qsl_compat_metadata.py +++ b/tests/test_qsl_compat_metadata.py @@ -3,7 +3,7 @@ ROOT = Path(__file__).resolve().parents[1] -QPK_REVISION = "c7646a7168b3dafa763ef7751a182d23e8de7790" +QPK_REVISION = "f6f2079f6b53d3cbb3b72bdcc5c2476ed8f170f1" QPK_URL = ( "quant-platform-kit @ git+https://github.com/QuantStrategyLab/" f"QuantPlatformKit.git@{QPK_REVISION}" @@ -53,28 +53,11 @@ def test_qpk_pin_lock_and_ci_are_dependency_enabled() -> None: assert "python -m pip check" in ci -def test_research_extra_installs_frozen_sdk_and_consumer_checker(): - import importlib.metadata - import inspect - import json - import ai_gateway_client - from quant_platform_kit.strategy_lifecycle import research_promotion_cycle - from quant_platform_kit.strategy_lifecycle.promotion_actionable_runner import run_actionable_research_promotion - - assert ai_gateway_client.AiGatewayClient - assert research_promotion_cycle.run_research_promotion_cycle - assert importlib.metadata.version("quant-platform-kit") == "1.0.0" - installed_qpk = json.loads(importlib.metadata.distribution("quant-platform-kit").read_text("direct_url.json")) - assert installed_qpk["vcs_info"]["commit_id"] == QPK_REVISION - assert {"research_identity", "admit_new_research", "read_pending_shadow"} <= set(inspect.signature(run_actionable_research_promotion).parameters) +def test_research_extra_has_no_legacy_ai_sdk(): project = tomllib.loads((ROOT / "pyproject.toml").read_text()) - sdk = "ai-gateway-client @ git+https://github.com/QuantStrategyLab/AIAuditBridge.git@60bd64a2ae059a082614181eeb845b46df395523" - assert project["project"]["optional-dependencies"]["research"] == [sdk] + assert project["project"]["optional-dependencies"]["research"] == [] lock = tomllib.loads((ROOT / "uv.lock").read_text()) - package = next(p for p in lock["package"] if p["name"] == "ai-gateway-client") - assert package["source"]["git"].endswith("#60bd64a2ae059a082614181eeb845b46df395523") - installed_sdk = json.loads(importlib.metadata.distribution("ai-gateway-client").read_text("direct_url.json")) - assert installed_sdk["vcs_info"]["commit_id"] == "60bd64a2ae059a082614181eeb845b46df395523" + assert "ai-gateway-client" not in {p["name"] for p in lock["package"]} ci = (ROOT / ".github/workflows/ci.yml").read_text() assert 'consumer-qpk-pin' in ci assert '--pin-file external/QuantPlatformKit/QPK_PIN' not in ci @@ -85,5 +68,7 @@ def test_evidence_gate_uses_exact_installed_consumer_without_source_override(): assert 'tomllib.load(stream)["project"]["dependencies"]' in workflow assert "--no-deps" not in workflow assert "ref: main" not in workflow - assert 'ref: 60bd64a2ae059a082614181eeb845b46df395523' in workflow + assert "AIAuditBridge" not in workflow + assert "AI_SERVICE_REVIEWERS_JSON" in workflow + assert "CODEX_AUDIT_SERVICE_URL" not in workflow assert "python -m pip check" in workflow diff --git a/tests/test_run_walk_forward_backtest.py b/tests/test_run_walk_forward_backtest.py index 01abf1d..1471edd 100644 --- a/tests/test_run_walk_forward_backtest.py +++ b/tests/test_run_walk_forward_backtest.py @@ -37,7 +37,7 @@ def test_run_walk_forward_persists_independent_lifecycle_baseline(tmp_path: Path records = [ json.loads(path.read_text(encoding="utf-8")) - for path in (tmp_path / "backtest" / "cn_equity" / "cn_index_etf_tactical_rotation").glob("*.json") + for path in (tmp_path / "backtest" / "cn_equity" / "cn_index_etf_tactical_rotation").rglob("*.json") ] assert payload["baseline"]["sharpe_ratio"] is not None diff --git a/uv.lock b/uv.lock index 3dd57df..ceaa928 100644 --- a/uv.lock +++ b/uv.lock @@ -13,11 +13,6 @@ resolution-markers = [ "python_full_version < '3.12' and sys_platform != 'emscripten' and sys_platform != 'win32'", ] -[[package]] -name = "ai-gateway-client" -version = "0.1.0" -source = { git = "https://github.com/QuantStrategyLab/AIAuditBridge.git?rev=60bd64a2ae059a082614181eeb845b46df395523#60bd64a2ae059a082614181eeb845b46df395523" } - [[package]] name = "cn-equity-strategies" version = "0.1.0" @@ -28,19 +23,15 @@ dependencies = [ ] [package.optional-dependencies] -research = [ - { name = "ai-gateway-client" }, -] test = [ { name = "pytest" }, ] [package.metadata] requires-dist = [ - { name = "ai-gateway-client", marker = "extra == 'research'", git = "https://github.com/QuantStrategyLab/AIAuditBridge.git?rev=60bd64a2ae059a082614181eeb845b46df395523" }, { name = "pandas", specifier = ">=2.0" }, { name = "pytest", marker = "extra == 'test'", specifier = ">=8" }, - { name = "quant-platform-kit", git = "https://github.com/QuantStrategyLab/QuantPlatformKit.git?rev=c7646a7168b3dafa763ef7751a182d23e8de7790" }, + { name = "quant-platform-kit", git = "https://github.com/QuantStrategyLab/QuantPlatformKit.git?rev=f6f2079f6b53d3cbb3b72bdcc5c2476ed8f170f1" }, ] provides-extras = ["test", "research"] @@ -324,7 +315,7 @@ wheels = [ [[package]] name = "quant-platform-kit" version = "1.0.0" -source = { git = "https://github.com/QuantStrategyLab/QuantPlatformKit.git?rev=c7646a7168b3dafa763ef7751a182d23e8de7790#c7646a7168b3dafa763ef7751a182d23e8de7790" } +source = { git = "https://github.com/QuantStrategyLab/QuantPlatformKit.git?rev=f6f2079f6b53d3cbb3b72bdcc5c2476ed8f170f1#f6f2079f6b53d3cbb3b72bdcc5c2476ed8f170f1" } [[package]] name = "six"