From bc05f702a7f6ce9ed22c7fbcc9db2935326b83bf Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Wed, 9 Sep 2026 04:45:13 +0800 Subject: [PATCH 1/2] feat: wire bounded CN index ETF research into watcher Co-Authored-By: Codex --- .github/workflows/ci.yml | 29 +- .../strategy_optimization_watcher.yml | 54 ++ ...-index-etf-research-dispatch-2026-09-09.md | 70 ++ .../cn-index-etf-research.json.example | 52 ++ scripts/run_cn_index_etf_research.py | 434 +++++++++++ tests/test_run_cn_index_etf_research.py | 688 ++++++++++++++++++ 6 files changed, 1326 insertions(+), 1 deletion(-) create mode 100644 docs/cn-index-etf-research-dispatch-2026-09-09.md create mode 100644 ops/codex-audit/cn-index-etf-research.json.example create mode 100644 scripts/run_cn_index_etf_research.py create mode 100644 tests/test_run_cn_index_etf_research.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 1be0c564..10acafaa 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -48,7 +48,7 @@ jobs: python3 -m pip install 'quant-strategy-plugins[ai] @ git+https://github.com/QuantStrategyLab/QuantStrategyPlugins.git@6b76d512c8273deb804c0770a203558284778399' python3 -m pip check python3 -m ruff check . - python3 -m pytest tests ops/quant-monitor/tests -q + python3 -m pytest tests ops/quant-monitor/tests -q --ignore=tests/test_run_cn_index_etf_research.py - uses: actions/setup-node@v6 with: @@ -59,3 +59,30 @@ jobs: set -euo pipefail node --experimental-default-type=module --test cloudflare/codex-audit-proxy/tests/index.test.mjs node --experimental-default-type=module --test cloudflare/ai-gateway-dash/tests/index.test.mjs + + - name: Install the isolated CN research dependency set + run: | + set -euo pipefail + python3 -m venv "${RUNNER_TEMP}/aab-cn-research" + "${RUNNER_TEMP}/aab-cn-research/bin/python" -m pip install pytest 'cn-equity-strategies[research] @ git+https://github.com/QuantStrategyLab/CnEquityStrategies.git@2a0c5c9aafacfbe6519fb4029ca4ac18e4996a66' + "${RUNNER_TEMP}/aab-cn-research/bin/python" -m pip check + - name: Verify the real CN caller without network or model execution + env: + TZ: UTC + PYTEST_DISABLE_PLUGIN_AUTOLOAD: "1" + PYTHONDONTWRITEBYTECODE: "1" + run: | + "${RUNNER_TEMP}/aab-cn-research/bin/python" - <<'PY' + import socket + import pytest + + def blocked(*args, **kwargs): + raise AssertionError("real socket prohibited") + + socket.socket.connect = blocked + socket.create_connection = blocked + raise SystemExit(pytest.main([ + "-q", "-p", "no:cacheprovider", "--tb=short", + "tests/test_run_cn_index_etf_research.py", + ])) + PY diff --git a/.github/workflows/strategy_optimization_watcher.yml b/.github/workflows/strategy_optimization_watcher.yml index 7d23a02e..d344ca37 100644 --- a/.github/workflows/strategy_optimization_watcher.yml +++ b/.github/workflows/strategy_optimization_watcher.yml @@ -46,6 +46,9 @@ jobs: strategy-optimization-watcher: runs-on: ubuntu-latest timeout-minutes: 15 + outputs: + source_repo: ${{ env.SOURCE_REPO }} + dry_run: ${{ env.STRATEGY_WATCH_DRY_RUN }} env: FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true" SOURCE_REPO: ${{ github.event.inputs.source_repo || vars.STRATEGY_WATCH_SOURCE_REPO || 'QuantStrategyLab/CryptoLivePoolPipelines' }} @@ -327,3 +330,54 @@ jobs: name: strategy-optimization-watcher-${{ github.run_id }} path: bridge/data/output/strategy_optimization_watcher/ if-no-files-found: warn + + cn-index-etf-research: + needs: strategy-optimization-watcher + if: >- + github.ref == 'refs/heads/main' && + vars.CN_INDEX_ETF_RESEARCH_ENABLED == 'true' && + needs.strategy-optimization-watcher.outputs.source_repo == 'QuantStrategyLab/CnEquitySnapshotPipelines' && + needs.strategy-optimization-watcher.outputs.dry_run == 'false' + runs-on: [self-hosted, codex-vps] + timeout-minutes: 60 + permissions: + contents: read + actions: read + id-token: write + concurrency: + group: cn-index-etf-research-vps + cancel-in-progress: false + env: + PYTHONDONTWRITEBYTECODE: "1" + CODEX_AUDIT_SERVICE_URL: ${{ secrets.CODEX_AUDIT_SERVICE_URL }} + CODEX_AUDIT_SERVICE_AUDIENCE: ${{ vars.CODEX_AUDIT_SERVICE_AUDIENCE || 'quant-codex-audit' }} + AI_GATEWAY_RESEARCH_PROVIDERS: codex + steps: + - name: Checkout approved Bridge job + uses: actions/checkout@v6.0.3 + with: + persist-credentials: false + + - name: Read this workflow's verified watcher artifact + uses: actions/download-artifact@v5 + with: + name: strategy-optimization-watcher-${{ github.run_id }} + path: data/input/cn-index-etf-watcher + + - name: Run the policy-bound CN research job + run: | + set -euo pipefail + mkdir -p data/output/cn-index-etf-research + /opt/codex-cn-index-etf-research/venv/bin/python \ + -m scripts.run_cn_index_etf_research \ + --watcher-result data/input/cn-index-etf-watcher/result.json \ + > data/output/cn-index-etf-research/result.json + + - name: Upload sanitized research status + if: always() + uses: actions/upload-artifact@v7 + with: + name: cn-index-etf-research-${{ github.run_id }} + path: data/output/cn-index-etf-research/result.json + if-no-files-found: warn + retention-days: 7 diff --git a/docs/cn-index-etf-research-dispatch-2026-09-09.md b/docs/cn-index-etf-research-dispatch-2026-09-09.md new file mode 100644 index 00000000..be3d9586 --- /dev/null +++ b/docs/cn-index-etf-research-dispatch-2026-09-09.md @@ -0,0 +1,70 @@ +# CN 指数 ETF 的受控研究作业 + +本片从 AAB `dfccd5e21fa5e5ed0c3d226845ee988bdbab572a` 接续。源码验证、依赖采用、生产部署和真实研究周期分别验收；下列接线不代表已有真实数据或 shadow 成功。 + +## 实际入口和权限 + +已有 `strategy_optimization_watcher.yml` 的 CN 分支增加单一 VPS job。只有 main、`CN_INDEX_ETF_RESEARCH_ENABLED=true`、上游来源为 `QuantStrategyLab/CnEquitySnapshotPipelines` 且非 dry-run 时运行。它下载同一次 workflow 的原 watcher artifact，执行 `scripts.run_cn_index_etf_research`。现有来源 allowlist、来源 workflow 配置和排班不自动扩展；CN producer 未获配置或没有合法任务时不能执行。 + +作业使用 `/opt/codex-cn-index-etf-research/venv/bin/python`，固定并发组 `cn-index-etf-research-vps`，`cancel-in-progress=false`。QPK 的目录锁与票据位于 `/var/lib/codex-audit-bridge/cn-index-etf-research/research_promotion_tickets`；这只保证该 VPS、同一持久目录的串行执行，不声称跨主机互斥。 + +授权来自 root-owned `/etc/codex-audit-bridge-policy/cn-index-etf-research.json`。文件及父目录不得被普通用户或组改写；初始模板 `ops/codex-audit/cn-index-etf-research.json.example` 是 disabled 且所有实际输入留空。watcher 的 `strategy_diagnosis` task 只作触发证据，仍保持原 P3/no-order/size-zero 权限，不能替代这份独立配置。模板的示例费用是模拟执行假设，不是实盘预算。 + +模型调用只走实际 Actions OIDC 的既有 SDK `execute`，固定 Codex、`research_stage=optimization`、`review_only`，使用精确 CN 代码版本作为 `source_ref`。服务仍独立验证 OIDC 身份及允许的 workflow；脚本检查环境字段并不自行签发身份。静态 service token 被拒绝，无付费 API/Cursor fallback。SDK 固定每次提交及轮询的 job/provider/stage/model/effort。明确额度延期交给原票据保存 `retry_at`；恢复后实际 execute 前仍检查冻结窗口截止，错过时返回确定的 `optimization_needed=false / forward_window_start_elapsed` 并终止该次研究，零新 HTTP/试验。已完成 diagnosis 的 shadow/awaiting 尾部不会再调用这一模型闭包。未知结果不重新调用。 + +OIDC `repository` 保持 `QuantStrategyLab/AIAuditBridge`，请求 `source_repository` 为 `QuantStrategyLab/CnEquityStrategies`。既有服务实际验证 caller/direct-repository、workflow/ref allowlist，以及 source allowlist 和同组织边界，不能把 CN 源码声明改成 CN OIDC 身份。本轮部署负责人已只读核实生产默认 OIDC、AAB caller/direct、watcher workflow@main、main ref、CN source 和 public visibility 均匹配；没有发模型请求，这不是实际 Actions job 的身份验收。 + +## 输入与版本 + +配置必须明确三个实际安装版本：CN `code_revision`、QPK `qpk_revision`、AAB SDK `sdk_revision`，均为已批准的 40 位 Git commit。运行时核对已安装 distribution 的 `direct_url.json` VCS commit；不能用裸版本号、editable 临时源码或本地 wheel 路径冒充该部署验证。解释器及包由部署方安装，workflow 不安装依赖或修改代码。 + +`inputs.development/validation` 各自绑定已有本地许可数据包路径和已批准 manifest SHA-256。CN 原 `read_index_etf_input` 读取 `research_input_manifest.v1.json`，再用原 `preflight_index_etf_research_job` 验证开发、三折 WFA、锁定 OOS、费用及实际代码。返回的五字段 identity 必须等于 root policy 已冻结值；参数空间由既有 CN 模块限定为 12 个组合，模型不能改变。身份计算包括实际源码字节，不能只改环境变量或沿用旧 Git ref 来复用旧实验。 + +`drift.path` 引用真实生产 `DriftResult.to_dict()` 文件，绑定 profile/domain/source_revision，保留原 `as_of` 和有限 score，拒绝 future、suppressed、缺 baseline 或状态不一致。新研究仍由 QPK 原 7 个自然日时效门拒绝旧观测。task.created_at 不会变成观测日期，severity 不会生成 score。历史 task 和原观测仅可定位已经完成前置阶段的同一 shadow 票据；是否准许只读恢复由 QPK 持锁检查决定。 + +新票据首次保存、任何模型或回测之前，QPK 在同一目录锁内调用 `admit_one_new_experiment(ticket_dir, created_at)`。它先用同一个将写入票据的 UTC 时间确认仍严格早于已冻结首 session 的 09:25+08:00；错过这一点的新实验在模型、回测、当日票据保存前拒绝。已有票据不会重做新实验准入。然后只数原目录票据：当日已有一个、坏文件、未来/缺失时间或未知身份则拒绝。已有终态/unknown/pending 票据不再次准入，没有第二计数表或队列。试验与逐次参数记录由 CN 原函数存入独立实验目录。 + +## 真实 shadow 读取与候选 + +`shadow.forward_policy` 是完整 `ForwardObservationPolicy` 构造参数，候选、窗口、要求交易日数、基准和理由必须来自实际批准策略，不继承其他市场的默认期限。当前入口仅支持 XSHG 固定窗口、shadow-only；没有 paper、broker 或下单路径。`calendar_path/calendar_sha256` 绑定真实交易日历 JSON 日期数组；不能用工作日推算代替。 + +`shadow.observation_path` 由拥有观测责任的 producer 提供，顶层包含 `strategy_profile/domain/source_revision/research_identity/current_params/proposed_params/observations`。前六项必须匹配实际保存的 proposal、CN 版本及五字段身份。`observations` 是从窗口第一交易日至当前交易日的顺序数组；每项沿既有 paired adapter 字段：`forward_observation_receipt/baseline_id/observed_at/input_snapshot_sha256/candidate/baseline`。reader 使用 root policy 注入 `ForwardObservationPolicy`，逐项验证原 receipt 和 paired evidence，再把已验证前项传给后项。不会创建或补写 forward receipt。 + +`frozen_dependency_digests` 必须提供 `p2_config/p3_evidence/risk_policy/strategy_release/plugin_bundle` 的已批准真实根；每条 receipt 必须一致，p1 manifest 则绑定本次实际 input snapshot。baseline_id、当前/候选参数、代码和来源必须匹配；proposal 必须严格早于首个计入交易日的 09:25+08:00（含等于也拒绝，沿 CN next-open 的冻结时序）。每条观测的上海时区日期必须等于它声明的 session，时间不得早于 proposal 或该 session 的 15:00 收盘，亦不得晚于当前时间。候选产生过晚时，需由拥有配置责任者从真实日历选择下一合格窗口；reader 不改写已有 policy 日期。完整连续链必须覆盖真实 calendar 的全部要求交易日，单条合法 receipt 不能冒充已完成窗口。 + +forward policy、真实 calendar 摘要、固定来源摘要和 baseline 配置在模型前校验。缺 observation 文件或合法窗口尚未收满时返回 pending，按配置的 60–86400 秒后仅重读同一个 provider。QPK 复用已完成 AI/优化/回测，禁止重新调用初始 record callback。坏链或错身份明确失败；读取结果未知仍按 unknown 停车。完成时将完整 observation 交回 QPK 原 paired validator，之后才可能到 awaiting_human。旧观测跨 7 日只能继续这条已验证尾部，不能改写 as_of 创建新研究。 + +`console` 配置包含同一 HTTPS 站点的 `sync_url/pull_url` 与受限 `token_path`；token 由批准的本地文件消费，不进入环境、输出或工件。使用原 `make_console_research_promotion_sync/pull`：先 GET，只有确认不存在才 POST；未知提交后仅 GET 回收；人工接受仍只是意图，`live_authority_granted=false`。未配置 console 或输入时在模型前停车，零 POST。 + +## 发布与真实验收仍需的材料 + +部署方需一次采用最终通过审查并已发布的 CN/QPK/SDK commit，建立固定解释器与持久目录，保留原服务配置、凭据及 monitor 的两份本地文档改动。先保持变量及 policy disabled，验证安装来源、缺配置出口与零模型/零 POST，再配置真实许可输入、观测文件、原始 drift 和 console 只读/提交权限。 + +目前没有真实 CN 许可数据包、完整 forward producer 工件或生产周期成功证据。离线测试使用明确合成结构与拦截 HTTP，不得称作真实回测晋级。CLI 只输出固定状态、任务 ID、原观测日期/来源、研究 key、是否恢复/确认及延期时间；上传此摘要，不上传原始行情、许可证、完整票据、shadow legs 或底层异常。 + + +## 本轮离线验证（2026-09-09） + +生产与测试冻结为这六个文件：原 watcher workflow、required CI 的隔离验证步骤、新受控入口及专项测试、disabled 配置模板、本交接。没有改模型 HTTP 服务、SDK 协议、交易入口或第二队列。最终采用版本： + +- CN：`2a0c5c9aafacfbe6519fb4029ca4ac18e4996a66`。 +- QPK：`b5654244aa5d08bce2b4b4f931436268d57216df`。 +- AAB SDK：`60bd64a2ae059a082614181eeb845b46df395523`。 + +这三个 Git commit 已实际非 editable 安装至 `/tmp/aab-cn-final-integration-20260909`。初次安装器复用全局同版本号旧 CN，真实 identity gate 拒绝；在隔离环境显式安装精确版本后，`direct_url.json` 和 import 位置均来自该 venv。没有使用 QPK/CN source overlay 代替安装结果。 + +验证命令为该 venv 的 `python -m pytest -q -p no:cacheprovider --tb=short tests/test_run_cn_index_etf_research.py tests/test_run_strategy_optimization_watcher.py tests/test_strategy_optimization_watcher_workflow.py`。实际以清空环境、`TZ=UTC`、禁自动插件/pyc，并拦截 socket connect 的 wrapper 运行，**97 passed，无 skip**。日志位于 `/tmp/aab-cn-dispatch-focused-20260909.log`。Ruff、actionlint、`git diff --check` 均通过。 + +关键回归先出现失败再修复：旧入口不存在；浅层编排未调用真实 runner；shadow 缺真实 reader；安装来源未拒绝混合 editable/VCS 元数据；future session 可被早于该交易日的时间标记完成；旧 session 可事后补时间，或首日收盘后才生成候选仍被计为 forward；已过冻结点的新实验未在 AI 前拒绝；shadow 配置错误在模型后才暴露；QRT reader 的静默 printer 不接受实际 `flush` 参数。最终真实 CN reader/preflight/数字 optimizer/QPK tickets 和已安装 SDK 均参与离线测试：合成平价数据运行 baseline + 12 组后正确 reject，同一票据不再调用模型，新观测仍受 UTC 每日一项限制。完整 shadow 两 receipt 用例和 QRT GET→POST未知→GET、再次只 GET 都通过各自实际 adapter；不将这些分段离线结果说成真实候选晋级。 + +新增 SDK→实际服务 auth/execute handler 的八个隔离用例：允许的 AAB 身份与 CN source 正常提交；错 repository/workflow/ref/direct/source、跨组织和静态 token 均零 job、零扣额。RSA/JWKS、额度和模型执行使用明确测试替身；实际签名、远端 job 和 forward producer 仍需部署后的独立证据。 + +额外运行未改动的 `test_verify_subscription_ai_consumer.py` 时出现 8 个环境依赖失败（组合 158 passed）：它导入本机全局旧 QSP `1f3a27b8fd83d71b583f4f5160a748e95fbefaa1`，触发旧 feedback/HTTP 行为。本轮未安装 QSP，也未修改该脚本或测试，两者与基线一致；不把这次扩展运行称为通过。原输出保留在 `/tmp/aab-cn-final-tests-20260909.log`，不扩大本片写集。 + + +后续经独立审查补齐同一 shadow 时间因果边界和 CI 采用。`.github/workflows/ci.yml` 在现有 required `test` job 内增加独立 venv 步骤：安装 `cn-equity-strategies[research]` 的上述精确 commit，其已发布依赖绑定 QPK/SDK 两个精确版本；执行 `pip check` 后拦截真实 socket，运行整个 CN 专项。原主 suite 仅显式排除这一个文件，由同一个 required `test` job 的隔离 venv 承接；专项任何失败都会使该 required check 失败。没有新增可选 job、`needs`/skip 门或更改远端分支保护，既有 QSP 依赖与测试不变。专项已取消所有 `importorskip`；缺包、错 pin 或实际调用断裂都失败，不以 skip 代替覆盖。 + +该 CI 安装过程又在全新、不继承 system-site-packages 的 `/tmp/aab-cn-ci-rehearsal-20260909` 实际重演，只有 `pytest` 与固定 `CN[research]` 及其依赖。实际 Python 为 3.13.7（GitHub CI 配置 3.12）。`pip check` 通过；最后三文件组合 **103 passed，无 skip**，其中 CN 专项 77、原 watcher 26。最终三文件组合日志位于 `/tmp/aab-cn-ci-rehearsal-20260909.log`；前面的 97 项是前一冻结点结果。此前将隔离安装/运行移动到既有 required job 的结构回归实际先 RED 后该单例 1 passed。最终 103 项还覆盖后补的 quota-deferred 跨截止恢复：旧代码重复产生 5 个 HTTP 请求，修复后仅首次 429 准入的 3 个请求，零试验，原票据确定性终止。Ruff/actionlint/diffcheck 均通过。未在此环境安装 QSP，也不将前述 QSP 扩展检查的失败写成通过。 + + +独立 reviewer 已完成全部源码及增量审查，无未关闭 P1/P2：原 97 项、时间边界定向验证及最后实际 caller/额度延期/required CI 的 5 项均通过；包含首 session 09:24:59 允许、等于 09:25 和收盘后冻结拒绝。评审不等于真实数据或线上模型验收，发布/部署仍由主任务统一执行。 diff --git a/ops/codex-audit/cn-index-etf-research.json.example b/ops/codex-audit/cn-index-etf-research.json.example new file mode 100644 index 00000000..7631127d --- /dev/null +++ b/ops/codex-audit/cn-index-etf-research.json.example @@ -0,0 +1,52 @@ +{ + "enabled": false, + "candidate_id": "cn_index_etf_tactical_rotation", + "domain": "cn_equity", + "code_revision": "2a0c5c9aafacfbe6519fb4029ca4ac18e4996a66", + "qpk_revision": "b5654244aa5d08bce2b4b4f931436268d57216df", + "sdk_revision": "60bd64a2ae059a082614181eeb845b46df395523", + "research_identity": { + "code_revision": null, + "input_revision": null, + "param_space_revision": null, + "cost_model_revision": null, + "validator_revision": null + }, + "drift": {"path": null, "source_revision": null}, + "inputs": { + "development": {"path": null, "manifest_sha256": null}, + "validation": {"path": null, "manifest_sha256": null} + }, + "plan": { + "development_start": null, + "development_end": null, + "folds": [], + "locked_oos_start": null, + "locked_oos_end": null, + "purge_days": null, + "embargo_days": null + }, + "execution_config": { + "initial_cash": 1000000.0, + "minimum_commission": 5.0, + "cash_reserve_ratio": 0.02, + "max_previous_volume_participation": 0.01, + "lot_size": 100 + }, + "cost_model": { + "model_id": "cn_index_etf.next_open.v1", + "commission_bps": 3.0, + "slippage_bps": 5.0, + "market_impact_bps": 0.0 + }, + "shadow": { + "forward_policy": null, + "observation_path": null, + "calendar_path": null, + "calendar_sha256": null, + "baseline_id": null, + "frozen_dependency_digests": null, + "retry_after_seconds": 86400 + }, + "console": null +} diff --git a/scripts/run_cn_index_etf_research.py b/scripts/run_cn_index_etf_research.py new file mode 100644 index 00000000..0dd0fc4d --- /dev/null +++ b/scripts/run_cn_index_etf_research.py @@ -0,0 +1,434 @@ +#!/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 +from types import SimpleNamespace +from typing import Any +from zoneinfo import ZoneInfo + +from service.research_task import validate_strategy_diagnosis_task + +PROFILE = "cn_index_etf_tactical_rotation" +STRATEGY_REPOSITORY = "QuantStrategyLab/CnEquityStrategies" +BRIDGE_REPOSITORY = "QuantStrategyLab/AIAuditBridge" +WORKFLOW_REF = f"{BRIDGE_REPOSITORY}/.github/workflows/strategy_optimization_watcher.yml@refs/heads/main" +POLICY_PATH = Path("/etc/codex-audit-bridge-policy/cn-index-etf-research.json") +STATE_ROOT = Path("/var/lib/codex-audit-bridge/cn-index-etf-research") +_REVISION = re.compile(r"[0-9a-f]{40}") +_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") == BRIDGE_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("CODEX_AUDIT_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 _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"), ("ai-gateway-client", "sdk_revision")): + _installed_revision(package, policy[revision]) + from ai_gateway_client import AiGatewayClient, GatewayConfig + 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 + return SimpleNamespace(cn=cn, client=AiGatewayClient, config=GatewayConfig, + 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) + + +def _diagnosis(runtime, drift, revision): + config = runtime.config.from_env() + if config.research_providers != ("codex",): + raise ValueError("codex_only_required") + client = runtime.client(config) + context = runtime.context(strategy_profile=PROFILE, domain="cn_equity", drift=drift, + current_params=runtime.cn.BASELINE_PARAMS) + + def diagnose(*_): + result = client.execute(runtime.prompt(context), mode="review_only", research_stage="optimization", + allowed_providers=["codex"], source_repository=STRATEGY_REPOSITORY, source_ref=revision, timeout=600) + raw = result.raw + if not result.success and isinstance(raw, dict) and raw.get("status") == "deferred": + return {"optimization_needed": False, "reason": "codex_research_deferred", "retry_at": raw.get("retry_at")} + if (result.success is not True or result.provider != "codex" or result.error or result.note + or not isinstance(raw, dict) or raw.get("status") != "succeeded"): + 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": "codex_research_decision", "provider": result.provider, "model": result.model, + "reasoning_effort": raw["reasoning_effort"], "job_id": raw["job_id"]} + return diagnose + + +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) + 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"]) + 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, admit_new_research=admit_new) + return _summary(result["status"], result["reason"], task_id=task["task_id"], + observation_as_of=drift["as_of"], observation_source_revision=drift["source_revision"], + **{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_run_cn_index_etf_research.py b/tests/test_run_cn_index_etf_research.py new file mode 100644 index 00000000..a2ac1f60 --- /dev/null +++ b/tests/test_run_cn_index_etf_research.py @@ -0,0 +1,688 @@ +"""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 patch +from types import SimpleNamespace +import hashlib +import importlib +import io +import urllib.error + +import pytest + +from scripts import run_cn_index_etf_research as job +from service.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/AIAuditBridge", + "GITHUB_REF": "refs/heads/main", "GITHUB_EVENT_NAME": "workflow_dispatch", + "GITHUB_WORKFLOW_REF": "QuantStrategyLab/AIAuditBridge/.github/workflows/strategy_optimization_watcher.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", + "CODEX_AUDIT_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", [{"CODEX_AUDIT_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/AIAuditBridge/.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_actual_entrypoint_calls_cn_runner_with_persistent_callbacks(shadow_inputs, tmp_path): + # This checks orchestration only; separate source integration runs the CN code. + from unittest.mock import Mock + binding, identity, _, _ = shadow_inputs + runtime = SimpleNamespace(cn=SimpleNamespace(preflight_index_etf_research_job=Mock(return_value=identity), + run_index_etf_research_job=Mock(return_value={"status": "parked", "reason": "synthetic_gate_rejection", + "research_key": "f" * 64, "resumed": False, "console_synced": None})), + read_input=Mock(return_value=object()), fold=lambda **kw: kw, + execution_config=lambda **kw: kw, cost_model=lambda **kw: kw, + probe=Mock(return_value={"actionable": True, "status": "critical"}), + drift=lambda **kw: kw, status=lambda value: value) + drift = write(tmp_path / "drift.json", dict(strategy_profile=job.PROFILE, domain="cn_equity", + source_revision=REVISION, as_of="2026-09-09", drift_score=.8, status="critical")) + policy = dict(enabled=True, code_revision=REVISION, candidate_id=job.PROFILE, domain="cn_equity", + research_identity=identity, drift={"path": str(drift), "source_revision": REVISION}, shadow=binding, console={}, + inputs={name: {"path": str(tmp_path), "manifest_sha256": "f" * 64} for name in ("development", "validation")}, + plan=dict(development_start="2020-01-01", development_end="2020-12-31", folds=[], + locked_oos_start="2024-01-01", locked_oos_end="2025-01-01", purge_days=1, embargo_days=1), + execution_config={}, cost_model={}) + sync, pull, diagnose = Mock(), Mock(), Mock() + with patch.object(job, "_read_policy", return_value=policy), patch.object(job, "_load_runtime", return_value=runtime), \ + patch.object(job, "_diagnosis", return_value=diagnose), patch.object(job, "_console_bindings", return_value=(sync, pull)), \ + patch.object(job, "STATE_ROOT", tmp_path / "state"): + result = job.run_from_watcher(watcher()) + assert result["reason"] == "synthetic_gate_rejection" + kwargs = runtime.cn.run_index_etf_research_job.call_args.kwargs + assert callable(kwargs["admit_new_research"]) + assert kwargs["record_shadow"] is kwargs["read_pending_shadow"] + assert kwargs["sync_console"] is sync and kwargs["pull_console"] is pull + assert kwargs["diagnose"]() == {"optimization_needed": False, "reason": "forward_window_start_elapsed"} + diagnose.assert_not_called() + assert kwargs["as_of"] == "2026-09-09" and kwargs["source_revision"] == REVISION + assert kwargs["ticket_dir"] == tmp_path / "state" / "research_promotion_tickets" + assert "ticket" not in result and "trial_records_path" not in result + + +class Response: + def __init__(self, value): + self.body = json.dumps(value).encode() + def __enter__(self): + return self + def __exit__(self, *_): + return None + def read(self): + return self.body + + +def diagnosis_runtime(): + importlib.import_module("quant_platform_kit") + from client.config import GatewayConfig + from client.gateway_client import AiGatewayClient + from quant_platform_kit.strategy_lifecycle.codex_integration import AiOptimizationContext, build_optimization_prompt + from quant_platform_kit.strategy_lifecycle.contracts import DriftResult, DriftStatus + drift = DriftResult(strategy_profile=job.PROFILE, domain="cn_equity", as_of=NOW.date(), + drift_score=.8, status=DriftStatus.CRITICAL, source_revision=REVISION) + return SimpleNamespace(config=GatewayConfig, client=AiGatewayClient, context=AiOptimizationContext, + prompt=build_optimization_prompt, cn=SimpleNamespace(BASELINE_PARAMS={"top_n": 1})), drift + + +@pytest.mark.parametrize("mismatch", [None, "job_id", "provider", "model", "research_stage", "reasoning_effort"]) +def test_real_sdk_binds_optimization_job_and_route_without_paid_fallback(mismatch): + runtime, drift = diagnosis_runtime() + route = dict(job_id="original-experiment", provider="codex", model="gpt-5.6-sol", + research_stage="optimization", reasoning_effort="high") + final = {**route, "status": "succeeded", "output": '{"optimization_needed":true,"recommended_method":"grid_search"}'} + if mismatch: + final[mismatch] = "different" + replies = [Response({"value": "synthetic-oidc"}), Response({"codex_research_routing": "v1"}), + Response({**route, "status": "queued"}), Response({"value": "synthetic-oidc"}), Response(final)] + with patch("client.gateway_client.urllib.request.urlopen", side_effect=replies) as http, \ + patch("client.gateway_client.time.sleep"), patch.object(runtime.client, "analyze", side_effect=AssertionError("paid fallback")): + if mismatch: + with pytest.raises(ValueError, match="research_model_outcome_unavailable"): + job._diagnosis(runtime, drift, REVISION)() + else: + result = job._diagnosis(runtime, drift, REVISION)() + assert result["optimization_needed"] is True and result["model"] == "gpt-5.6-sol" + assert result["job_id"] == route["job_id"] + submitted = json.loads(http.call_args_list[2].args[0].data) + assert submitted["source_ref"] == REVISION and submitted["source_repository"] == job.STRATEGY_REPOSITORY + assert submitted["mode"] == "review_only" and submitted["allowed_providers"] == ["codex"] + + +def test_real_sdk_quota_defers_and_sanitizes_without_polling(): + runtime, drift = diagnosis_runtime() + error = urllib.error.HTTPError("https://synthetic.invalid", 429, "private-error", {}, + io.BytesIO(json.dumps({"status": "deferred", "retry_at": NOW.timestamp() + 3600, "stderr": "private-error"}).encode())) + with patch("client.gateway_client.urllib.request.urlopen", side_effect=[ + Response({"value": "synthetic-oidc"}), Response({"codex_research_routing": "v1"}), error, + ]) as http: + result = job._diagnosis(runtime, drift, REVISION)() + assert result == {"optimization_needed": False, "reason": "codex_research_deferred", "retry_at": NOW.timestamp() + 3600} + assert http.call_count == 3 and "private" not in str(result) + + +def test_watcher_uses_one_opt_in_vps_owner_with_same_run_artifact(): + text = (Path(__file__).resolve().parents[1] / ".github/workflows/strategy_optimization_watcher.yml").read_text() + research = text.split(" cn-index-etf-research:", 1)[1] + for required in ("needs: strategy-optimization-watcher", "runs-on: [self-hosted, codex-vps]", + "vars.CN_INDEX_ETF_RESEARCH_ENABLED == 'true'", "github.ref == 'refs/heads/main'", + "needs.strategy-optimization-watcher.outputs.source_repo == 'QuantStrategyLab/CnEquitySnapshotPipelines'", + "needs.strategy-optimization-watcher.outputs.dry_run == 'false'", "group: cn-index-etf-research-vps", + "cancel-in-progress: false", "id-token: write", "AI_GATEWAY_RESEARCH_PROVIDERS: codex", + "name: strategy-optimization-watcher-${{ github.run_id }}", + "/opt/codex-cn-index-etf-research/venv/bin/python", "-m scripts.run_cn_index_etf_research"): + assert required in research + assert "CODEX_AUDIT_SERVICE_TOKEN" not in research + assert "issues: write" not in research and "pip install" not in research + assert "path: data/output/cn-index-etf-research/result.json" in research + + +def input_package(root, start, end): + """Historical *schema* stand-in; all prices/license text are synthetic.""" + import pandas as pd + from quant_platform_kit.data.research_input import canonical_research_input_manifest_bytes + days = list(pd.bdate_range(start, end).strftime("%Y-%m-%d")) + rows = [dict(date=day, symbol=symbol, open=10., high=11., low=9., close=10., volume=100000, + suspended=False, limit_up=11., limit_down=9., status_known_at=day+"T09:00:00+08:00", + available_at=day+"T15:01:00+08:00") for day in days for symbol in ("510300", "510500")] + license_bytes = b"Synthetic test fixture; not a real license or market evidence." + docs = {"normalized/daily.json": rows, + "calendar/sessions.json": {"start_date": days[0], "end_date": days[-1], "sessions": days}, + "corporate_actions/events.json": {"complete_from": days[0], "complete_through": days[-1], "events": []}, + "evidence/license_identity.json": {"source_identity": "official:synthetic-fixture", "revision": "test-v1", + "retention_scope": "private-retention-permitted", "content_sha256": hashlib.sha256(license_bytes).hexdigest()}} + members = {key: json.dumps(value, separators=(",", ":")).encode() for key, value in docs.items()} + members["evidence/license.bin"] = license_bytes + manifest = dict(schema_version="research_input_manifest.v1", manifest_id="synthetic-test", + research_input_contract_id="qsl.cn_index_etf.execution_input.v1", domain="cn_equity", profile=job.PROFILE, + artifact_type="cn_index_etf_execution_history", observed_at="2026-09-08T00:00:00Z", + effective_at=days[-1]+"T15:01:00+08:00", as_of="2026-09-08T00:00:00Z", + producer={"repository": "QuantStrategyLab/CnEquitySnapshotPipelines", "commit_sha": "a"*40, + "tree_sha": "b"*40, "tool": "synthetic.fixture", "tool_version": "1"}, + calendar={"calendar_id": "SSE", "timezone": "Asia/Shanghai", "session_date": days[-1], + "source": "official:synthetic-fixture", "source_revision": "sha256:"+hashlib.sha256(members["calendar/sessions.json"]).hexdigest()}, + adjustment={"policy": "raw", "source": "official:synthetic-fixture", "source_revision": "test-v1"}, + sources=[{"source_id": "synthetic-fixture", "revision": "test-v1", "observed_at": "2026-09-08T00:00:00Z", + "content_sha256": hashlib.sha256(members["normalized/daily.json"]).hexdigest()}], + members=[{"path": path, "media_type": "application/json" if path.endswith(".json") else "text/plain", + "size_bytes": len(content), "sha256": hashlib.sha256(content).hexdigest()} for path, content in sorted(members.items())]) + encoded = canonical_research_input_manifest_bytes(manifest) + for name, content in {**members, "research_input_manifest.v1.json": encoded}.items(): + path = root / name + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(content) + return {"path": str(root), "manifest_sha256": hashlib.sha256(encoded).hexdigest()} + + +@pytest.mark.parametrize(("expired_window", "defer_past_window"), [(False, False), (True, False), (False, True)]) +def test_installed_cn_reader_preflight_real_numeric_search_ticket_reuse_and_daily_cap(shadow_inputs, tmp_path, expired_window, defer_past_window): + # No CN runner, preflight, input reader, optimizer or QPK stage is mocked. + importlib.import_module("cn_equity_strategies") + importlib.import_module("ai_gateway_client") + from unittest.mock import Mock + from datetime import date + binding, _, _, _ = shadow_inputs + if not expired_window: + # A new candidate can only begin a still-future approved observation window. + binding["forward_policy"]["observation_start_session"] = "2026-09-10" + calendar = write(Path(binding["calendar_path"]), ["2026-09-10", "2026-09-11"]) + binding["calendar_sha256"] = hashlib.sha256(calendar.read_bytes()).hexdigest() + revision = "2a0c5c9aafacfbe6519fb4029ca4ac18e4996a66" + policy = dict(enabled=True, candidate_id=job.PROFILE, domain="cn_equity", code_revision=revision, + qpk_revision="b5654244aa5d08bce2b4b4f931436268d57216df", sdk_revision="60bd64a2ae059a082614181eeb845b46df395523", shadow=binding, console={}, + inputs={"development": input_package(tmp_path / "development", "2019-01-02", "2020-12-31"), + "validation": input_package(tmp_path / "validation", "2020-01-02", "2025-01-08")}, + plan=dict(development_start="2020-01-02", development_end="2020-12-31", folds=[dict( + train_start=f"{year}-01-04", train_end=f"{year}-11-15", test_start=f"{year}-11-19", test_end=f"{year}-12-31") + for year in (2021, 2022, 2023)], locked_oos_start="2024-01-08", locked_oos_end="2025-01-08", purge_days=1, embargo_days=1), + execution_config={}, cost_model={"model_id": "cn_index_etf.next_open.v1", "commission_bps": 3., "slippage_bps": 5.}) + drift_path = write(tmp_path / "drift.json", dict(strategy_profile=job.PROFILE, domain="cn_equity", + source_revision=revision, as_of="2026-09-09", drift_score=.8, status="critical")) + policy["drift"] = {"path": str(drift_path), "source_revision": revision} + # These packages must be noneditable installations from the approved commits. + runtime = job._load_runtime(policy) + arguments = dict(development_input=runtime.read_input(Path(policy["inputs"]["development"]["path"]), expected_manifest_sha256=policy["inputs"]["development"]["manifest_sha256"]), + validation_input=runtime.read_input(Path(policy["inputs"]["validation"]["path"]), expected_manifest_sha256=policy["inputs"]["validation"]["manifest_sha256"]), + trusted_input_roots={name: value["manifest_sha256"] for name, value in policy["inputs"].items()}, code_revision=revision, + development_start=date(2020,1,2), development_end=date(2020,12,31), + folds=tuple(runtime.fold(**{key:date.fromisoformat(value) for key,value in fold.items()}) for fold in policy["plan"]["folds"]), + locked_oos_start=date(2024,1,8), locked_oos_end=date(2025,1,8), purge_days=1, embargo_days=1) + policy["research_identity"] = runtime.cn.preflight_index_etf_research_job(**arguments) + route = dict(job_id="real-caller-synthetic-http", provider="codex", research_stage="optimization", + model="gpt-5.6-sol", reasoning_effort="high") + replies = [Response({"value":"synthetic-oidc"}), Response({"codex_research_routing":"v1"}), + Response({**route,"status":"queued"}), Response({"value":"synthetic-oidc"}), + Response({**route,"status":"succeeded","output":'{"optimization_needed":true,"recommended_method":"grid_search"}'})] + initial_replies = replies + if defer_past_window: + quota_error = urllib.error.HTTPError("https://synthetic.invalid", 429, "synthetic quota", {}, + io.BytesIO(json.dumps({"status": "deferred", "retry_at": NOW.timestamp()+3600}).encode())) + initial_replies = replies[:2] + [quota_error] + sync, pull = Mock(), Mock() + with patch.object(job, "_read_policy", return_value=policy), patch.object(job, "_load_runtime", return_value=runtime), \ + patch.object(job, "STATE_ROOT", tmp_path / "state"), patch.object(job, "_console_bindings", return_value=(sync,pull)), \ + patch("quant_platform_kit.strategy_lifecycle.production_drift_health_probe.datetime", FixedClock), \ + patch("quant_platform_kit.strategy_lifecycle.research_promotion_cycle.datetime", FixedClock), \ + patch(runtime.client.__module__ + ".urllib.request.urlopen", side_effect=initial_replies) as http, \ + patch(runtime.client.__module__ + ".time.sleep"): + first = job.run_from_watcher(watcher(revision)) + if defer_past_window: + assert http.call_count == 3 + saved = list((tmp_path / "state" / "research_promotion_tickets").glob("*.json")) + assert len(saved) == 1 + assert json.loads(saved[0].read_text())["research_progress"]["stages"]["diagnose"]["status"] == "deferred" + later = datetime(2026, 9, 10, 2, tzinfo=timezone.utc) + + class AfterWindow(datetime): + @classmethod + def now(cls, tz=None): + return later.astimezone(tz) + + http.side_effect = replies + with patch.object(job, "_now", return_value=later), \ + patch("quant_platform_kit.strategy_lifecycle.production_drift_health_probe.datetime", AfterWindow), \ + patch("quant_platform_kit.strategy_lifecycle.research_promotion_cycle.datetime", AfterWindow): + second = job.run_from_watcher(watcher(revision)) + assert second["research_key"] == first["research_key"] + assert http.call_count == 3 + assert not list((tmp_path / "state").rglob("trials.json")) + stage = json.loads(saved[0].read_text())["research_progress"]["stages"]["diagnose"] + assert stage["status"] == "completed" + assert stage["result"]["optimization_needed"] is False + assert stage["result"]["reason"] == "forward_window_start_elapsed" + sync.assert_not_called() + return + if expired_window: + assert first["reason"] == "new_research_not_admitted" + assert http.call_count == 0 + assert not list((tmp_path / "state").rglob("trials.json")) + assert not list((tmp_path / "state" / "research_promotion_tickets").glob("*.json")) + sync.assert_not_called() + return + second = job.run_from_watcher(watcher(revision)) + assert first["status"] == second["status"] == "parked" # Flat synthetic data correctly rejects improvement. + assert first["research_key"] == second["research_key"] and second["resumed"] is True + assert http.call_count == 5 + raw = json.loads(drift_path.read_text()) + raw["drift_score"] = .81 + write(drift_path, raw) + third = job.run_from_watcher(watcher(revision)) + assert third["reason"] == "new_research_not_admitted" and http.call_count == 5 + tickets = list((tmp_path / "state" / "research_promotion_tickets").glob("*.json")) + assert len(tickets) == 1 + ticket = json.loads(tickets[0].read_text()) + assert ticket["notes"] == ["recommendation=reject"] + assert set(ticket["research_progress"]["stages"]) == {"diagnose", "optimize"} + trials = list((tmp_path / "state" / "experiments").rglob("trials.json")) + assert len(trials) == 1 and len(json.loads(trials[0].read_text())) == 13 + sync.assert_not_called() + + +@pytest.mark.parametrize("denied", [None, "repository", "workflow", "ref", "direct", "source", "cross_org", "static"]) +def test_sdk_to_actual_auth_and_execute_handler_preserves_aab_caller_cn_source(denied): + """Real claim/source gates; RSA/JWKS, quota and job execution stay offline.""" + from service import auth + from service import ai_gateway_service as gateway + from unittest.mock import Mock + runtime, drift = diagnosis_runtime() + claims = dict(aud="quant-codex-audit", iss=auth.GITHUB_OIDC_ISSUER, repository=job.BRIDGE_REPOSITORY, + workflow_ref=job.WORKFLOW_REF, ref="refs/heads/main", run_id="12345", exp=4102444800) + service_env = dict(CODEX_AUDIT_SERVICE_AUTH="github-oidc", + CODEX_AUDIT_SERVICE_ALLOWED_REPOSITORIES=job.BRIDGE_REPOSITORY, + CODEX_AUDIT_SERVICE_ALLOWED_DIRECT_REPOSITORIES=job.BRIDGE_REPOSITORY, + CODEX_AUDIT_SERVICE_ALLOWED_WORKFLOW_REFS=job.WORKFLOW_REF, + CODEX_AUDIT_SERVICE_ALLOWED_REFS="refs/heads/main", + CODEX_AUDIT_SERVICE_ALLOWED_SOURCE_REPOSITORIES=job.STRATEGY_REPOSITORY) + if denied == "repository": + claims["repository"] = job.STRATEGY_REPOSITORY + elif denied == "workflow": + claims["workflow_ref"] = job.WORKFLOW_REF.replace("strategy_optimization_watcher", "another") + elif denied == "ref": + claims["ref"] = "refs/heads/feature" + elif denied == "direct": + service_env["CODEX_AUDIT_SERVICE_ALLOWED_DIRECT_REPOSITORIES"] = "Another/direct" + elif denied == "source": + service_env["CODEX_AUDIT_SERVICE_ALLOWED_SOURCE_REPOSITORIES"] = job.BRIDGE_REPOSITORY + elif denied == "cross_org": + claims["repository"] = "Another/approved" + service_env["CODEX_AUDIT_SERVICE_ALLOWED_REPOSITORIES"] = claims["repository"] + service_env["CODEX_AUDIT_SERVICE_ALLOWED_DIRECT_REPOSITORIES"] = claims["repository"] + elif denied == "static": + service_env["CODEX_AUDIT_SERVICE_TOKEN"] = "synthetic-oidc" + route = dict(job_id="original-experiment", provider="codex", model="gpt-5.6-sol", + research_stage="optimization", reasoning_effort="high") + quota, submit = Mock(), Mock(return_value={**route, "status": "queued"}) + checked = [] + + def admit(_quota, repository, payload): + assert repository == job.STRATEGY_REPOSITORY + payload.update({key: route[key] for key in ("provider", "model", "research_stage", "reasoning_effort")}) + + def http(request, **_): + url = request if isinstance(request, str) else request.full_url + if "/oidc" in url: + return Response({"value": "synthetic-oidc"}) + if url.endswith("/healthz"): + return Response({"codex_research_routing": "v1"}) + if url.endswith("/execute/jobs"): + handler = object.__new__(gateway.AiGatewayRequestHandler) + handler.path, handler.headers = "/v1/ai/execute/jobs", dict(request.header_items()) + handler.headers["Content-Length"] = str(len(request.data)) + handler.rfile = io.BytesIO(request.data) + with patch.dict(os.environ, service_env), \ + patch.object(auth, "_jwt_parts", return_value=({"alg": "RS256", "kid": "test"}, dict(claims), b"x", b"y")), \ + patch.object(auth, "_load_jwks", return_value={"keys": [{"kid": "test"}]}), \ + patch.object(auth, "_verify_rs256"), patch.object(gateway, "_audit_log"), \ + patch.object(gateway, "_json_response") as response: + gateway.AiGatewayRequestHandler.do_POST(handler) + code, body = response.call_args.args[1:3] + checked.append(code) + if code != 202: + raise urllib.error.HTTPError(url, code, "synthetic-service-rejection", {}, io.BytesIO(json.dumps(body).encode())) + return Response(body) + assert url.endswith("/execute/jobs/original-experiment") + return Response({**route, "status": "succeeded", "output": '{"optimization_needed":false}'}) + + with patch("client.gateway_client.urllib.request.urlopen", side_effect=http), patch("client.gateway_client.time.sleep"), \ + patch.object(gateway, "get_quota_manager", return_value=quota), \ + patch.object(gateway, "_cleanup_expired_jobs"), patch.object(gateway, "_find_active_job_by_dedupe_key", return_value=None), \ + patch.object(gateway, "_active_job_count", return_value=0), patch.object(gateway, "_admit_codex_execute", side_effect=admit), \ + patch.object(gateway, "_submit_job", submit), patch.object(gateway, "get_health_monitor"), \ + patch.object(runtime.client, "analyze", side_effect=AssertionError("paid fallback")): + if denied: + with pytest.raises(ValueError, match="research_model_outcome_unavailable"): + job._diagnosis(runtime, drift, REVISION)() + else: + assert job._diagnosis(runtime, drift, REVISION)()["optimization_needed"] is False + if denied: + assert checked == [401] + submit.assert_not_called() + quota.record_execute.assert_not_called() + else: + assert checked == [202] + actual_claims, payload = submit.call_args.args + assert actual_claims["repository"] == job.BRIDGE_REPOSITORY + assert actual_claims["auth_method"] == "github_oidc" and actual_claims["workflow_ref"] == job.WORKFLOW_REF + assert payload["source_repository"] == job.STRATEGY_REPOSITORY and payload["source_ref"] == REVISION + quota.record_execute.assert_called_once_with(job.STRATEGY_REPOSITORY, provider="codex") + + +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_actual_console_adapter_reads_before_post_and_never_repeats_unknown_write(tmp_path, capsys): + importlib.import_module("quant_platform_kit") + from quant_platform_kit.strategy_lifecycle.research_promotion_cycle import ResearchPromotionTicket, ResearchPromotionState + secret = tmp_path / "synthetic.token" + secret.write_text("synthetic-test-only") + binding = dict(sync_url="https://synthetic.invalid/sync", pull_url="https://synthetic.invalid/pull", token_path=str(secret)) + ticket = ResearchPromotionTicket(ticket_id="rpt_"+"f"*64, strategy_profile=job.PROFILE, domain="cn_equity", + state=ResearchPromotionState.AWAITING_HUMAN, drift_status="critical", drift_score=.8, + created_at=NOW.isoformat(), updated_at=NOW.isoformat()) + methods = [] + + def http(request, **_): + methods.append(request.method) + assert request.get_header("Authorization") == "Bearer synthetic-test-only" + if request.method == "POST": + assert json.loads(request.data)["live_authority_granted"] is False + raise OSError("synthetic-private-transport-error") + raise urllib.error.HTTPError(request.full_url, 404, "synthetic absence", {}, None) + + with patch.object(job, "_protected_file") as protected, patch("urllib.request.urlopen", side_effect=http): + sync, pull = job._console_bindings(binding) + assert methods == [] + protected.assert_called_once_with(secret, secret=True) + assert sync(ticket) is False + assert sync(ticket) is False + assert methods == ["GET", "POST", "GET", "GET"] + assert capsys.readouterr().out == "" + + +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 + + +def test_required_ci_runs_the_complete_cn_slice_in_an_isolated_pinned_environment(): + text = (Path(__file__).resolve().parents[1] / ".github/workflows/ci.yml").read_text() + assert "python3 -m pytest tests ops/quant-monitor/tests -q --ignore=tests/test_run_cn_index_etf_research.py" in text + assert " cn-research:\n" not in text + research = text.split(" - name: Install the isolated CN research dependency set", 1)[1] + assert "python3 -m venv" in research and "--system-site-packages" not in research + assert research in text.split(" test:\n", 1)[1] + assert "cn-equity-strategies[research] @ git+https://github.com/QuantStrategyLab/CnEquityStrategies.git@2a0c5c9aafacfbe6519fb4029ca4ac18e4996a66" in research + assert '"${RUNNER_TEMP}/aab-cn-research/bin/python" -m pip check' in research + assert '"tests/test_run_cn_index_etf_research.py"' in research + assert "socket.socket.connect = blocked" in research + assert "quant-strategy-plugins" not in research + assert "continue-on-error" not in research and "if:" not in research + # Missing optional packages must fail this dedicated job, never skip it. + assert "importorskip" not in Path(__file__).read_text().split("def test_required_ci_runs_", 1)[0] + + +@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 From d366e08358a066454281c3aa08fb2ef00d7e9e7e Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Wed, 9 Sep 2026 04:57:35 +0800 Subject: [PATCH 2/2] fix: isolate CN integration checks from build metadata Co-Authored-By: Codex --- .github/workflows/ci.yml | 12 ++++++++++++ docs/cn-index-etf-research-dispatch-2026-09-09.md | 9 +++++++++ tests/test_run_cn_index_etf_research.py | 4 ++++ 3 files changed, 25 insertions(+) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 10acafaa..b0006f24 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -72,8 +72,17 @@ jobs: PYTEST_DISABLE_PLUGIN_AUTOLOAD: "1" PYTHONDONTWRITEBYTECODE: "1" run: | + set -euo pipefail + # Main-suite packaging leaves egg-info in the checkout. Test the same + # HEAD (including the PR merge) without those build-time metadata files. + cn_test_source="$(mktemp -d "${RUNNER_TEMP}/aab-cn-research-source.XXXXXX")" + git archive HEAD | tar -x -C "${cn_test_source}" + cd "${cn_test_source}" "${RUNNER_TEMP}/aab-cn-research/bin/python" - <<'PY' + import importlib.metadata + from pathlib import Path import socket + import sys import pytest def blocked(*args, **kwargs): @@ -81,6 +90,9 @@ jobs: socket.socket.connect = blocked socket.create_connection = blocked + for package in ("cn-equity-strategies", "quant-platform-kit", "ai-gateway-client"): + installed = importlib.metadata.distribution(package) + assert Path(installed.locate_file("")).resolve().is_relative_to(Path(sys.prefix).resolve()) raise SystemExit(pytest.main([ "-q", "-p", "no:cacheprovider", "--tb=short", "tests/test_run_cn_index_etf_research.py", diff --git a/docs/cn-index-etf-research-dispatch-2026-09-09.md b/docs/cn-index-etf-research-dispatch-2026-09-09.md index be3d9586..8ea9b869 100644 --- a/docs/cn-index-etf-research-dispatch-2026-09-09.md +++ b/docs/cn-index-etf-research-dispatch-2026-09-09.md @@ -68,3 +68,12 @@ forward policy、真实 calendar 摘要、固定来源摘要和 baseline 配置 独立 reviewer 已完成全部源码及增量审查，无未关闭 P1/P2：原 97 项、时间边界定向验证及最后实际 caller/额度延期/required CI 的 5 项均通过；包含首 session 09:24:59 允许、等于 09:25 和收盘后冻结拒绝。评审不等于真实数据或线上模型验收，发布/部署仍由主任务统一执行。 + + +### PR #169：主安装遗留元数据与隔离 CI（2026-09-09） + +远端 CI `34276709282` 的 required `test` 在 CN 专项出现 3 failed / 74 passed；安装与 `pip check` 已通过。按受测 merge HEAD `5467e9b32b8d698e895f2cbe72bf70bd9414c1ec` 在临时源码副本先实际安装本仓，生成 `ai_gateway_client.egg-info`，再用固定 CN 隔离解释器运行，三个实际 CN caller 用例同样 RED。SDK 模块真实 import 来自隔离 site-packages，但 distribution 查找先读 cwd 的生成元数据，其无 VCS `direct_url.json`，严格安装版本门正确拒绝。前次干净源码验证未重演主安装顺序，不能替代此 CI 结果。 + +修复仅在原 required `test` 的专项步骤：用 `git archive HEAD`（保留 PR merge 的受测版本）导出唯一临时源码目录并进入该目录后运行；原主 checkout 的安装文件完整保留。专项还检查三个 distribution 路径确在隔离解释器目录内，生产 `_installed_revision` 和全部 pin、权限、时间/费用门不变。没有 skip、放松版本门或移出 required check。 + +同一保留构建残留的临时环境，执行更新后的 CI 步骤实际 **77 passed**；归档入口逐字节等于受测 HEAD，未含 egg-info，原生成元数据字节不变。永久 CI 结构回归先 RED 后 **1 passed**，actionlint、Ruff、diffcheck 通过。RED/GREEN 日志分别为 `/tmp/aab-cn-pr169-ci-red-20260909.log`、`/tmp/aab-cn-pr169-ci-green-20260909.log`；完整远端原日志 `/tmp/aab-cn-pr169-ci-full-20260909.log`。这是本地修复验证，后续 PR CI 状态仍以实际新 run 为准。 diff --git a/tests/test_run_cn_index_etf_research.py b/tests/test_run_cn_index_etf_research.py index a2ac1f60..5d123e4c 100644 --- a/tests/test_run_cn_index_etf_research.py +++ b/tests/test_run_cn_index_etf_research.py @@ -673,6 +673,10 @@ def test_required_ci_runs_the_complete_cn_slice_in_an_isolated_pinned_environmen assert "cn-equity-strategies[research] @ git+https://github.com/QuantStrategyLab/CnEquityStrategies.git@2a0c5c9aafacfbe6519fb4029ca4ac18e4996a66" in research assert '"${RUNNER_TEMP}/aab-cn-research/bin/python" -m pip check' in research assert '"tests/test_run_cn_index_etf_research.py"' in research + assert 'git archive HEAD | tar -x -C "${cn_test_source}"' in research + assert 'cd "${cn_test_source}"' in research + assert research.index("git archive HEAD") < research.index('cd "${cn_test_source}"') < research.index("import socket") + assert 'is_relative_to(Path(sys.prefix).resolve())' in research assert "socket.socket.connect = blocked" in research assert "quant-strategy-plugins" not in research assert "continue-on-error" not in research and "if:" not in research